尽管LangChain的基本输出解析器能处理简单的字符串格式或基本JSON,但生产应用经常遇到更复杂、较难预测或偶尔格式错误的大语言模型(LLM)输出。有效提取结构化信息、处理变化和从错误中恢复,需要更高级的解析策略。本文介绍处理LLM响应的方法,确保为后续任务可靠地提取数据。尽管进行了提示工程,LLM并不总是完美遵循请求的输出格式。它们可能会添加解释性文本、遗漏必填字段,或生成语法不正确的结构(例如格式错误的JSON)。仅仅依赖基本字符串分割或标准JSON解析可能导致脆弱的应用,从而意外失败。运用Pydantic进行结构化验证一种有效方法是使用Pydantic模型定义期望的输出结构。Pydantic使用Python类型注解提供数据验证和设置管理。LangChain通过 PydanticOutputParser 与Pydantic集成。你定义一个Pydantic模型,代表你期望LLM返回的模式。解析器随后自动生成包含在提示中的格式说明,指导LLM形成正确的结构。更重要的是,它使用Pydantic模型解析LLM的字符串输出,验证数据类型,检查必填字段,并将原始文本转换为结构化Python对象。from pydantic import BaseModel, Field from typing import List, Optional from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import PromptTemplate from langchain_openai import ChatOpenAI # 或任何其他兼容的LLM # 定义你期望的数据结构 class ProductReview(BaseModel): product_name: str = Field(description="被评论产品的名称。") rating: int = Field(description="1到5的评分。", ge=1, le=5) summary: str = Field(description="评论的简短总结。") pros: Optional[List[str]] = Field(description="可选的优点列表。") cons: Optional[List[str]] = Field(description="可选的缺点列表。") # 设置解析器 parser = PydanticOutputParser(pydantic_object=ProductReview) # 定义包含格式说明的提示模板 prompt_template = """ 分析以下产品评论并提取重要信息。 {format_instructions} 评论文本: {review_text} """ prompt = PromptTemplate( template=prompt_template, input_variables=["review_text"], partial_variables={"format_instructions": parser.get_format_instructions()} ) # 示例用法 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) chain = prompt | llm | parser review = "This laptop is amazing! Super fast (5/5), great screen. Only downside is the battery life could be better." structured_output = chain.invoke({"review_text": review}) print(structured_output) print(f"Rating: {structured_output.rating}")使用 PydanticOutputParser 不仅提供解析,还提供基于你的模型定义的自动验证(例如,确保 rating 在1到5之间)。如果输出不符合规范,Pydantic会引发验证错误,这通常比一般的解析失败提供更多信息。实现解析失败的重试逻辑有时,LLM生成的输出几乎正确,但首次解析失败(例如,JSON中缺少逗号,或多余的注释)。而不是立即失败,你可以实现重试逻辑。LangChain提供 RetryOutputParser 和更精巧的 RetryWithErrorOutputParser。这些解析器封装一个主解析器(例如 PydanticOutputParser 或 SimpleJsonOutputParser)。如果主解析器失败,重试解析器会捕获异常。它随后格式化一个新的提示,包含原始提示、错误输出和错误信息。这个新提示指示LLM根据错误修正其先前的输出。这个重试循环可以显著提高从不太可靠的LLM响应中提取结构化数据的成功率。from langchain_core.output_parsers import PydanticOutputParser, RetryWithErrorOutputParser from langchain_core.prompts import PromptTemplate from langchain_core.prompt_values import StringPromptValue from langchain_openai import ChatOpenAI # 或任何其他兼容的LLM # 假设使用前面示例中的ProductReview Pydantic模型 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) main_parser = PydanticOutputParser(pydantic_object=ProductReview) # 用重试机制封装主解析器 retry_parser = RetryWithErrorOutputParser.from_llm( parser=main_parser, llm=llm, # 用于尝试修正的LLM max_retries=2 # 重试次数 ) # 提示模板最初保持不变 prompt_template = """ 分析以下产品评论并提取重要信息。 {format_instructions} 评论文本: {review_text} """ prompt = PromptTemplate( template=prompt_template, input_variables=["review_text"], partial_variables={"format_instructions": main_parser.get_format_instructions()} # 仍使用主解析器的说明 ) review = "Decent phone (4 stars). Good camera. Wish it had more storage." prompt_value = prompt.format_prompt(review_text=review) llm_output = llm.invoke(prompt_value) try: structured_output = retry_parser.parse_with_prompt( llm_output.content, prompt_value ) print("成功解析:") print(structured_output) except Exception as e: print(f"重试后解析失败:{e}")RetryWithErrorOutputParser 智能地使用失败解析尝试中的错误信息来指导LLM修正,使其比简单地要求LLM“再试一次”更有效。运用结构化输出能力(工具调用)现代LLM(例如OpenAI的GPT系列、Anthropic的Claude 3、Google的Gemini)通过工具调用功能原生支持结构化数据生成。无需对字符串输出进行提示工程,你只需提供一个模式,模型便会生成与该模式匹配的参数。LangChain通过 .with_structured_output() 方法简化了这一点。此方法接受Pydantic模型或JSON Schema,并自动管理与模型工具调用API的交互。它直接返回一个经过验证的对象,消除了单独解析步骤或手动参数提取的需要。from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field # 定义 Pydantic 模型 class WeatherInfo(BaseModel): """特定位置的天气信息。""" location: str = Field(description="城市和州,例如:旧金山,加利福尼亚州") temperature: int = Field(description="当前华氏温度") forecast: str = Field(description="简短的天气预报(例如:晴朗、多云)") # 初始化 LLM llm = ChatOpenAI(model="gpt-4o", temperature=0) # 配置 LLM 直接返回结构化对象 structured_llm = llm.with_structured_output(WeatherInfo) # 示例调用 # response = structured_llm.invoke("What's the weather like in Boston today?") # print(response) # 输出将是 WeatherInfo 的一个实例: # WeatherInfo(location='Boston, MA', temperature=72, forecast='Partly Cloudy')可用时,使用LLM的原生结构化输出能力通常是最可靠的方法。它减少了解析仅仅“尝试”遵循格式说明的自由格式文本时存在的歧义。组合解析器和自定义逻辑对于非常复杂的情况,你可能需要链接多个解析器,甚至通过继承LangChain的 BaseOutputParser 来实现自定义解析逻辑。例如,LLM可能会生成一个包含文本摘要和JSON块的响应。你可以使用自定义解析器首先使用正则表达式或字符串操作提取JSON块,然后将该块输入到 PydanticOutputParser 或 JsonOutputParser 中。开发一个自定义解析器涉及实现 parse 方法(以及可能的 get_format_instructions 方法)。这提供了最大灵活性,但需要仔细设计和测试。from langchain_core.output_parsers import BaseOutputParser from typing import Any import re import json class CustomJsonExtractorParser(BaseOutputParser): """ 一个自定义解析器,用于提取并解析封装在 ```json ... ``` 中的JSON块。 """ def parse(self, text: str) -> Any: match = re.search(r"```json\s*(.*?)\s*```", text, re.DOTALL) if not match: raise ValueError(f"Could not find JSON block in text: {text}") json_str = match.group(1) try: return json.loads(json_str) except json.JSONDecodeError as e: raise ValueError(f"Failed to parse JSON: {e}\nJSON string: {json_str}") def get_format_instructions(self) -> str: return "请将JSON输出封装在 ```json\n ... \n``` 标签内。" # 示例用法 custom_parser = CustomJsonExtractorParser() llm_output = "这是分析结果:\n```json\n{\n \"key\": \"value\",\n \"number\": 123\n}\n```\n如果需要更多详情,请告诉我。" parsed_data = custom_parser.parse(llm_output) print(parsed_data)处理不可恢复的错误即使有重试机制和可靠的解析器,有些LLM输出仍可能无法解析或验证失败。生产系统必须优雅地处理这些情况。策略包括:日志记录: 记录错误输出和解析错误,以供后续分析,并可能进行微调或提示调整。默认值/回退: 返回默认对象或 None,以允许应用流程继续,并可能将结果标记为不确定。错误传播: 将错误沿着调用堆栈向上级传播,由更高级别的应用逻辑处理。人工干预: 对于关键应用,将失败的解析尝试路由到人工审查队列。选择正确的策略取决于应用对错误的容忍度以及提取数据的重要性。通过采用这些高级解析策略,运用Pydantic进行结构和验证,实现重试逻辑,运用原生结构化输出功能,并为不可恢复的错误做好准备,你可以构建更具韧性和可靠的LangChain应用,能够处理生产环境中LLM输出固有的可变性。这些方法是本课程中讨论的复杂链和代理的基本构成要素。