Agno Agent 输入输出 Cookbook 套件：9 个 I/O 实战示例的测试验证与逐项实现解析

本篇基于 agno 仓库 cookbook/02_agents/02_input_output/ 目录下的测试日志（TEST_LOG.md），梳理该目录下 9 个 Agent 输入/输出示例的验证结果、运行环境与耗时基线，并逐一还原每个示例的代码实现与对应的 Agent 参数，帮助读者快速掌握 agno 中 expected_output、input_schema、output_schema、parser_model、output_model、流式响应、结果落盘等输入输出机制的完整用法。

一、测试套件概览：环境、基线与验证结果

TEST_LOG.md 记录了该套件的测试元信息：

• 测试日期：2026-02-13
• 运行环境：.venvs/demo/bin/python，pgvector 服务处于运行状态
• 结果：9 个示例全部 PASS，均标记为 untagged 层级（未指定优先级标签）

各示例的验证结果与实测耗时如下（直接取自测试日志）：

示例文件	状态	层级	说明	实测耗时
expected_output.py	PASS	untagged	使用 expected_output 引导回复格式	4s
input_formats.py	PASS	untagged	演示多种输入格式	2s
input_schema.py	PASS	untagged	演示输入 schema 校验	101s
output_model.py	PASS	untagged	使用独立 output model 精修输出	49s
output_schema.py	PASS	untagged	演示结构化输出 schema	18s
parser_model.py	PASS	untagged	演示 parser model 结构化抽取	46s
response_as_variable.py	PASS	untagged	将 Agent 响应捕获为变量	12s
save_to_file.py	PASS	untagged	自动保存响应到文件	10s
streaming.py	PASS	untagged	逐 token 流式输出响应	9s

从耗时分布可以直观看出各示例对模型调用链路的差异：input_formats.py（2s）与 expected_output.py（4s）只走单次生成；而 input_schema.py（101s）涉及 HackerNews 工具调用的多轮推理，output_model.py（49s）与 parser_model.py（46s）则包含“主模型 + 第二模型”的双重调用，耗时自然更长。

运行前置条件与运行方式

根据同目录的 README.md，运行该套件需要：

1. 使用 direnv allow 加载环境变量（其中包含 OPENAI_API_KEY）；
2. 执行 ./scripts/demo_setup.sh 创建演示环境，然后用 .venvs/demo/bin/python 运行各 cookbook 脚本；
3. 部分示例需要可选的本地服务（如 pgvector）或特定服务商的 API key。

单文件运行命令为：

.venvs/demo/bin/python cookbook/02_agents/02_input_output/<file>.py

二、expected_output：用“目标提示”约束回复形态

expected_output.py 演示了 expected_output 参数——它给 Agent 提供一个明确的“回复应该长什么样”的目标：

from agno.agent import Agent
from agno.models.openai import OpenAIResponses

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    # expected_output gives the agent a clear target for what the response should look like
    expected_output="A numbered list of exactly 5 items, each with a title and one-sentence description.",
    markdown=True,
)

if __name__ == "__main__":
    agent.print_response(
        "What are the most important principles of clean code?",
        stream=True,
    )

expected_output 与 instructions 的区别在于：前者不是行为规则，而是对回复成品形态的规格说明（如“恰好 5 条、每条含标题和一句话描述”）。在 Agent 源码 中，该字段声明为 expected_output: Optional[str] = None，默认不启用。实测中该示例 4 秒内完成一次流式生成，验证了参数注入不会改变运行链路，仅影响最终回复结构。

三、input_formats：结构化消息与多模态输入

[ input_formats.py ] 演示了向 Agent 传递结构化消息字典而非纯字符串的能力，此处是一个带图片 URL 的多模态输入：

from agno.agent import Agent

agent = Agent()

if __name__ == "__main__":
    agent.print_response(
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What's in this image?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg",
                    },
                },
            ],
        },
        stream=True,
        markdown=True,
    )

关键点在于 print_response 的入参既可以是字符串，也可以是符合消息格式（role + 多段 content 列表）的字典，从而在程序侧直接构造图文混合输入。该示例是全套件中耗时最短的（2s），说明结构化输入在解析层开销极低。

四、input_schema：用 Pydantic 模型做输入校验

[ input_schema.py ] 展示了如何用 Pydantic 模型定义结构化输入契约，并让 Agent 接收字典或模型实例两种形式：

from typing import List
from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.tools.hackernews import HackerNewsTools
from pydantic import BaseModel, Field


