如何让 LlamaIndex FunctionAgent 返回 Pydantic 结构化的响应-港品优选

如何让 LlamaIndex FunctionAgent 返回 Pydantic 结构化的响应

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index

在做 Agent 应用时，常见的诉求不是拿到一段自然语言，而是拿到一份可以直接被程序消费的、符合指定 Schema 的结果。LlamaIndex 的FunctionAgent（以及ReActAgent、AgentWorkflow）支持把 Agent 的执行结果转成结构化的 JSON，有两种方式：

output_cls—— 指定一个 Pydantic 模型作为输出 Schema；
structured_output_fn—— 提供一个自定义函数，对 Agent 的对话记录做校验或改写，转成你需要的任意模型。

本文以FunctionAgent为主路径，按 Using Structured Output 的示例走一遍完整流程：定义 Pydantic 模型、配置 Agent、运行任务、读取并校验结构化结果，最后介绍在 Agent 运行过程中流式获取结构化结果的事件。

准备：安装依赖与配置 API Key

按 Building an agent 的准备步骤，安装核心包和 OpenAI LLM 包：

pip install llama-index-core llama-index-llms-openai python-dotenv

Agent 需要一个可用的 LLM。以 OpenAI 为例，在项目根目录创建.env文件并写入 API Key（OPENAI_API_KEY的值替换为你自己的 Key）：

OPENAI_API_KEY=sk-proj-xxxx

文档同时提醒：Agent 对模型能力有要求，较小的模型表现可能不够可靠，可以选择任何受支持的 LLM（包括本地模型），不限于 OpenAI。

主路径：用`output_cls`指定 Pydantic 输出模型

这是最直接的方式：定义一个BaseModel子类描述期望的输出结构，在创建FunctionAgent时通过output_cls传入。文档示例是一个计算器场景：

import asyncio from dotenv import load_dotenv load_dotenv() from llama_index.core.agent.workflow import FunctionAgent from llama_index.llms.openai import OpenAI from pydantic import BaseModel, Field llm = OpenAI(model="gpt-4.1") # 定义结构化输出格式 class MathResult(BaseModel): operation: str = Field(description="the performed operation") result: int = Field(description="the result of the operation") # 定义工具 def multiply(x: int, y: int): """Multiply two numbers""" return x * y # 定义 agent，关键是 output_cls 参数 agent = FunctionAgent( tools=[multiply], name="calculator", system_prompt="You are a calculator agent who can multiply two numbers using the `multiply` tool.", output_cls=MathResult, llm=llm, ) async def main(): response = await agent.run("What is 3415 * 43144?") print(response.structured_response) print(response.get_pydantic_model(MathResult)) if __name__ == "__main__": asyncio.run(main())

几点说明：

output_cls接收一个 Pydantic 模型类，Agent 运行结束后会按这个模型校验输出。模型字段的Field(description=...)是给模型理解每个字段含义用的，描述写清楚有助于输出准确。
agent.run()是异步调用。在 Notebook 环境可以直接await；在普通 Python 脚本里需要像上面一样包一层async def main()并用asyncio.run(main())启动（这一写法来自 Building an agent）。
如果所用模型不支持流式输出（流式默认开启），可以设置FunctionAgent(..., streaming=False)关闭流式来规避报错。

如何读取和验证结构化结果

AgentOutput提供两个入口读取结构化结果（实现见 workflow_events.py）：

# 原始字典形式 response.structured_response # Optional[Dict[str, Any]] # 按 Pydantic 模型校验后的实例 response.get_pydantic_model(MathResult)

structured_response是Dict[str, Any]类型，当没有配置output_cls或structured_output_fn时为None，所以访问前先确认已经配置了其中之一；
get_pydantic_model(model)内部调用model.model_validate(self.structured_response)，返回经过 Pydantic 校验的模型实例。字段类型、必填项不满足时 Pydantic 会抛出校验错误，这本身就是一道可用性检查：能从get_pydantic_model拿到实例，说明 Agent 输出的键值与你的 Schema 完全匹配。

运行脚本后，终端会依次打印structured_response字典和MathResult实例，字段内容与你的BaseModel定义一一对应，即说明结构化输出链路打通。

替代路径：`structured_output_fn`自定义转换函数

当output_cls的默认转换不满足需求时（比如需要基于完整对话历史再做一次改写或校验），可以提供自定义函数。它接收 Agent 运行产生的ChatMessage序列，返回一个字典（可被转成BaseModel子类）。文档示例（冰淇淋口味场景）：

