《AI大模型应用开发实战从入门到精通共60篇》009、LangChain之Model I/O：模型调用与输出解析

LangChain之Model I/O：模型调用与输出解析

从一次诡异的JSON解析失败说起

上周三凌晨两点，我被值班电话叫醒——生产环境上一个基于GPT-4的文档摘要服务突然大面积报错。日志里躺着一行让我血压飙升的错误：json.decoder.JSONDecodeError: Expecting property name enclosed in double quotes。

我盯着屏幕看了五分钟，脑子里飞速过了一遍最近的上线记录。没改过prompt，没动过模型参数，甚至连LangChain版本都没升。那问题出在哪？

后来抓包看了原始响应，真相让我哭笑不得：模型返回的JSON里，某个字段的值被截断了，最后多了一个逗号——{"title": "报告", "content": "这是一份关于...",}。多了一个逗号，整个JSON就炸了。

这就是Model I/O里最典型的“模型输出不可控”问题。你永远不知道大模型会在哪个细节上给你“惊喜”。今天这篇笔记，就聊聊LangChain里模型调用和输出解析的那些坑，以及怎么填。

模型调用：别被“一行代码”骗了

很多人觉得LangChain调用模型就是llm.invoke("你好")这么简单。对，demo确实这么写，但生产环境这么写，等着你的就是各种玄学问题。

基础调用其实有门道

fromlangchain_openaiimportChatOpenAI# 别这样写——硬编码API Key，代码一提交就泄露# llm = ChatOpenAI(api_key="sk-xxx")# 正确姿势：从环境变量读取importos llm=ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY"),model="gpt-4-turbo",temperature=0.1,# 这里踩过坑：摘要任务temperature设太高，每次输出都不一样max_tokens=4096,# 别偷懒不设这个，默认值可能不够用timeout=30,# 加个超时，防止模型卡死拖垮整个服务)

temperature这个参数我吃过亏。做文档摘要时，我图省事用了默认值0.7，结果同一个文档每次摘要的关键点都不一样，测试用例根本没法写。后来改成0.1，输出稳定多了。记住：需要确定性输出的任务，temperature往低了调。

流式调用：用户体验的救星

用户等模型完整输出再看到结果，体验极差。流式调用是必选项：

# 流式输出，逐token返回forchunkinllm.stream("讲个冷笑话"):print(chunk.content,end="",flush=True)

这里有个坑：流式返回的是AIMessageChunk对象，不是字符串。如果你直接print(chunk)，会看到一堆元数据。记得取.content。

异步调用：别阻塞你的Web服务

如果你的应用是FastAPI这类异步框架，同步调用会阻塞事件循环。正确做法：

importasynciofromlangchain_openaiimportChatOpenAI llm=ChatOpenAI(model="gpt-4-turbo")asyncdefasync_chat():# 异步调用，不阻塞主线程response=awaitllm.ainvoke("写一首诗")returnresponse.content# 多个请求并发处理asyncdefbatch_chat(prompts):tasks=[llm.ainvoke(p)forpinprompts]results=awaitasyncio.gather(*tasks)return[r.contentforrinresults]

asyncio.gather这个用法我踩过坑——如果某个请求超时，整个gather都会抛异常。建议加个return_exceptions=True参数，单独处理失败的任务。

输出解析：从“信任模型”到“校验模型”

回到开头的JSON解析问题。模型输出天然不可靠，你不能假设它永远按格式返回。LangChain提供了几种解析器，但用起来都有讲究。

PydanticOutputParser：结构化输出的利器

fromlangchain.output_parsersimportPydanticOutputParserfrompydanticimportBaseModel,FieldclassDocumentSummary(BaseModel):title:str=Field(description="文档标题，不超过20字")key_points:list[str]=Field(description="关键要点列表，3-5个")summary:str=Field(description="摘要内容，不超过200字")parser=PydanticOutputParser(pydantic_object=DocumentSummary)# 构造prompt时，一定要把格式说明加进去prompt=PromptTemplate(template="请分析以下文档并返回结构化结果。\n{format_instructions}\n文档内容：{document}",input_variables=["document"],partial_variables={"format_instructions":parser.get_format_instructions()})

get_format_instructions()会自动生成一段格式说明，告诉模型怎么输出。但这里有个坑：模型可能不遵守格式说明。我遇到过模型返回了正确的JSON结构，但字段名拼写错误——"key_points"写成了"keypoint"。解析器直接报错。

容错解析：给模型一点“改错”的机会

