Basic Workflow Tracing

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.tools.hackernews import HackerNewsTools
from agno.tracing import setup_tracing
from agno.workflow.condition import Condition
from agno.workflow.step import Step
from agno.workflow.types import StepInput
from agno.workflow.workflow import Workflow

# Set up database for traces
db = SqliteDb(db_file="tmp/traces.db")

# Enable tracing (call once at startup)
setup_tracing(db=db)

# === BASIC AGENTS ===
researcher = Agent(
    name="Researcher",
    instructions="Research the given topic and provide detailed findings.",
    tools=[HackerNewsTools()],
)

summarizer = Agent(
    name="Summarizer",
    instructions="Create a clear summary of the research findings.",
)

fact_checker = Agent(
    name="Fact Checker",
    instructions="Verify facts and check for accuracy in the research.",
    tools=[HackerNewsTools()],
)

writer = Agent(
    name="Writer",
    instructions="Write a comprehensive article based on all available research and verification.",
)

# === CONDITION EVALUATOR ===


def needs_fact_checking(step_input: StepInput) -> bool:
    """Determine if the research contains claims that need fact-checking"""
    return True


# === WORKFLOW STEPS ===
research_step = Step(
    name="research",
    description="Research the topic",
    agent=researcher,
)

summarize_step = Step(
    name="summarize",
    description="Summarize research findings",
    agent=summarizer,
)

# Conditional fact-checking step
fact_check_step = Step(
    name="fact_check",
    description="Verify facts and claims",
    agent=fact_checker,
)

write_article = Step(
    name="write_article",
    description="Write final article",
    agent=writer,
)

# === BASIC LINEAR WORKFLOW ===
workflow = Workflow(
    name="Basic Linear Workflow",
    description="Research -> Summarize -> Condition(Fact Check) -> Write Article",
    db=db,
    steps=[
        research_step,
        summarize_step,
        Condition(
            name="fact_check_condition",
            description="Check if fact-checking is needed",
            evaluator=needs_fact_checking,
            steps=[fact_check_step],
        ),
        write_article,
    ],
)

# Run the workflow - traces are captured automatically
workflow.print_response("Write an article on AI agents?")

# Query traces from the database
traces, count = db.get_traces(workflow_id=workflow.id, limit=10)
print(f"\nFound {count} traces for workflow '{workflow.name}'")
for trace in traces:
    print(f"  - {trace.name}: {trace.duration_ms}ms ({trace.status})")

uv pip install -U openai agno sqlalchemy opentelemetry-api opentelemetry-sdk openinference-instrumentation-agno