Chapter 3 · Running Agents and Parsing RunOutput

Learning objectives: Master the differences between three run methods — run()/arun()/print_response() — learn to extract content, messages, and metrics from RunOutput, and understand the correct approach to multi-turn conversations.

3.1 Three Ways to Run

The official Agno docs are clear: use print_response() for development and debugging; use run() or arun() for production code.

# run_basics.py
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

load_dotenv()

agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
)

# Method 1: Print directly during development — human-readable but no return value
agent.print_response("Explain dependency injection in one sentence")

# Method 2: Synchronous run — returns a RunOutput object; use this in production code
response = agent.run("Explain closures in one sentence")
print(response.content)      # The actual answer lives here

# Method 3: Asynchronous run (coroutine) — ideal for high-concurrency servers
import asyncio

async def main():
    resp = await agent.arun("Explain decorators in one sentence")
    print(resp.content)

asyncio.run(main())

print_response internally calls run() — it just also formats and prints the output. You must use run()/arun() when writing unit tests or web APIs because you need to assert or transform the return value.

3.2 RunOutput Core Fields

The return value of run() is a RunOutput object. Core attributes per the official docs:

Field	Meaning
content	Final answer text (string; Pydantic object for structured output)
messages	Complete message list sent to the model this run (includes system/tool messages)
metrics	Token usage, duration, and other metrics
run_id / session_id	Unique ID for this run and the session it belongs to
agent_id / agent_name	Agent identifier that handled this request
reasoning_content	Reasoning process text for thinking models

# inspect_output.py —— Dissect a RunOutput
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
))

r = agent.run("What is idempotency?")

print("Answer:", r.content[:80], "...")
print("run_id:", r.run_id)
print("Message count:", len(r.messages))
for m in r.messages:
    print("-", m.role, ":", str(m.content)[:50])
print("Input tokens:", r.metrics.input_tokens, "/ Output tokens:", r.metrics.output_tokens)

What metrics is good for

metrics is the foundation for cost accounting: log token counts per run and you can aggregate spend by user or feature at month-end. Chapter 24 covers observability in depth.

3.3 Multi-Turn Conversations: Why the Second Run "Forgets"

Here's a trap every beginner steps into:

# memory_trap.py —— Demonstrates default no-memory behavior
agent = Agent(model=make_model())   # make_model per section 2.3

agent.run("My name is Xiaogao and my favorite number is 7")
resp = agent.run("What's my name?")   # ❌ Model won't know!
print(resp.content)

There is no shared state between two run() calls — each invocation is a completely fresh context, and the model has no idea what was said previously. Three solutions exist, in order of increasing complexity:

1. Manual history stitching: prepend previous Q&A to the next input yourself;
2. Chat history: let Agno carry the same session's historical runs automatically (requires storage; Chapters 14–15);
3. Memory: cross-session user-level memory (Chapter 16).

This section uses the simplest manual approach to nail down the concept:

# manual_history.py —— Manually maintaining conversation history
history = []

def chat(question: str) -> str:
    prompt = ""
    for q, a in history:                       # Stitch history into the prompt
        prompt += f"User: {q}\nAssistant: {a}\n"
    prompt += f"User: {question}"
    resp = agent.run(prompt)
    history.append((question, resp.content))   # Record this round
    return resp.content

chat("My favorite color is blue")
print(chat("What color do I like?"))                 # ✅ Answers "blue"

The problems are obvious: the prompt grows longer, token cost climbs linearly, and there's no persistence. This approach is only for understanding the principle — use the session mechanism from Chapter 14 in production.

3.4 Concurrent Runs with Asyncio

The value of arun only shows in concurrent scenarios: three independent tasks done serially requires three network round-trips; concurrent execution needs only the time of the slowest one:

# concurrent_runs.py —— asyncio.gather for three concurrent requests
import asyncio
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1"),
)

async def main():
    questions = ["Difference between threads and processes", "Difference between TCP and UDP", "Difference between GET and POST"]
    # Fire all three requests concurrently; gather collects every result
    results = await asyncio.gather(*(agent.arun(q) for q in questions))
    for q, r in zip(questions, results):
        print(f"[{q}] -> {r.content[:40]}...")

asyncio.run(main())

Note the generator expression inside asyncio.gather: one coroutine per question. If any request fails, the whole gather raises — use return_exceptions=True in production to handle failures individually.

3.5 Chapter Summary

• Use print_response for development, run for business code, arun for concurrency;
• Key RunOutput fields: content for the answer, messages to inspect context, metrics for cost;
• By default each run() is independent — the model has no cross-call memory;
• Manual history stitching works for learning but is too expensive for production; use session storage + chat history (Chapters 14–15).

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. In a production FastAPI web service handling Agent requests, which method is recommended?

2. After calling agent.run() twice consecutively, the second question references the first but the model cannot answer. Why?

3. To check how many tokens a single run consumed, which RunOutput field should you inspect?

4. Regarding the messages field, which statement is correct?

🛠️ Hands-On Exercises

1. Write a script that asks the same question via both run and arun; print both run_id and metrics to confirm they are independent runs.
2. Extend 3.3's manual_history.py: cap history at 3 turns, dropping the oldest when exceeded; observe how answer quality changes.
3. Use asyncio.gather to ask 5 unrelated questions concurrently; record total elapsed time, then switch to serial execution and compare.

With synchronous running mastered, proceed to Chapter 4: Streaming Output and Real-Time Responses.