import asyncio import json from dotenv import load_dotenv load_dotenv() from llama_index.core.agent.workflow import FunctionAgent from llama_index.core.llms import ChatMessage from llama_index.llms.openai import OpenAI from pydantic import BaseModel, Field from typing import List, Dict, Any llm = OpenAI(model="gpt-4.1") class Flavor(BaseModel): flavor: str with_sugar: bool async def structured_output_parsing( messages: List[ChatMessage], ) -> Dict[str, Any]: # 用结构化 LLM 对对话历史做二次提取 sllm = llm.as_structured_llm(Flavor) messages.append( ChatMessage( role="user", content="Given the previous message history, structure the output based on the provided format.", ) ) response = await sllm.achat(messages) return json.loads(response.message.content) def get_flavor(ice_cream_shop: str): return "Strawberry with no extra sugar" agent = FunctionAgent( tools=[get_flavor], name="ice_cream_shopper", system_prompt="You are an agent that knows the ice cream flavors in various shops.", structured_output_fn=structured_output_parsing, llm=llm, ) async def main(): response = await agent.run( "What strawberry flavor is available at Gelato Italia?" ) print(response.structured_response) print(response.get_pydantic_model(Flavor)) if __name__ == "__main__": asyncio.run(main())

这里的关键点：

structured_output_fn接收 Agent 记忆中的ChatMessage列表，返回字典；文档说明它适合"validate or rewrite the agent's conversation into any model"的进阶用法。
示例中通过llm.as_structured_llm(Flavor)拿到一个结构化 LLM，把对话历史加上一条引导消息后再调用achat，相当于用第二次 LLM 调用来完成从对话到 Schema 的转换。
读取方式与主路径相同：response.structured_response取字典，response.get_pydantic_model(Flavor)取校验后的模型。

优先级注意：如果同时传了output_cls和structured_output_fn，structured_output_fn会被忽略（见 base_agent.py 中两个字段的描述：output_cls非空时structured_output_fnis ignored）。两条路径只能选一条。

可选：在 Agent 运行过程中流式获取结构化输出

如果希望在 Agent 执行期间就拿到结构化结果（而不是等run结束），可以监听AgentStreamStructuredOutput事件。文档给出的做法是不await运行结果，改为遍历handler.stream_events()：

from llama_index.core.agent.workflow import ( AgentInput, AgentOutput, ToolCall, ToolCallResult, AgentStreamStructuredOutput, ) handler = agent.run("What strawberry flavor is available at Gelato Italia?") async for event in handler.stream_events(): if isinstance(event, AgentInput): print(event) elif isinstance(event, AgentStreamStructuredOutput): print(event.output) print(event.get_pydantic_model(Flavor)) elif isinstance(event, ToolCallResult): print(event) elif isinstance(event, ToolCall): print(event) elif isinstance(event, AgentOutput): print(event) else: pass response = await handler

说明：

原示例代码中这一行写作event.get_pydantic_model(Weather)，但Weather是文档前面多 Agent 示例里的模型，与冰淇淋 Agent 对应的模型是Flavor；上面代码块已按上下文修正为Flavor，请以你实际定义的模型类为准。
遍历结束后再await handler得到最终AgentOutput，随后仍可用response.structured_response/response.get_pydantic_model(...)读取。
AgentStreamStructuredOutput属于AgentWorkflow的事件体系，事件列表见 Streaming output and events。

边界与注意事项

结构化输出不仅限于单 Agent：文档明确AgentWorkflow多 Agent 工作流同样支持output_cls和structured_output_fn，用法是在构建AgentWorkflow时传入，例如AgentWorkflow(agents=[...], root_agent=..., output_cls=Weather)（见 Using Structured Output 中的天气 Agent 多 Agent 示例）。
若所用模型不支持流式，设置streaming=False创建FunctionAgent（来自 Building an agent 的提示）。
structured_response在未配置output_cls或structured_output_fn时为None；get_pydantic_model校验失败会抛 Pydantic 的校验异常，这两点可以直接用作"结构化输出是否按预期工作"的判断依据。

完成以上任一路径后，response.get_pydantic_model(你的模型类)能返回校验通过的实例，即代表FunctionAgent的结构化响应链路已经按你的 Schema 生效。

【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析