Agno Agent 输入输出实战指南：input_schema、output_schema、流式响应与 Followups 全解析

本文基于 Agno 官方 cookbook 中的输入输出示例目录，系统讲解 Agent 的输入格式（多模态消息、input_schema 校验）、输出控制（expected_output、output_schema、output_model、parser_model）、流式响应、把响应作为变量捕获、自动落盘以及内置 followups 建议的用法。读完本文，你可以直接复制 cookbook 中的完整示例运行，并理解这些参数在 Agno 源码中的落地位置与行为细节。

目录说明与运行前提

cookbook/02_agents/02_input_output/README.md 是本文的主体文档，它给出了一组围绕 Agent 输入/输出的示例清单：

示例文件	用途
expected_output.py	用 expected_output 参数为 Agent 回复提供格式提示
input_formats.py	演示多种输入格式（含多模态消息）
input_schema.py	演示 input_schema 输入校验
output_model.py	用独立的 output_model 精炼主模型回复
output_schema.py	用 output_schema 返回结构化数据
parser_model.py	演示 parser_model 结构化抽取
response_as_variable.py	把 Agent 响应捕获为变量
save_to_file.py	自动把 Agent 响应保存到文件
streaming.py	逐 token 流式输出 Agent 响应
followup_suggestions.py	获取带 AI 生成后续问题建议的响应

运行前提（与 README 一致）：

• 用 direnv allow 加载环境变量（包括 OPENAI_API_KEY）；
• 先用仓库根目录的 ./scripts/demo_setup.sh 创建 demo 环境，再用 .venvs/demo/bin/python 运行 cookbook 脚本；
• 部分示例需要可选的本地服务（如 pgvector）或特定服务商的 API key。

运行方式统一为：

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

expected_output：给回复定一个“靶子”

expected_output.py 展示的最轻量级输出控制手段是在构造 Agent 时声明期望的回复形态：

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

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    # expected_output 告诉 Agent 回复应该长什么样
    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 是 Agent 的一个可选字符串属性，定义为 expected_output: Optional[str] = None（见 agent.py）。从源码结构看，它作为一种提示（hint）参与系统指令组装，用来引导模型输出格式，而不是像 output_schema 那样强制解析成 Pydantic 对象——两者的定位不同：前者是“软约束”，后者是“硬约束”。

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

input_formats.py 演示了向 Agent 传入消息级（message-level）的输入，而不只是普通字符串。示例直接构造了一个带 role 和 content 列表的字典，其中 content 同时包含文本块和图片块：

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

从源码看，run/print_response 的 input 参数类型为 Optional[Union[str, List, Dict, Message]]（见 _run.py），即支持字符串、消息列表、字典和 Message 对象四种形态；字典形态按 OpenAI 风格的消息体解析，其中 type: "image_url" 的内容块用于传递图片，从而实现一次调用内文本 + 图像的多模态输入。

input_schema：用 Pydantic 模型约束输入

input_schema.py 演示了“先结构化、再执行”的输入模式。先定义一个 Pydantic 模型描述输入契约，再把它挂到 Agent 上：

from typing import List
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,
)

运行时有两种等价的传参方式：

# 方式一：传与 input_schema 匹配的 dict
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,
    )
)

在源码中，input_schema 定义为 Optional[Type[BaseModel]]（见 agent.py），即只能接受 Pydantic BaseModel 的子类类型。从示例可以看出 Pydantic 的 Field(default=...) 会生效（sources_required 有默认值 5），因此调用方可以省略带默认值的字段。这个模式适合需要“强输入契约”的场景：上游程序（而非用户）以结构化参数驱动 Agent，字段类型不匹配时会在进入模型调用前就被拦截。

output_schema 与 parser_model：结构化输出的两条路径

output_schema：让 run.content 直接是 Pydantic 对象

output_schema.py 展示了最基础的结构化输出：把 Pydantic 模型作为 output_schema 传给 Agent，run.content 返回的就是该模型的实例：

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)  # BreakingNewsSummary 实例

Field 里的 description 会进入传给模型的 JSON Schema，是指导模型正确填值的关键，建议为每个字段都写上清晰的描述。

parser_model：指定谁来“解析”结构化结果

parser_model.py 在 output_schema 的基础上增加了 parser_model——一个专门负责把模型原始输出解析/校验成目标 Pydantic 结构的模型。示例定义了一个字段较多、带取值约束的 NationalParkAdventure 模型（如 difficulty_rating: int = Field(..., ge=1, le=5)、estimated_days: int = Field(..., ge=1, le=14)，以及带默认值 default=[] 的 special_permits_needed），然后：

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)

在源码中可以看到 parser_model、parser_model_prompt、output_model、output_model_prompt 四个属性并列存在（见 agent.py）。从示例文件之间的对照注释可以确认两者的分工：output_model.py 的文档字符串明确写道“For structured JSON output, use parser_model instead”——即 output_model 用于生成精炼的自然语言最终回复，parser_model 用于结构化 JSON 的解析抽取，两者不要混用。

