Workflow Metrics

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

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",
)

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",
)

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),
    ],
)

response = workflow.run(input="AI trends in 2024")

# Workflow metrics
if response.metrics:
    pprint(response.metrics.to_dict())

# Step metrics
if response.metrics:
    for step_name, step_metrics in response.metrics.steps.items():
        print(f"{step_name}: {step_metrics.executor_name} ({step_metrics.executor_type})")
        if step_metrics.metrics:
            print(f"  Duration: {step_metrics.metrics.duration:.2f}s, Tokens: {step_metrics.metrics.total_tokens}")

# Session metrics
pprint(workflow.get_session_metrics().to_dict())