class ResearchTopic(BaseModel):
    """Structured research topic with specific requirements"""
    topic: str
    focus_areas: List[str] = Field(description="Specific areas to focus on")
    target_audience: str = Field(description="Who this research is for")
    sources_required: int = Field(description="Number of sources needed", default=5)


hackernews_agent = Agent(
    name="Hackernews Agent",
    model=OpenAIResponses(id="gpt-5-mini"),
    tools=[HackerNewsTools()],
    role="Extract key insights and content from Hackernews posts",
    input_schema=ResearchTopic,
)

if __name__ == "__main__":
    # 方式一：传入符合 schema 的字典
    hackernews_agent.print_response(
        input={
            "topic": "AI",
            "focus_areas": ["AI", "Machine Learning"],
            "target_audience": "Developers",
            "sources_required": "5",
        }
    )
    # 方式二：直接传入 Pydantic 模型实例
    hackernews_agent.print_response(
        input=ResearchTopic(
            topic="AI",
            focus_areas=["AI", "Machine Learning"],
            target_audience="Developers",
            sources_required=5,
        )
    )

值得注意的细节：字典方式下 sources_required 传的是字符串 "5"，而模型字段类型是 int——测试日志记录该示例 101s 内 PASS，说明 agno 对字典输入做了兼容处理（Pydantic 的宽松类型转换）。在源码中 input_schema 字段 声明为 input_schema: Optional[Type[BaseModel]] = None，即只接受 Pydantic 模型类，不支持裸字典定义。

该示例耗时最长（101s），原因是配置了 HackerNewsTools 后 Agent 会进行真实工具检索，属于多轮推理而非单次生成。

五、output_model：双模型协作精修最终回复

[ output_model.py ] 演示了 agno 的一个特色机制：用独立的 output model 替换主模型输出。

from agno.agent import Agent, RunOutput
from agno.models.openai import OpenAIResponses
from rich.pretty import pprint

agent = Agent(
    model=OpenAIResponses(id="gpt-5-mini"),
    description="You are a helpful chef that provides detailed recipe information.",
    output_model=OpenAIResponses(id="gpt-5.2"),
    output_model_prompt="You are a world-class culinary writer. Rewrite the recipe with vivid descriptions, pro tips, and elegant formatting.",
)

if __name__ == "__main__":
    run: RunOutput = agent.run("Give me a recipe for pad thai.")
    pprint(run.content)

文件头注释说明了其适用场景：output_model 接收同样的对话并生成自己的回复，替换主模型的输出。典型用法是让便宜模型（如 gpt-5-mini）负责推理与工具调用，再交给更强的模型（如 gpt-5.2）产出精修的最终答案，从而在成本与质量之间取得平衡。注释同时提醒：结构化 JSON 输出应改用 parser_model（见下一节）。对应源码字段为 output_model，默认 None。

六、output_schema 与 parser_model：结构化输出的两种路径

output_schema：模型直接产出结构化数据

[ output_schema.py ] 用 output_schema 让 Agent 的 run.content 直接成为符合 Pydantic 模型的结构化对象：

class BreakingNewsSummary(BaseModel):
    topic: str = Field(..., description="The topic or region being summarized")
    summary: str = Field(..., description="A concise summary of the latest developments")
    key_updates: List[str] = Field(..., description="Important updates or headlines related to the topic")
    overall_sentiment: str = Field(..., description="Overall tone of the news coverage, such as positive or mixed")

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    description="You summarize current events into clean structured outputs.",
    output_schema=BreakingNewsSummary,
)

if __name__ == "__main__":
    run: RunOutput = agent.run("Latest news from France?")
    pprint(run.content)

源码中 output_schema 的类型为 Optional[Union[Type[BaseModel], Dict[str, Any]]]，即同时支持 Pydantic 模型类和 JSON Schema 字典两种写法。

parser_model：用第二模型做结构化抽取

[ parser_model.py ] 则展示了 output_schema + parser_model 的组合：主模型正常推理（甚至可以使用工具），由单独的 parser 模型负责把结果解析为目标结构。示例定义了一个包含 11 个字段的 NationalParkAdventure 模型（园区名、最佳季节、招牌景点、推荐步道、野生动物、摄影点、露营选项、安全提示、隐藏亮点、难度评级 ge=1, le=5、建议天数 ge=1, le=14、特殊许可），并利用 Field(ge=..., le=...) 约束数值范围：

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    description="You help people plan amazing national park adventures and provide detailed park guides.",
    output_schema=NationalParkAdventure,
    parser_model=OpenAIResponses(id="gpt-5.2"),
)

if __name__ == "__main__":
    run: RunOutput = agent.run(national_parks[random.randint(0, len(national_parks) - 1)])
    pprint(run.content)

