from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.team import Team
from agno.tools.hackernews import HackerNewsTools
from agno.tools.yfinance import YFinanceTools
from agno.workflow import Step, Workflow
from rich.pretty import pprint
# Define agents
hackernews_agent = Agent(
name="HackerNews Agent",
model=OpenAIResponses(id="gpt-5.2"),
tools=[HackerNewsTools()],
role="Extract key insights from HackerNews posts",
)
finance_agent = Agent(
name="Finance Agent",
model=OpenAIResponses(id="gpt-5.2"),
tools=[YFinanceTools()],
role="Get stock prices and financial data",
)
# Define research team
research_team = Team(
name="Research Team",
members=[hackernews_agent, finance_agent],
instructions="Research tech topics from HackerNews and financial data",
)
content_planner = Agent(
name="Content Planner",
model=OpenAIResponses(id="gpt-5.2"),
instructions="Plan a content schedule based on research",
)
# Create workflow
workflow = Workflow(
name="Content Creation Workflow",
db=SqliteDb(db_file="tmp/workflow.db"),
steps=[
Step(name="Research Step", team=research_team),
Step(name="Content Planning Step", agent=content_planner),
],
)
# Run workflow
response = workflow.run(input="AI trends in 2024")
# Print workflow-level metrics
print("Workflow Metrics")
if response.metrics:
pprint(response.metrics.to_dict())
# Print workflow duration
if response.metrics and response.metrics.duration:
print(f"\nTotal execution time: {response.metrics.duration:.2f} seconds")
# Print step-level metrics
print("Step Metrics")
if response.metrics:
for step_name, step_metrics in response.metrics.steps.items():
print(f"\nStep: {step_name}")
print(f"Executor: {step_metrics.executor_name} ({step_metrics.executor_type})")
if step_metrics.metrics:
print(f"Duration: {step_metrics.metrics.duration:.2f}s")
print(f"Tokens: {step_metrics.metrics.total_tokens}")
# Print session metrics
print("Session Metrics")
pprint(workflow.get_session_metrics().to_dict())