fromlangchain.output_parsersimportOutputFixingParserfromlangchain_openaiimportChatOpenAI# 用另一个模型来修复解析错误fixing_parser=OutputFixingParser.from_llm(parser=parser,llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0))try:result=fixing_parser.parse(raw_output)exceptExceptionase:# 如果修复也失败了，记录日志并返回默认值logger.error(f"解析失败，原始输出：{raw_output}",exc_info=e)result=DocumentSummary(title="解析失败",key_points=[],summary="模型输出格式异常，请重试")

OutputFixingParser的原理是：当解析失败时，把错误信息和原始输出一起发给另一个模型，让它尝试修复。但注意，这增加了额外的一次API调用和延迟。高并发场景下慎用。

自定义解析器：处理非标准输出

有时候模型输出既不是JSON也不是Pydantic对象，比如一段带标记的文本：

标题：XXX 要点1：... 要点2：... 摘要：...

这时候自己写解析器更灵活：

fromlangchain.schemaimportBaseOutputParserclassMarkdownSummaryParser(BaseOutputParser):defparse(self,text:str):lines=text.strip().split("\n")result={}current_key=Nonecurrent_value=[]forlineinlines:if"："inlineor":"inline:# 保存上一个字段ifcurrent_key:result[current_key]="\n".join(current_value).strip()# 新字段开始key,value=line.split("：",1)if"："inlineelseline.split(":",1)current_key=key.strip()current_value=[value.strip()]else:# 多行内容追加ifcurrent_key:current_value.append(line.strip())# 保存最后一个字段ifcurrent_key:result[current_key]="\n".join(current_value).strip()returnresultdefget_format_instructions(self):return"请按以下格式输出：\n标题：...\n要点1：...\n要点2：...\n摘要：..."

这个解析器处理了多行内容的情况，比简单的按行分割靠谱。但依然有边界情况——如果模型输出的标题里包含冒号，解析就会出错。所以永远给解析器加一个fallback逻辑。

实战中的那些“坑”

坑1：Token限制导致的截断

模型输出被截断是最常见的问题。你设了max_tokens=4096，但模型生成到一半被截断，返回的JSON不完整。

解决方案：在prompt里明确要求模型在输出结束时加一个特殊标记，比如[END]。解析时检查这个标记是否存在，不存在就说明被截断了，触发重试或降级处理。

坑2：模型“幻觉”字段

模型可能输出你根本没要求的字段。比如你只要求title和content，它自作主张加了个author。Pydantic解析器默认会忽略未定义的字段，但如果你设置了extra="forbid"，就会报错。

建议：解析器设置extra="ignore"，只提取你需要的字段，忽略多余的。

坑3：编码问题

中文环境下，模型可能返回全角字符的冒号、逗号。JSON解析器只认半角符号。我写了个预处理函数：

defnormalize_output(text:str)->str:# 全角转半角text=text.replace("：",":").replace("，",",").replace("“","\"").replace("”","\"")# 去除BOM头text=text.lstrip("\ufeff")returntext

坑4：重试策略

模型调用失败是常态。别用简单的指数退避，我推荐分级重试：

• 第一次失败：立即重试
• 第二次失败：等待1秒后重试
• 第三次失败：等待5秒后重试，同时切换模型（比如从gpt-4降到gpt-3.5-turbo）
• 第四次失败：记录详细日志，返回默认结果，触发告警

个人经验总结

1. 永远不要信任模型输出。无论你prompt写得多么详细，模型总有办法给你“惊喜”。输出解析不是锦上添花，是必选项。

2. 解析器要分层。第一层做格式校验（JSON是否合法），第二层做内容校验（字段是否存在、类型是否正确），第三层做业务校验（比如摘要长度是否合理）。每一层失败都有对应的处理逻辑。

3. 日志要打全。每次模型调用，记录完整的输入prompt、原始输出、解析结果、解析耗时。线上问题排查时，这些日志就是你的救命稻草。

4. 给用户留退路。如果解析失败，别直接抛500错误。返回一个“系统正在处理，请稍后重试”的友好提示，或者展示模型原始输出（虽然丑，但能用）。

5. 测试用例要覆盖边界。空字符串、超长文本、特殊字符、全角符号、JSON注释（模型真会输出注释）——这些都要写测试。我见过最离谱的是模型在JSON里加了C语言风格的/* */注释。

Model I/O是整个LangChain应用的入口和出口，这里出问题，后面所有逻辑都是白搭。花时间把这一层打磨好，比追求花哨的Chain和Agent实在得多。