Berack/upo-app-agents
0
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Demo: aggiunto tools

Browse Source
This commit is contained in:
Giacomo Bertolazzi
2025-10-14 17:54:02 +02:00
parent 2fc71fbc55
commit 89a16459f9
1 changed files with 14 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
demos/agno_workflow.py
View File

@@ -8,11 +8,17 @@ from agno.workflow.types import StepOutput, StepInput
from agno.workflow.parallel import Parallel from agno.workflow.parallel import Parallel
from agno.workflow.workflow import Workflow from agno.workflow.workflow import Workflow
def my_sum(a: int, b: int) -> int:
return a + b
def my_mul(a: int, b: int) -> int:
return a * b
def build_agent(instructions: str) -> Agent: def build_agent(instructions: str) -> Agent:
return Agent( return Agent(
instructions=instructions, instructions=instructions,
model=Ollama(id='qwen3:1.7b') model=Ollama(id='qwen3:1.7b'),
tools=[my_sum]
) )
def remove_think(text: str) -> str: def remove_think(text: str) -> str:
@@ -31,12 +37,12 @@ def combine_steps_output(inputs: StepInput) -> StepOutput:
return StepOutput(content=content) return StepOutput(content=content)
async def main(): async def main():
query = "Come posso fare per dormire meglio?" query = "Quanto fa 50 + 150 * 50?"
s1 = Step(name="Translate", agent=build_agent(instructions="Transform in English the user query. Respond only with the summarized query in English.")) s1 = Step(name="Translate", agent=build_agent(instructions="Transform in English the user query. DO NOT answer the question and output ONLY the translated question."))
s2 = Step(name="Predict", agent=build_agent(instructions="You will be given a question in English. Provide a summarized answer in a concise manner. Ouput ONLY the answer.")) s2 = Step(name="Predict", agent=build_agent(instructions="You will be given a question in English. You can use the tools at your disposal. Answer the question and output ONLY the answer."))
step_a = Step(name="Lang", agent=build_agent(instructions="Detect the language and output ONLY the language code. Es: 'en' for English, 'it' for Italian, 'ja' for Japanese.")) step_a = Step(name="Lang", agent=build_agent(instructions="Detect the language from the question and output ONLY the language code. Es: 'en' for English, 'it' for Italian, 'ja' for Japanese."))
step_b = Steps(name="Answer", steps=[s1, s2]) step_b = Steps(name="Answer", steps=[s1, s2])
step_c = Step(name="Combine", executor=combine_steps_output) step_c = Step(name="Combine", executor=combine_steps_output)
step_f = Step(name="Final", agent=build_agent(instructions="Translate the phrase in the language code provided. Respond only with the translated answer.")) step_f = Step(name="Final", agent=build_agent(instructions="Translate the phrase in the language code provided. Respond only with the translated answer."))
@@ -49,10 +55,11 @@ async def main():
result = "" result = ""
async for event in await wf.arun(query, stream=True, stream_intermediate_steps=True): async for event in await wf.arun(query, stream=True, stream_intermediate_steps=True):
content = event.content if hasattr(event, 'content') and type(event.content) == str else "" # type: ignore content = getattr(event, 'content', '')
step_name = getattr(event, 'step_name', '')
if event.event in [WorkflowRunEvent.step_completed]: if event.event in [WorkflowRunEvent.step_completed]:
print(f"{str(event.event)} --- {event.step_name} --- {remove_think(content).replace('\n', '\\n')[:80]}") # type: ignore print(f"{str(event.event)} --- {step_name} --- {remove_think(content).replace('\n', '\\n')[:80]}")
if event.event in [WorkflowRunEvent.workflow_completed]: if event.event in [WorkflowRunEvent.workflow_completed]:
result = remove_think(content) result = remove_think(content)
print(f"\nFinal result: {result}") print(f"\nFinal result: {result}")