output_model：用一个更强的模型“润色”最终回复

output_model.py 演示了“双模型”分工：主模型负责推理和工具调用，output_model 拿到同样的对话后生成自己的回复并替换主模型的输出：

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_prompt 用来单独定制输出模型的写作风格，与 Agent 主体的 description/instructions 解耦。

response_as_variable 与 save_to_file：响应的捕获与落盘

把响应捕获为变量

response_as_variable.py 展示了非流式调用时如何拿到完整的 RunOutput 对象：

from agno.agent import Agent, 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)
    # 流式版本（示例中注释掉）：
    # for response in agent.run("...", stream=True):
    #     pprint(response)

run_response.content 承载文本（或 output_schema 定义的结构化对象），后续代码可以直接对响应做程序化处理。

自动保存到文件

save_to_file.py 只加一个参数即可让每次响应自动写入文件：

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

源码层面的行为比示例注释更丰富。在 _run.py 中可以看到：save_response_to_file 是一个模板字符串，支持 {name}、{session_id}、{user_id}、{message}、{run_id} 五个占位符；框架会在写入前对这些值做路径穿越字符清理（/、\、.. 均被替换），父目录不存在时自动 mkdir(parents=True)，且只有 content 是字符串时才写入文件；若输入不是字符串，会记录一条“未将 input 用于输出文件名”的警告。因此实际生产中可以把它配置成如 tmp/{run_id}_{name}.md 这样的模式来区分不同会话的输出。

streaming：逐 token 流式输出

streaming.py 是最基础的流式用法——print_response(..., stream=True) 即可逐 token 打印：

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

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

markdown=True 让终端按 Markdown 样式渲染流式内容。流式与非流式的分界也很清晰：stream=True 时 run 返回事件迭代器（Iterator[RunOutputEvent]，参见 response_as_variable.py 中被注释掉的流式版本），逐事件处理；stream=False（默认）时返回单个 RunOutput。

followups：一次标志位开启内置“后续问题建议”

followup_suggestions.py 演示了内置 followups 功能。它的文档字符串把关键概念讲得很清楚：

• followups=True 开启功能；
• num_followups 控制建议数量（默认 3）；
• followup_model 可选地用一个更便宜的模型专门生成建议；
• run_response.followups 拿到结构化结果；
• 主回复完全不受约束，照常自由流式输出。

完整示例：

agent = Agent(
    model=OpenAIResponses(id="gpt-5.6-luna"),
    instructions="You are a knowledgeable assistant. Answer questions thoroughly.",
    followups=True,
    num_followups=3,
    # 可选：用更便宜的模型生成 followups
    # followup_model=OpenAIResponses(id="gpt-5.6-luna"),
    markdown=True,
)

if __name__ == "__main__":
    run: RunOutput = agent.run("Which national park is the best?")
    print(run.content)  # 主回复：完整自由文本
    if run.followups:
        for i, suggestion in enumerate(run.followups, 1):
            print(f"  {i}. {suggestion}")

原理上，主响应结束后 Agent 会额外发起一次模型调用，生成结构化的 followup 问题列表并挂到 RunOutput 上。源码印证了这一点：followups: bool = False、num_followups: int = 3、followup_model: Optional[Model] = None 均为 Agent 的正式属性（见 agent.py），且构造函数中校验了 num_followups < 1 时抛出 ValueError（agent.py）；生成逻辑集中在 _response.py 中，统一以 agent.num_followups 作为生成数量上限。这套机制适合对话产品：主回答保持自然语言体验，追问建议作为结构化 UI 元素渲染。

小结：选对输入输出组合

结合本目录 10 个示例，可以形成一个选择矩阵：

需求	推荐参数/方式	示例文件
软性引导回复格式	expected_output	expected_output.py
多模态/消息体输入	dict 形式的 input	input_formats.py
强输入契约	input_schema（Pydantic 模型）	input_schema.py
结构化输出	output_schema	output_schema.py
结构化输出 + 指定解析模型	output_schema + parser_model	parser_model.py
更强模型润色最终文本	output_model + output_model_prompt	output_model.py
程序化处理响应	agent.run() 返回 RunOutput	response_as_variable.py
响应自动落盘	save_response_to_file（支持占位符模板）	save_to_file.py
逐 token 输出	print_response(..., stream=True)	streaming.py
自动追问建议	followups=True + num_followups	followup_suggestions.py

所有示例均可按 README.md 中的方式（./scripts/demo_setup.sh 创建环境后，用 .venvs/demo/bin/python cookbook/02_agents/02_input_output/<file>.py 运行）直接复制运行；涉及 OpenAI Responses 模型的示例需要配置 OPENAI_API_KEY。