1. 从“大海捞针”到“按图索骥”为什么我们需要结构化输出如果你用过早期的RAG检索增强生成系统或者尝试过直接向大语言模型LLM提问你大概率经历过这种“抓狂”时刻你问“帮我总结一下上周的销售数据”它给你回了一段洋洋洒洒的散文里面夹杂着各种数字和描述但你真正想要的可能是一个可以直接导入Excel的表格或者一个能喂给下游程序的JSON对象。你不得不手动从那段“小作文”里抠出关键信息费时费力还容易出错。这就是“非结构化输出”的典型困境。LLM很强大但它默认的“聊天”模式输出的是自然语言文本。这种文本对人类阅读友好但对机器处理极不友好。想象一下你想让AI自动分析100份产品反馈并把“产品名称”、“问题类型”、“严重程度”、“建议”这几个字段提取出来。如果AI每次都用一段话回复后续的自动化流程就卡壳了因为你得先写一个复杂的文本解析器。结构化输出就是为了解决这个问题而生。它本质上是一种“约束性生成”要求LLM严格按照我们预先定义好的格式来回答问题。这个格式可以是一个JSON Schema定义字段名、类型、是否必填一个Pydantic模型在Python中定义数据结构和验证规则甚至是一个简单的Markdown表格。它的目标是让AI的输出从“散文”变成“表格”或“数据库记录”变得可预测、可解析、可编程。而LlamaIndex作为构建高级RAG和AI应用的热门框架其核心价值之一就是充当LLM与你的数据、你的业务逻辑之间的“智能粘合剂”。它不仅要能帮你找到相关的数据片段检索更要能帮你把这些信息加工成你业务中真正需要的形态生成与结构化。因此掌握LlamaIndex的结构化输出能力意味着你能将AI从“一个聪明的聊天伙伴”升级为“一个可靠的数据处理流水线工人”。它能直接产出你的代码、你的数据库、你的报表系统所能理解和消费的“原料”。最近社区里关于llamaindex和llamaindex langgraph的讨论热度很高尤其是结合llamaindex rag实战时大家越来越不满足于简单的问答而是追求构建复杂、稳定、能嵌入生产流程的智能体Agent或工作流。在这种场景下结构化输出不再是“锦上添花”而是“雪中送炭”的必备技能。它决定了你的AI应用是停留在演示阶段还是能真正落地创造价值。2. LlamaIndex实现结构化输出的核心武器Pydantic与函数调用在LlamaIndex中实现结构化输出主要依赖于两大核心机制它们都深度整合了LLM的“函数调用”Function Calling或“工具使用”Tool Use能力。理解这两者的区别和适用场景是玩转这项技术的关键。2.1 基石Pydantic模型——定义你期望的“数据结构”Pydantic是一个Python库主要利用Python的类型注解来进行数据验证和设置管理。在LlamaIndex的语境下我们用它来定义一个“输出蓝图”。假设我们正在构建一个智能新闻阅读助手我们希望它从一篇长文中提取关键信息。我们可以这样定义一个Pydantic模型from pydantic import BaseModel, Field from typing import List, Optional class NewsSummary(BaseModel): 从新闻文章中提取的结构化摘要 headline: str Field(description新闻的核心标题需简洁有力) key_points: List[str] Field(description文章的3-5个核心要点每条不超过20字) sentiment: str Field(description文章的整体情感倾向可选值positive, neutral, negative) mentioned_companies: Optional[List[str]] Field(defaultNone, description文中提及的公司名称列表) summary: str Field(description一段完整的摘要约100字)这个NewsSummary类就是我们给LLM的“填空题”模板。Field中的description字段至关重要它是你与LLM沟通的“需求说明书”告诉它每个字段应该填什么内容、有什么格式要求。Optional表示该字段可以为空。为什么是Pydantic类型安全Python的类型提示str,List[str]为LLM和后续代码提供了明确的期望。验证内置Pydantic会自动验证LLM返回的数据是否符合类型定义如果LLM胡言乱语返回了一个数字给headline字段Pydantic会抛出验证错误让你的程序更健壮。无缝集成Pydantic是FastAPI等现代Python框架的标配这意味着你从LlamaIndex获得的结构化数据可以几乎无成本地转换为API响应、数据库记录或配置文件。2.2 机制一PydanticOutputParser——直接的格式转换器这是最直观的方式。你创建一个PydanticOutputParser将它和你定义的Pydantic模型绑定然后将其作为“输出处理器”插入到你的查询引擎或LLM调用中。from llama_index.core.output_parsers import PydanticOutputParser from llama_index.core.query_engine import CustomQueryEngine from llama_index.llms.openai import OpenAI # 1. 创建解析器绑定到我们的数据模型 parser PydanticOutputParser(output_clsNewsSummary) # 2. 构建一个提示模板其中包含格式指令 from llama_index.core import PromptTemplate prompt_str 请根据以下上下文信息提取并结构化新闻内容。 上下文 {context_str} 请严格按照以下JSON格式输出 {format_instructions} 文章内容 {query} prompt PromptTemplate(prompt_str, output_parserparser) # 3. 在查询引擎中使用 # 假设你已经有了一个索引index和检索器retriever query_engine index.as_query_engine( llmOpenAI(modelgpt-4), text_qa_templateprompt, # 使用我们自定义的提示模板 response_modecompact ) # 4. 执行查询 response query_engine.query(分析这篇关于人工智能的新闻) # 此时response.response 可能还是一段文本但我们可以用解析器处理 # 更常见的做法是在高级查询引擎中直接配置输出解析器工作流程LLM在生成文本时解析器会通过提示词中的{format_instructions}由解析器自动生成的一段关于JSON格式的描述来约束LLM。LLM输出一个符合格式的JSON字符串然后解析器会尝试将这个字符串解析并实例化成NewsSummary对象。优点概念简单与控制提示词结合紧密。缺点需要手动管理提示词和格式指令的拼接对于复杂嵌套对象提示词可能会变得冗长。2.3 机制二PydanticProgram——更强大的“结构化任务执行器”这是LlamaIndex更推荐、也更强大的方式。PydanticProgram是一个更高层次的抽象它将LLM视为一个可以执行“返回特定类型对象”这一任务的函数。from llama_index.program.openai import OpenAIPydanticProgram from llama_index.llms.openai import OpenAI # 1. 定义你的Pydantic模型 (同上NewsSummary) # 2. 创建PydanticProgram program OpenAIPydanticProgram.from_defaults( output_clsNewsSummary, llmOpenAI(modelgpt-4-turbo-preview), # 指定LLM prompt_template_str( 请根据用户提供的新闻文章生成一个结构化的摘要。\n 文章内容{input} ), ) # 3. 像调用函数一样调用它 result: NewsSummary program(input一篇很长的新闻文章文本...) print(result.headline) print(result.key_points) for company in result.mentioned_companies or []: print(company)背后的魔法当你调用program()时LlamaIndex在底层自动完成了一系列操作根据output_clsNewsSummary生成一个详细的JSON Schema。利用LLM的函数调用能力如OpenAI的tools参数将这个JSON Schema作为一个“工具”描述发送给LLM。LLM理解任务后会直接返回一个符合该Schema的JSON对象而不是中间的自然语言文本。PydanticProgram接收这个JSON并用它实例化NewsSummary对象同时进行Pydantic验证。核心优势更可靠直接利用LLM的原生函数调用功能格式遵从性远高于通过文本提示约束。更简洁开发者无需操心格式指令的拼接框架自动处理。更高效减少了“LLM生成文本 - 程序解析文本”的中间环节出错率更低。实操心得在绝大多数生产场景中优先选择PydanticProgram。它不仅是结构化输出更是将LLM“封装”成一个类型安全、功能明确的函数这是构建复杂AI工作流如使用LangGraph编排多个AI步骤的理想基石。只有当你需要对提示词进行极其精细的控制或者使用的LLM不支持函数调用时才考虑使用PydanticOutputParser。3. 实战构建一个会议纪要自动生成器让我们通过一个完整的例子将理论付诸实践。假设我们需要从一场会议的录音转写文本中自动提取结构化信息生成会议纪要。3.1 定义核心数据结构首先我们需要思考一份会议纪要包含哪些结构化信息。这比新闻摘要更复杂涉及多层嵌套。from pydantic import BaseModel, Field from typing import List, Optional from datetime import time from enum import Enum class Speaker(BaseModel): name: str Field(description发言人姓名或标识) department: Optional[str] Field(defaultNone, description所属部门) class AgendaItem(BaseModel): topic: str Field(description讨论议题) start_time: Optional[str] Field(defaultNone, description开始时间格式 HH:MM) end_time: Optional[str] Field(defaultNone, description结束时间格式 HH:MM) key_discussion_points: List[str] Field(description该议题下的关键讨论点) decisions_made: List[str] Field(description达成的决议或结论) action_items: List[str] Field(description产生的行动项格式建议为‘负责人任务描述’) primary_speakers: List[Speaker] Field(description主要发言人列表) class MeetingType(Enum): STANDUP 每日站会 BRAINSTORM 头脑风暴 REVIEW 评审会 DECISION 决策会 OTHER 其他 class StructuredMeetingMinutes(BaseModel): 结构化会议纪要 meeting_title: str Field(description会议主题) meeting_type: MeetingType Field(description会议类型) date: str Field(description会议日期格式 YYYY-MM-DD) participants: List[Speaker] Field(description全体参会者列表) agenda: List[AgendaItem] Field(description会议议程项列表) overall_summary: str Field(description会议整体总结约200字) next_meeting_time: Optional[str] Field(defaultNone, description下次会议时间)这个模型定义了从会议文本到结构化数据的完整映射。注意AgendaItem中包含List[Speaker]StructuredMeetingMinutes中又包含List[AgendaItem]形成了一个嵌套结构。LLM特别是GPT-4级别完全有能力处理这种复杂嵌套。3.2 实现自动提取程序接下来我们使用OpenAIPydanticProgram来创建提取器。import os from llama_index.program.openai import OpenAIPydanticProgram from llama_index.llms.openai import OpenAI # 假设你的OpenAI API Key已设置在环境变量中 os.environ[OPENAI_API_KEY] your-api-key llm OpenAI(modelgpt-4-turbo-preview) # 复杂任务建议使用更强模型 meeting_minutes_program OpenAIPydanticProgram.from_defaults( output_clsStructuredMeetingMinutes, llmllm, prompt_template_str你是一个专业的会议秘书。请根据以下会议转录文本生成一份详尽的结构化会议纪要。 转录文本可能冗长、杂乱包含口语化表达和重复内容。你的任务是识别关键信息并将其精准地填充到以下数据结构中。 会议转录文本 {input} 请确保 1. 从文本中推断会议类型如站会、评审会等。 2. 识别并区分不同的议题AgendaItem。如果文本中没有明确的时间可以合理推断或留空。 3. 行动项action_items必须清晰最好有明确的负责人。 4. 参会者列表participants应从全文提及的人名中归纳得出。 , ) # 读取会议转录文本 with open(meeting_transcript.txt, r, encodingutf-8) as f: transcript f.read() # 执行提取 try: minutes: StructuredMeetingMinutes meeting_minutes_program(inputtranscript) print(f会议主题{minutes.meeting_title}) print(f会议类型{minutes.meeting_type.value}) print(f参会人数{len(minutes.participants)}) for i, item in enumerate(minutes.agenda, 1): print(f\n议题{i}: {item.topic}) print(f 决议{item.decisions_made}) print(f 行动项{item.action_items}) except Exception as e: print(f解析失败{e}) # 这里可以加入重试或降级逻辑3.3 处理复杂性与提升鲁棒性上面的基础版本可能会因为转录文本质量差、信息模糊而失败。我们需要增强程序的鲁棒性。策略一提供少量示例Few-Shot Prompting在prompt_template_str中除了指令还可以提供一两个例子。这能极大地提升LLM对任务格式和期望的理解。prompt_template_str 你是一个专业的会议秘书。请根据以下会议转录文本生成一份详尽的结构化会议纪要。 **输出必须严格遵循下面定义的JSON Schema。** **示例1** 输入文本“今天我们讨论Q2预算老王说营销部分超了10%小李建议削减活动规模最后决定下周再审。” 输出结构{...} 这里可以粘贴一个符合StructuredMeetingMinutes模型的JSON示例 **示例2** ... 第二个示例 现在请处理真实的会议转录文本 {input} 策略二实现验证与重试机制LLM的输出可能偶尔不符合Pydantic模型。我们需要捕获这些错误并进行处理。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from pydantic import ValidationError retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((ValidationError, ValueError)), reraiseTrue ) def get_structured_minutes_with_retry(transcript: str) - StructuredMeetingMinutes: 带重试的结构化提取 # 可以在每次重试时微调提示词例如增加“请更加仔细地提取行动项负责人”等指令 minutes meeting_minutes_program(inputtranscript) # PydanticProgram内部会进行验证验证失败会抛出ValidationError return minutes # 使用重试函数 try: minutes get_structured_minutes_with_retry(transcript) except Exception as e: print(f经过多次尝试仍然解析失败{e}) # 降级方案回退到非结构化输出或通知人工处理策略三与RAG流程结合如果会议转录文本非常长比如数小时超出了LLM的上下文窗口直接扔给LLM是不行的。这时就需要结合LlamaIndex的RAG能力。索引转录文本将长转录文本分割成块建立向量索引。分层摘要先让LLM为每个文本块生成一个AgendaItem的草稿。汇总与精炼将所有草稿AgendaItem和相关元数据如发言人频率作为上下文再次调用PydanticProgram让它生成最终的、统一的StructuredMeetingMinutes。这个过程可以用LangGraph来编排定义“检索 - 初步提取 - 汇总”的工作流这正是llamaindex langgraph在rag实战中的高级应用场景。踩坑实录在定义嵌套模型时Field(description...)的描述语质量直接决定输出质量。不要写“讨论要点”这种模糊的话要写“列出关于产品设计的3个主要反对意见及其理由”。越具体LLM执行得越好。另外对于Optional字段如果LLM经常忽略可以在描述中强调“如果未提及请设置为null”并在后续代码中做好空值处理。4. 进阶在LangGraph智能体工作流中应用结构化输出当你的应用从单一问答升级到多步骤、有状态的智能体工作流时结构化输出的价值会呈指数级放大。LangGraph是一个用于构建有状态、多智能体应用的框架而LlamaIndex提供了与它深度集成的能力。设想一个“市场调研智能体”工作流步骤一搜索根据用户提出的公司名从网络搜索最新新闻。步骤二分析对抓取的新闻内容进行情感分析和关键事件提取结构化输出。步骤三报告将多个分析结果汇总生成一份统一的调研报告另一种结构化输出。在这个工作流中步骤二和步骤三的输出必须是结构化的这样才能被后续的节点步骤作为可靠的、类型明确的输入数据来消费。4.1 定义智能体间的“通信协议”在LangGraph中每个节点的输入和输出通常放在一个共享的“状态”字典里。结构化输出模型就是节点间最好的“通信协议”。from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph, END import operator # 1. 定义工作流的全局状态 class AgentState(TypedDict): company_name: str raw_news_articles: List[str] # 步骤一的输出 news_analyses: List[NewsSummary] # 步骤二的输出结构化 final_report: Optional[MarketResearchReport] # 步骤三的输出另一个结构化模型 errors: List[str] # 2. 定义“分析”节点函数 def analyze_news_node(state: AgentState) - AgentState: 分析原始新闻生成结构化NewsSummary列表 analyses [] for article in state[raw_news_articles]: try: # 使用前面定义的PydanticProgram analysis: NewsSummary news_analysis_program(inputarticle) analyses.append(analysis) except Exception as e: state[errors].append(f分析文章失败{e}) # 可以放入一个空的或标记错误的分析对象 analyses.append(NewsSummary(headline解析失败, key_points[], sentimentneutral, summary)) return {news_analyses: analyses} # 更新状态中的结构化数据 # 3. 定义“报告生成”节点函数 def generate_report_node(state: AgentState) - AgentState: 基于结构化分析列表生成最终报告 if not state[news_analyses]: return {final_report: None} # 我们可以定义另一个Pydantic模型 MarketResearchReport # 它可能包含 company_name, overall_sentiment, risk_factors(List), opportunity_areas(List) 等字段 # 然后创建一个新的 program 来汇总 news_analyses report_input f公司{state[company_name]}\n分析结果{state[news_analyses]} try: report: MarketResearchReport report_generation_program(inputreport_input) return {final_report: report} except Exception as e: state[errors].append(f生成报告失败{e}) return {final_report: None} # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(search_news, search_news_node) # 假设已实现 workflow.add_node(analyze_news, analyze_news_node) workflow.add_node(generate_report, generate_report_node) workflow.set_entry_point(search_news) workflow.add_edge(search_news, analyze_news) workflow.add_edge(analyze_news, generate_report) workflow.add_edge(generate_report, END) app workflow.compile()4.2 结构化输出带来的优势在这个工作流中NewsSummary和MarketResearchReport这两个Pydantic模型起到了关键作用接口清晰analyze_news_node的产出类型是List[NewsSummary]generate_report_node的消费类型也是它。这避免了节点之间传递模糊的文本字符串需要靠“默契”或复杂的解析来理解。错误隔离如果一个新闻分析失败了我们可以捕获异常在NewsSummary对象中标记错误而不会让整个工作流崩溃。后续节点可以通过检查字段来判断数据质量。可测试性你可以轻松地为analyze_news_node编写单元测试用一篇固定的文章断言其输出是否符合NewsSummary的格式和预期的内容。状态可观测在调试时你可以检查state[news_analyses]看到的是一个清晰的对象列表每个对象都有明确的字段而不是一堆杂乱无章的文本。核心经验在基于LangGraph或任何工作流引擎构建复杂AI应用时将每个核心步骤的输入和输出用Pydantic模型进行结构化定义是保证系统可维护、可调试、可扩展的最重要实践。它迫使你明确每个节点的“契约”将不可靠的LLM自由文本输出转化为你代码中可靠的、类型化的数据流。这才是llamaindex rag实战从玩具走向生产系统的关键一步。5. 性能优化与生产级考量将结构化输出用于生产环境除了功能正确我们还需要关注性能、成本和稳定性。5.1 提示词工程精确控制降低成本LLM的令牌Token使用量直接关联成本。复杂的Pydantic模型会产生冗长的JSON Schema增加提示词长度。优化策略1简化模型描述在Field(description...)中使用最精炼的语言。避免冗长的句子用分号分隔要点。# 欠佳 topic: str Field(description这是一个讨论的议题请你从文本中找出大家主要在讨论什么话题并用一个简短的名词性短语概括它) # 更佳 topic: str Field(description讨论议题用简短名词短语概括)优化策略2使用OpenAIPydanticProgram的function_call模式这是默认且推荐的方式。它利用OpenAI API的tools参数比将Schema塞进system或user提示词更高效、更可靠。确保你使用的模型如gpt-4-turbo-preview,gpt-3.5-turbo支持此功能。优化策略3流式输出与部分解析对于生成时间较长的复杂对象可以考虑是否支持流式输出。虽然目前PydanticProgram通常返回完整对象但你可以设计自己的流程先让LLM输出最重要的字段如headline,sentiment再根据需要逐步获取其他字段从而实现快速初步响应。5.2 错误处理与降级方案LLM并非百分之百可靠必须设计容错机制。验证失败处理PydanticProgram在实例化对象时会自动验证。捕获ValidationError并根据错误类型采取行动。from pydantic import ValidationError try: result program(inputtext) except ValidationError as e: logging.warning(fLLM输出验证失败: {e.errors()}) # 降级方案A使用一个包含错误信息的默认对象 result OutputModel(..., error解析失败, raw_llm_outputfallback_text) # 降级方案B触发一次重试并附加更严格的指令 result program_with_stricter_prompt(inputtext)LLM API异常处理网络超时、速率限制、服务不可用等。使用具有重试和回退策略的客户端如tenacity库。from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type import openai retry( stopstop_after_attempt(5), waitwait_random_exponential(multiplier1, max60), retryretry_if_exception_type((openai.APITimeoutError, openai.RateLimitError)) ) def robust_program_call(input_text): return program(inputinput_text)内容安全与过滤如果处理用户生成的或来自不可信源的文本LLM可能被诱导输出有害或不符合格式的内容。除了Pydantic验证还应在业务逻辑层对关键字段如URL、人名进行额外的清洗和过滤。5.3 缓存与版本控制对于相同或相似的输入重复调用LLM生成结构化输出是巨大的浪费。语义缓存使用LlamaIndex的SemanticCache或类似向量缓存方案。当一个新的查询进来时先计算其嵌入向量在缓存中查找语义相似的已有查询及其结构化输出结果。如果找到且相似度超过阈值直接返回缓存的结果无需调用LLM。这能极大降低成本和延迟。模型版本化你的Pydantic模型NewsSummary可能会迭代。为模型添加版本号字段或在数据库存储时记录模型的结构定义Schema。这样当模型变更后你仍然能正确解析历史上缓存的数据或者知道哪些数据需要重新处理。class NewsSummary(BaseModel): model_version: str 1.0.1 # 显式声明版本 headline: str ... # ... 其他字段5.4 评估与监控如何知道你的结构化输出管道工作得好不好人工评估样本定期抽样检查评估提取的准确性、完整性和格式正确性。自动化指标模式符合率成功通过Pydantic验证的比例。关键字段填充率对于必填字段LLM成功提取的比例。与黄金标注的对比对于有标注的数据计算字段级别的精确率、召回率或F1分数。监控与告警监控API调用耗时、Token消耗、验证错误率。当错误率突然上升或耗时异常时触发告警。结构化输出不是一次性的技巧而是一套系统工程。从清晰的数据模型定义到可靠的程序调用再到生产环境的性能、鲁棒性和可观测性建设每一步都需要精心设计。当你把这些都做好LlamaIndex就不再只是一个检索工具而成为了你业务中一个强大的、自动化的“信息结构化工匠”。
1. 从“大海捞针”到“按图索骥”为什么我们需要结构化输出如果你用过早期的RAG检索增强生成系统或者尝试过直接向大语言模型LLM提问你大概率经历过这种“抓狂”时刻你问“帮我总结一下上周的销售数据”它给你回了一段洋洋洒洒的散文里面夹杂着各种数字和描述但你真正想要的可能是一个可以直接导入Excel的表格或者一个能喂给下游程序的JSON对象。你不得不手动从那段“小作文”里抠出关键信息费时费力还容易出错。这就是“非结构化输出”的典型困境。LLM很强大但它默认的“聊天”模式输出的是自然语言文本。这种文本对人类阅读友好但对机器处理极不友好。想象一下你想让AI自动分析100份产品反馈并把“产品名称”、“问题类型”、“严重程度”、“建议”这几个字段提取出来。如果AI每次都用一段话回复后续的自动化流程就卡壳了因为你得先写一个复杂的文本解析器。结构化输出就是为了解决这个问题而生。它本质上是一种“约束性生成”要求LLM严格按照我们预先定义好的格式来回答问题。这个格式可以是一个JSON Schema定义字段名、类型、是否必填一个Pydantic模型在Python中定义数据结构和验证规则甚至是一个简单的Markdown表格。它的目标是让AI的输出从“散文”变成“表格”或“数据库记录”变得可预测、可解析、可编程。而LlamaIndex作为构建高级RAG和AI应用的热门框架其核心价值之一就是充当LLM与你的数据、你的业务逻辑之间的“智能粘合剂”。它不仅要能帮你找到相关的数据片段检索更要能帮你把这些信息加工成你业务中真正需要的形态生成与结构化。因此掌握LlamaIndex的结构化输出能力意味着你能将AI从“一个聪明的聊天伙伴”升级为“一个可靠的数据处理流水线工人”。它能直接产出你的代码、你的数据库、你的报表系统所能理解和消费的“原料”。最近社区里关于llamaindex和llamaindex langgraph的讨论热度很高尤其是结合llamaindex rag实战时大家越来越不满足于简单的问答而是追求构建复杂、稳定、能嵌入生产流程的智能体Agent或工作流。在这种场景下结构化输出不再是“锦上添花”而是“雪中送炭”的必备技能。它决定了你的AI应用是停留在演示阶段还是能真正落地创造价值。2. LlamaIndex实现结构化输出的核心武器Pydantic与函数调用在LlamaIndex中实现结构化输出主要依赖于两大核心机制它们都深度整合了LLM的“函数调用”Function Calling或“工具使用”Tool Use能力。理解这两者的区别和适用场景是玩转这项技术的关键。2.1 基石Pydantic模型——定义你期望的“数据结构”Pydantic是一个Python库主要利用Python的类型注解来进行数据验证和设置管理。在LlamaIndex的语境下我们用它来定义一个“输出蓝图”。假设我们正在构建一个智能新闻阅读助手我们希望它从一篇长文中提取关键信息。我们可以这样定义一个Pydantic模型from pydantic import BaseModel, Field from typing import List, Optional class NewsSummary(BaseModel): 从新闻文章中提取的结构化摘要 headline: str Field(description新闻的核心标题需简洁有力) key_points: List[str] Field(description文章的3-5个核心要点每条不超过20字) sentiment: str Field(description文章的整体情感倾向可选值positive, neutral, negative) mentioned_companies: Optional[List[str]] Field(defaultNone, description文中提及的公司名称列表) summary: str Field(description一段完整的摘要约100字)这个NewsSummary类就是我们给LLM的“填空题”模板。Field中的description字段至关重要它是你与LLM沟通的“需求说明书”告诉它每个字段应该填什么内容、有什么格式要求。Optional表示该字段可以为空。为什么是Pydantic类型安全Python的类型提示str,List[str]为LLM和后续代码提供了明确的期望。验证内置Pydantic会自动验证LLM返回的数据是否符合类型定义如果LLM胡言乱语返回了一个数字给headline字段Pydantic会抛出验证错误让你的程序更健壮。无缝集成Pydantic是FastAPI等现代Python框架的标配这意味着你从LlamaIndex获得的结构化数据可以几乎无成本地转换为API响应、数据库记录或配置文件。2.2 机制一PydanticOutputParser——直接的格式转换器这是最直观的方式。你创建一个PydanticOutputParser将它和你定义的Pydantic模型绑定然后将其作为“输出处理器”插入到你的查询引擎或LLM调用中。from llama_index.core.output_parsers import PydanticOutputParser from llama_index.core.query_engine import CustomQueryEngine from llama_index.llms.openai import OpenAI # 1. 创建解析器绑定到我们的数据模型 parser PydanticOutputParser(output_clsNewsSummary) # 2. 构建一个提示模板其中包含格式指令 from llama_index.core import PromptTemplate prompt_str 请根据以下上下文信息提取并结构化新闻内容。 上下文 {context_str} 请严格按照以下JSON格式输出 {format_instructions} 文章内容 {query} prompt PromptTemplate(prompt_str, output_parserparser) # 3. 在查询引擎中使用 # 假设你已经有了一个索引index和检索器retriever query_engine index.as_query_engine( llmOpenAI(modelgpt-4), text_qa_templateprompt, # 使用我们自定义的提示模板 response_modecompact ) # 4. 执行查询 response query_engine.query(分析这篇关于人工智能的新闻) # 此时response.response 可能还是一段文本但我们可以用解析器处理 # 更常见的做法是在高级查询引擎中直接配置输出解析器工作流程LLM在生成文本时解析器会通过提示词中的{format_instructions}由解析器自动生成的一段关于JSON格式的描述来约束LLM。LLM输出一个符合格式的JSON字符串然后解析器会尝试将这个字符串解析并实例化成NewsSummary对象。优点概念简单与控制提示词结合紧密。缺点需要手动管理提示词和格式指令的拼接对于复杂嵌套对象提示词可能会变得冗长。2.3 机制二PydanticProgram——更强大的“结构化任务执行器”这是LlamaIndex更推荐、也更强大的方式。PydanticProgram是一个更高层次的抽象它将LLM视为一个可以执行“返回特定类型对象”这一任务的函数。from llama_index.program.openai import OpenAIPydanticProgram from llama_index.llms.openai import OpenAI # 1. 定义你的Pydantic模型 (同上NewsSummary) # 2. 创建PydanticProgram program OpenAIPydanticProgram.from_defaults( output_clsNewsSummary, llmOpenAI(modelgpt-4-turbo-preview), # 指定LLM prompt_template_str( 请根据用户提供的新闻文章生成一个结构化的摘要。\n 文章内容{input} ), ) # 3. 像调用函数一样调用它 result: NewsSummary program(input一篇很长的新闻文章文本...) print(result.headline) print(result.key_points) for company in result.mentioned_companies or []: print(company)背后的魔法当你调用program()时LlamaIndex在底层自动完成了一系列操作根据output_clsNewsSummary生成一个详细的JSON Schema。利用LLM的函数调用能力如OpenAI的tools参数将这个JSON Schema作为一个“工具”描述发送给LLM。LLM理解任务后会直接返回一个符合该Schema的JSON对象而不是中间的自然语言文本。PydanticProgram接收这个JSON并用它实例化NewsSummary对象同时进行Pydantic验证。核心优势更可靠直接利用LLM的原生函数调用功能格式遵从性远高于通过文本提示约束。更简洁开发者无需操心格式指令的拼接框架自动处理。更高效减少了“LLM生成文本 - 程序解析文本”的中间环节出错率更低。实操心得在绝大多数生产场景中优先选择PydanticProgram。它不仅是结构化输出更是将LLM“封装”成一个类型安全、功能明确的函数这是构建复杂AI工作流如使用LangGraph编排多个AI步骤的理想基石。只有当你需要对提示词进行极其精细的控制或者使用的LLM不支持函数调用时才考虑使用PydanticOutputParser。3. 实战构建一个会议纪要自动生成器让我们通过一个完整的例子将理论付诸实践。假设我们需要从一场会议的录音转写文本中自动提取结构化信息生成会议纪要。3.1 定义核心数据结构首先我们需要思考一份会议纪要包含哪些结构化信息。这比新闻摘要更复杂涉及多层嵌套。from pydantic import BaseModel, Field from typing import List, Optional from datetime import time from enum import Enum class Speaker(BaseModel): name: str Field(description发言人姓名或标识) department: Optional[str] Field(defaultNone, description所属部门) class AgendaItem(BaseModel): topic: str Field(description讨论议题) start_time: Optional[str] Field(defaultNone, description开始时间格式 HH:MM) end_time: Optional[str] Field(defaultNone, description结束时间格式 HH:MM) key_discussion_points: List[str] Field(description该议题下的关键讨论点) decisions_made: List[str] Field(description达成的决议或结论) action_items: List[str] Field(description产生的行动项格式建议为‘负责人任务描述’) primary_speakers: List[Speaker] Field(description主要发言人列表) class MeetingType(Enum): STANDUP 每日站会 BRAINSTORM 头脑风暴 REVIEW 评审会 DECISION 决策会 OTHER 其他 class StructuredMeetingMinutes(BaseModel): 结构化会议纪要 meeting_title: str Field(description会议主题) meeting_type: MeetingType Field(description会议类型) date: str Field(description会议日期格式 YYYY-MM-DD) participants: List[Speaker] Field(description全体参会者列表) agenda: List[AgendaItem] Field(description会议议程项列表) overall_summary: str Field(description会议整体总结约200字) next_meeting_time: Optional[str] Field(defaultNone, description下次会议时间)这个模型定义了从会议文本到结构化数据的完整映射。注意AgendaItem中包含List[Speaker]StructuredMeetingMinutes中又包含List[AgendaItem]形成了一个嵌套结构。LLM特别是GPT-4级别完全有能力处理这种复杂嵌套。3.2 实现自动提取程序接下来我们使用OpenAIPydanticProgram来创建提取器。import os from llama_index.program.openai import OpenAIPydanticProgram from llama_index.llms.openai import OpenAI # 假设你的OpenAI API Key已设置在环境变量中 os.environ[OPENAI_API_KEY] your-api-key llm OpenAI(modelgpt-4-turbo-preview) # 复杂任务建议使用更强模型 meeting_minutes_program OpenAIPydanticProgram.from_defaults( output_clsStructuredMeetingMinutes, llmllm, prompt_template_str你是一个专业的会议秘书。请根据以下会议转录文本生成一份详尽的结构化会议纪要。 转录文本可能冗长、杂乱包含口语化表达和重复内容。你的任务是识别关键信息并将其精准地填充到以下数据结构中。 会议转录文本 {input} 请确保 1. 从文本中推断会议类型如站会、评审会等。 2. 识别并区分不同的议题AgendaItem。如果文本中没有明确的时间可以合理推断或留空。 3. 行动项action_items必须清晰最好有明确的负责人。 4. 参会者列表participants应从全文提及的人名中归纳得出。 , ) # 读取会议转录文本 with open(meeting_transcript.txt, r, encodingutf-8) as f: transcript f.read() # 执行提取 try: minutes: StructuredMeetingMinutes meeting_minutes_program(inputtranscript) print(f会议主题{minutes.meeting_title}) print(f会议类型{minutes.meeting_type.value}) print(f参会人数{len(minutes.participants)}) for i, item in enumerate(minutes.agenda, 1): print(f\n议题{i}: {item.topic}) print(f 决议{item.decisions_made}) print(f 行动项{item.action_items}) except Exception as e: print(f解析失败{e}) # 这里可以加入重试或降级逻辑3.3 处理复杂性与提升鲁棒性上面的基础版本可能会因为转录文本质量差、信息模糊而失败。我们需要增强程序的鲁棒性。策略一提供少量示例Few-Shot Prompting在prompt_template_str中除了指令还可以提供一两个例子。这能极大地提升LLM对任务格式和期望的理解。prompt_template_str 你是一个专业的会议秘书。请根据以下会议转录文本生成一份详尽的结构化会议纪要。 **输出必须严格遵循下面定义的JSON Schema。** **示例1** 输入文本“今天我们讨论Q2预算老王说营销部分超了10%小李建议削减活动规模最后决定下周再审。” 输出结构{...} 这里可以粘贴一个符合StructuredMeetingMinutes模型的JSON示例 **示例2** ... 第二个示例 现在请处理真实的会议转录文本 {input} 策略二实现验证与重试机制LLM的输出可能偶尔不符合Pydantic模型。我们需要捕获这些错误并进行处理。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from pydantic import ValidationError retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((ValidationError, ValueError)), reraiseTrue ) def get_structured_minutes_with_retry(transcript: str) - StructuredMeetingMinutes: 带重试的结构化提取 # 可以在每次重试时微调提示词例如增加“请更加仔细地提取行动项负责人”等指令 minutes meeting_minutes_program(inputtranscript) # PydanticProgram内部会进行验证验证失败会抛出ValidationError return minutes # 使用重试函数 try: minutes get_structured_minutes_with_retry(transcript) except Exception as e: print(f经过多次尝试仍然解析失败{e}) # 降级方案回退到非结构化输出或通知人工处理策略三与RAG流程结合如果会议转录文本非常长比如数小时超出了LLM的上下文窗口直接扔给LLM是不行的。这时就需要结合LlamaIndex的RAG能力。索引转录文本将长转录文本分割成块建立向量索引。分层摘要先让LLM为每个文本块生成一个AgendaItem的草稿。汇总与精炼将所有草稿AgendaItem和相关元数据如发言人频率作为上下文再次调用PydanticProgram让它生成最终的、统一的StructuredMeetingMinutes。这个过程可以用LangGraph来编排定义“检索 - 初步提取 - 汇总”的工作流这正是llamaindex langgraph在rag实战中的高级应用场景。踩坑实录在定义嵌套模型时Field(description...)的描述语质量直接决定输出质量。不要写“讨论要点”这种模糊的话要写“列出关于产品设计的3个主要反对意见及其理由”。越具体LLM执行得越好。另外对于Optional字段如果LLM经常忽略可以在描述中强调“如果未提及请设置为null”并在后续代码中做好空值处理。4. 进阶在LangGraph智能体工作流中应用结构化输出当你的应用从单一问答升级到多步骤、有状态的智能体工作流时结构化输出的价值会呈指数级放大。LangGraph是一个用于构建有状态、多智能体应用的框架而LlamaIndex提供了与它深度集成的能力。设想一个“市场调研智能体”工作流步骤一搜索根据用户提出的公司名从网络搜索最新新闻。步骤二分析对抓取的新闻内容进行情感分析和关键事件提取结构化输出。步骤三报告将多个分析结果汇总生成一份统一的调研报告另一种结构化输出。在这个工作流中步骤二和步骤三的输出必须是结构化的这样才能被后续的节点步骤作为可靠的、类型明确的输入数据来消费。4.1 定义智能体间的“通信协议”在LangGraph中每个节点的输入和输出通常放在一个共享的“状态”字典里。结构化输出模型就是节点间最好的“通信协议”。from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph, END import operator # 1. 定义工作流的全局状态 class AgentState(TypedDict): company_name: str raw_news_articles: List[str] # 步骤一的输出 news_analyses: List[NewsSummary] # 步骤二的输出结构化 final_report: Optional[MarketResearchReport] # 步骤三的输出另一个结构化模型 errors: List[str] # 2. 定义“分析”节点函数 def analyze_news_node(state: AgentState) - AgentState: 分析原始新闻生成结构化NewsSummary列表 analyses [] for article in state[raw_news_articles]: try: # 使用前面定义的PydanticProgram analysis: NewsSummary news_analysis_program(inputarticle) analyses.append(analysis) except Exception as e: state[errors].append(f分析文章失败{e}) # 可以放入一个空的或标记错误的分析对象 analyses.append(NewsSummary(headline解析失败, key_points[], sentimentneutral, summary)) return {news_analyses: analyses} # 更新状态中的结构化数据 # 3. 定义“报告生成”节点函数 def generate_report_node(state: AgentState) - AgentState: 基于结构化分析列表生成最终报告 if not state[news_analyses]: return {final_report: None} # 我们可以定义另一个Pydantic模型 MarketResearchReport # 它可能包含 company_name, overall_sentiment, risk_factors(List), opportunity_areas(List) 等字段 # 然后创建一个新的 program 来汇总 news_analyses report_input f公司{state[company_name]}\n分析结果{state[news_analyses]} try: report: MarketResearchReport report_generation_program(inputreport_input) return {final_report: report} except Exception as e: state[errors].append(f生成报告失败{e}) return {final_report: None} # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(search_news, search_news_node) # 假设已实现 workflow.add_node(analyze_news, analyze_news_node) workflow.add_node(generate_report, generate_report_node) workflow.set_entry_point(search_news) workflow.add_edge(search_news, analyze_news) workflow.add_edge(analyze_news, generate_report) workflow.add_edge(generate_report, END) app workflow.compile()4.2 结构化输出带来的优势在这个工作流中NewsSummary和MarketResearchReport这两个Pydantic模型起到了关键作用接口清晰analyze_news_node的产出类型是List[NewsSummary]generate_report_node的消费类型也是它。这避免了节点之间传递模糊的文本字符串需要靠“默契”或复杂的解析来理解。错误隔离如果一个新闻分析失败了我们可以捕获异常在NewsSummary对象中标记错误而不会让整个工作流崩溃。后续节点可以通过检查字段来判断数据质量。可测试性你可以轻松地为analyze_news_node编写单元测试用一篇固定的文章断言其输出是否符合NewsSummary的格式和预期的内容。状态可观测在调试时你可以检查state[news_analyses]看到的是一个清晰的对象列表每个对象都有明确的字段而不是一堆杂乱无章的文本。核心经验在基于LangGraph或任何工作流引擎构建复杂AI应用时将每个核心步骤的输入和输出用Pydantic模型进行结构化定义是保证系统可维护、可调试、可扩展的最重要实践。它迫使你明确每个节点的“契约”将不可靠的LLM自由文本输出转化为你代码中可靠的、类型化的数据流。这才是llamaindex rag实战从玩具走向生产系统的关键一步。5. 性能优化与生产级考量将结构化输出用于生产环境除了功能正确我们还需要关注性能、成本和稳定性。5.1 提示词工程精确控制降低成本LLM的令牌Token使用量直接关联成本。复杂的Pydantic模型会产生冗长的JSON Schema增加提示词长度。优化策略1简化模型描述在Field(description...)中使用最精炼的语言。避免冗长的句子用分号分隔要点。# 欠佳 topic: str Field(description这是一个讨论的议题请你从文本中找出大家主要在讨论什么话题并用一个简短的名词性短语概括它) # 更佳 topic: str Field(description讨论议题用简短名词短语概括)优化策略2使用OpenAIPydanticProgram的function_call模式这是默认且推荐的方式。它利用OpenAI API的tools参数比将Schema塞进system或user提示词更高效、更可靠。确保你使用的模型如gpt-4-turbo-preview,gpt-3.5-turbo支持此功能。优化策略3流式输出与部分解析对于生成时间较长的复杂对象可以考虑是否支持流式输出。虽然目前PydanticProgram通常返回完整对象但你可以设计自己的流程先让LLM输出最重要的字段如headline,sentiment再根据需要逐步获取其他字段从而实现快速初步响应。5.2 错误处理与降级方案LLM并非百分之百可靠必须设计容错机制。验证失败处理PydanticProgram在实例化对象时会自动验证。捕获ValidationError并根据错误类型采取行动。from pydantic import ValidationError try: result program(inputtext) except ValidationError as e: logging.warning(fLLM输出验证失败: {e.errors()}) # 降级方案A使用一个包含错误信息的默认对象 result OutputModel(..., error解析失败, raw_llm_outputfallback_text) # 降级方案B触发一次重试并附加更严格的指令 result program_with_stricter_prompt(inputtext)LLM API异常处理网络超时、速率限制、服务不可用等。使用具有重试和回退策略的客户端如tenacity库。from tenacity import retry, stop_after_attempt, wait_random_exponential, retry_if_exception_type import openai retry( stopstop_after_attempt(5), waitwait_random_exponential(multiplier1, max60), retryretry_if_exception_type((openai.APITimeoutError, openai.RateLimitError)) ) def robust_program_call(input_text): return program(inputinput_text)内容安全与过滤如果处理用户生成的或来自不可信源的文本LLM可能被诱导输出有害或不符合格式的内容。除了Pydantic验证还应在业务逻辑层对关键字段如URL、人名进行额外的清洗和过滤。5.3 缓存与版本控制对于相同或相似的输入重复调用LLM生成结构化输出是巨大的浪费。语义缓存使用LlamaIndex的SemanticCache或类似向量缓存方案。当一个新的查询进来时先计算其嵌入向量在缓存中查找语义相似的已有查询及其结构化输出结果。如果找到且相似度超过阈值直接返回缓存的结果无需调用LLM。这能极大降低成本和延迟。模型版本化你的Pydantic模型NewsSummary可能会迭代。为模型添加版本号字段或在数据库存储时记录模型的结构定义Schema。这样当模型变更后你仍然能正确解析历史上缓存的数据或者知道哪些数据需要重新处理。class NewsSummary(BaseModel): model_version: str 1.0.1 # 显式声明版本 headline: str ... # ... 其他字段5.4 评估与监控如何知道你的结构化输出管道工作得好不好人工评估样本定期抽样检查评估提取的准确性、完整性和格式正确性。自动化指标模式符合率成功通过Pydantic验证的比例。关键字段填充率对于必填字段LLM成功提取的比例。与黄金标注的对比对于有标注的数据计算字段级别的精确率、召回率或F1分数。监控与告警监控API调用耗时、Token消耗、验证错误率。当错误率突然上升或耗时异常时触发告警。结构化输出不是一次性的技巧而是一套系统工程。从清晰的数据模型定义到可靠的程序调用再到生产环境的性能、鲁棒性和可观测性建设每一步都需要精心设计。当你把这些都做好LlamaIndex就不再只是一个检索工具而成为了你业务中一个强大的、自动化的“信息结构化工匠”。