两个示例的分工（以源码结构看）：output_schema 让主模型直接按结构作答；parser_model（源码字段）则把“结构化抽取”这一职责交给一个独立的模型调用，适合主模型输出不可控或主模型本身弱于格式遵循的场景。两者在测试日志中分别以 18s 与 46s 完成，均验证了端到端链路。

七、response_as_variable 与 save_to_file：响应的捕获与持久化

把响应捕获为变量

[ response_as_variable.py ] 演示了用 agent.run() 而非 print_response() 将完整 RunOutput 对象捕获到变量中，便于后续程序化处理：

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[YFinanceTools()],
    instructions=["Use tables where possible"],
    markdown=True,
)

if __name__ == "__main__":
    run_response: RunOutput = agent.run("What is the stock price of NVDA")
    pprint(run_response)

    # run_response_strem: Iterator[RunOutputEvent] = agent.run("...", stream=True)
    # for response in run_response_strem:
    #     pprint(response)

代码中注释保留了流式变体写法：agent.run(..., stream=True) 返回 Iterator[RunOutputEvent]，可迭代处理每个流式事件。

自动落盘

[ save_to_file.py ] 展示了 save_response_to_file 参数——Agent 每次运行后自动把响应写入指定文件：

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    save_response_to_file="tmp/agent_output.md",
    markdown=True,
)

if __name__ == "__main__":
    os.makedirs("tmp", exist_ok=True)
    agent.print_response("Write a brief guide on Python virtual environments.", stream=True)
    print(f"\nResponse saved to: {agent.save_response_to_file}")

注意示例中显式 os.makedirs("tmp", exist_ok=True) 创建父目录，说明落盘目标目录需要自行保证存在；源码中该字段 声明为 save_response_to_file: Optional[str] = None，默认不落盘。

八、streaming：逐 token 流式输出

[ streaming.py ] 是流式用法的最小示例：

from agno.agent import Agent
from agno.models.openai import OpenAIResponses

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    markdown=True,
)

if __name__ == "__main__":
    # Stream the response token by token
    agent.print_response(
        "Explain the difference between concurrency and parallelism.",
        stream=True,
    )

核心就是一个参数：print_response(..., stream=True)。print_response 内部完成富文本渲染与逐 token 打印，适合交互式演示；生产集成中则如第七节所述改用 run(..., stream=True) 手动消费事件流。该示例 9s 完成，与测试日志记录一致。

九、源码层参数对照与延伸说明

将上述 9 个示例涉及的 Agent 字段汇总，全部可在 libs/agno/agno/agent/agent.py 的字段声明中找到一一对应：

参数	源码声明行	对应示例	作用
expected_output	L252	expected_output.py	以自然语言描述回复成品形态
input_schema	L300	input_schema.py	限定输入必须为符合该 Pydantic 模型的数据
output_schema	L303	output_schema.py / parser_model.py	限定输出为结构化数据（模型类或 JSON Schema 字典）
parser_model	L305	parser_model.py	指定独立的解析模型完成结构化抽取
output_model	L309	output_model.py	用第二模型替换主模型输出做精修
save_response_to_file	L320	save_to_file.py	响应自动写入指定文件

补充两点边界说明：

1. 测试日志覆盖范围：TEST_LOG.md 记录了 9 个示例的验证结果；目录下还存在 followup_suggestions.py 与 followup_suggestions_streaming.py 两个未纳入该日志的示例。前者演示了 followups=True 开关与 num_followups（默认 3，源码校验其必须 ≥ 1，见 agent.py L648-L650）等参数，可作为本套件的延伸阅读。
2. 耗时数据的使用前提：表中耗时来自 2026-02-13 在 .venvs/demo/bin/python + pgvector 环境的单次实测，依赖具体模型与网络状态，只能作为链路复杂度的相对参考，不应视为性能承诺。

十、小结

• cookbook/02_agents/02_input_output/ 套件 9 个示例在 2026-02-13 的测试中全部 PASS，构成 agno Agent 输入输出能力的完整验证基线；
• 输入侧掌握 expected_output（形态约束）、消息字典输入（多模态）与 input_schema（Pydantic 校验）三种手段；
• 输出侧掌握 output_schema（直接结构化）、parser_model（独立解析模型）、output_model（双模型精修）三条路径，可按成本与可控性需求选型；
• 运行侧通过 run()/print_response() 捕获变量、事件流或经 save_response_to_file 自动落盘。

按 README.md 的前置步骤配置好 direnv 与 demo 环境后，即可用 .venvs/demo/bin/python cookbook/02_agents/02_input_output/<file>.py 逐一复现本文覆盖的全部行为。