LangChain 输出能力：结构化解析与流式传输实践

目录​编辑一、结构化输出1.1 with_structured_output()1.1.1 返回 Pydantic 对象1.1.2 返回 TypedDict1.1.3 返回JSON1.1.4 选择输出格式1.2 实用场景作为信息提取器二、流式输出2.1 stream()同步传输2.2 astream () 异步传输2.2.1 异步的相关概念一、结构化输出在 LangChain 中聊天模型提供了额外的功能结构化输出。一种使聊天模型以结构化格式例如 JSON进行响应的技术。例如可能希望将模型输出存储在数据库中并确保输出符合数据库模式。这种需求激发了结构化输出的概念其中可以指示模型使用特定的输出结构进行响应。这样做的核心原因是从 “字符串” 到 “对象” 的范式转换。想象一下在没有这个功能之前我们调用聊天模型得到的是一个AIMessage其内容是一个字符串。例如下述伪代码from langchain_deepseek import ChatDeepSeek modelChatDeepSeek(modeldeepseek-chat) ai_messmodel.invoke(你好吗) print(ai_mess.content) #你好呀我很好呢谢谢你关心 能和你聊聊天感觉特别开心你今天过得怎么样呀有没有什么新鲜事#想和我分享或者需要我帮忙的地方我随时在这儿听你说哦✨这个字符串对人类很友好但对程序不友好。如果我们想从这段文本中提取出相关的字段并用在后续逻辑中则需要编写复杂且容易出错的解析代码例如使用正则表达式。聊天模型的with_structured_output方法则允许我们预先定义一个期望的数据结构并要求大模型必须按照这个结构返回信息。1.1 with_structured_output()要想使用结构化输出能力LangChain 提供了一种方法.with_structured_output()该方法需要先定义输出结构然后执行通过.with_structured_output()得到的 Runnable 实例。步骤如下伪代码# 1. 定义输出结构 schema {foo: bar} # 2. 绑定schema其实是生成支持结构化返回的 Runnable 实例 model_with_structure model.with_structured_output(schema) # 3. 执行 structured_output model_with_structure.invoke(user_input)这是获得结构化输出的最简单、最可靠的方法。此方法将 输出结构 作为参数输入返回一个类似 model 的 Runnable。不同之处在于执行 Runnable 后的输出结果输出的不是字符串或消息而是输出与给定输出结构相对应的对象。该输出结构可以指定为 TypedDict 类、JSON Schema 或 Pydantic 类。如果使用 TypedDict 或 JSON Schema则 Runnable 将返回一个字典如果使用 Pydantic 类则将返回一个 Pydantic 对象。以下是该方法的详细定义with_structured_output( schema: dict[str, Any] | type[_BM] | type | None None, *, method: Literal[function_calling, json_mode, json_schema] json_schema, include_raw: bool False, strict: bool | None None, **kwargs: Any, ) - Runnable[PromptValue | str | Sequence[BaseMessage | list[str] | tuple[str, str] | str | dict[str, Any]], dict | _BM]请求参数schema表示输出结构。可以传入为JSON、TypedDict、Pydantic、OpenAI 函数 / 工具method表示 LLM 的生成方法json_schema默认表示使用 DeepSeek 的结构化输出 API。function_calling使用 DeepSeek 的工具调用以前称为函数调用json_mode使用 DeepSeek 的 JSON mode 。请注意如果使用json_mode则必须在模型调用中包含将输出格式化为所需 schema 的说明。include_raw如果为False默认则仅返回解析的结构化输出。如果在模型输出解析过程中发生错误则会引发错误。如果为True则将返回原始模型响应BaseMessage和解析的模型响应。如果在输出解析过程中发生错误它也会被捕获并返回。strict如果为True保证模型输出与 schema 完全匹配。输入 schema 也将根据支持的 schema 进行验证。如果为False输入 schema 将不会被验证模型输出也不会被验证。如果为None默认则不会将 strict 参数传递给模型。tools要绑定到聊天模型的工具列表。要求method 为json_schema、strictTrue、include_rawTrue。则生成的 AIMessage 将在raw中包含工具调kwargs(Any)任何附加参数都直接传递给 bind ()。返回值返回一个 Runnable 实例。将来执行时如果include_rawFalse且 schema 是 Pydantic 类则 Runnable 会输出 Pydantic 对象。否则 Runnable 输出一个字典。如果include_rawTrue则 Runnable 输出一个带有键的字典rawBaseMessageparsed如果出现解析错误则为 None否则类型取决于如上所述的 schema。parsing_errorOptional[BaseException]1.1.1 返回 Pydantic 对象我们可以设置执行 Runnable 后的输出结果指定为 Pydantic 类这将返回一个 Pydantic 对象。 当收到模型的响应后LangChain 会提取出代表 Pydantic 参数的 JSON 对象并用 Pydantic 模型对其进行解析和验证将这个验证后的 JSON 转换为一个可用的 Pydantic 对象实例返回。如下所示from typing import Optional from langchain_deepseek import ChatDeepSeek from pydantic import BaseModel, Field modelChatDeepSeek(modeldeepseek-chat) #这是一个pydantic对象 class Joke(BaseModel): 给用户讲一个笑话 setup:strField(...,description这是笑话的开头) punchline:strField(...,description这是笑话的妙语) rating:Optional[int]Field(defaultNone,description给这个笑话打分) model_with_structuredmodel.with_structured_output(schemaJoke) print(model_with_structured.invoke(给我讲一个小偷的笑话))运行结果setup一个小偷半夜溜进一家人的房子结果发现屋里有个小孩坐在沙发上看着他。 punchline小偷问“你爸妈呢”小孩说“他们去超市买东西了。”小偷说“那正好快告诉我你们家的钱藏在哪儿”小孩说“在爸爸的裤兜里。”小偷问“那爸爸的裤子在哪儿”小孩说“在爸爸腿上他正站在你身后呢。” rating8还支持嵌套输出from typing import Optional, List from langchain_deepseek import ChatDeepSeek from pydantic import BaseModel, Field modelChatDeepSeek(modeldeepseek-chat) #这是一个pydantic对象 class Joke(BaseModel): 给用户讲一个笑话 setup:strField(...,description这是笑话的开头) punchline:strField(...,description这是笑话的妙语) rating:Optional[int]Field(defaultNone,description给这个笑话打分) class OutPut(BaseModel): 这是给用户的总反馈 Jokes:List[Joke]Field(...,description这是所有笑话的汇总) model_with_structuredmodel.with_structured_output(schemaOutPut) print(model_with_structured.invoke(给我讲两个小偷的笑话))运行结果Jokes[Joke(setup一个小偷半夜溜进一户人家结果被主人逮住了。小偷求饶说大哥饶了我吧我家里还有八十岁的老母亲等着我养活呢, punchline主人说那正好我也有八十岁的老母亲不如你替我也养一个吧, rating7), Joke(setup两个小偷去偷银行一个放风一个去撬保险柜。放风的那个突然跑进来说快跑有人来了, punchline撬柜子的说怕什么他还得先打开门锁才能进来呢结果门锁是他们自己撬开的。, ratingNone)]1.1.2 返回 TypedDict先了解一下 TypedDict它用于为字典对象提供精确的、结构化的类型提示。它允许我们指定一个字典中应该有哪些键以及每个键对应的值的类型。最清晰、最常用的定义方式就是类似于定义一个类Python 3.8如下所示from typing import TypedDict class User(TypedDict): name: str age: int email: str is_active: bool True # 默认值这有什么用呢对于它非常重要的一个能力就是捕捉键名拼写错误与类型错误。如user1: User { name: Bob, age: 25, email: bobexample.com } # 类型检查器会捕获这些错误 bad_user: User { name: Dave, age: forty, # 错误应该是int emial: daveexample.com # 错误拼写错误 }因此我们也可以设置执行 Runnable 后的输出结果指定为 TypedDict 类这将返回一个字典且输出后会根据设定进行验证。以下是该方法的使用姿势from typing import Optional, List from langchain_deepseek import ChatDeepSeek from typing_extensions import Annotated,TypedDict modelChatDeepSeek(modeldeepseek-chat) #这是一个pydantic对象 class Joke(TypedDict): 给用户讲一个笑话 setup:Annotated[str,...,这是笑话的开头] punchline:Annotated[str,...,这是笑话的妙语] rating: Annotated[Optional[int], None, 为这个笑话打分] model_with_structuredmodel.with_structured_output(Joke) print(model_with_structured.invoke(给我讲一个小偷的笑话))运行结果{setup: 为什么小偷从不参加拳击比赛, punchline: 因为他们只会偷走胜利而不是用拳头赢回来, rating: 7}让我们加入include_rawTrue再来看看效果model_with_structuredmodel.with_structured_output(Joke,include_rawTrue) print(model_with_structured.invoke(给我讲一个小偷的笑话))输出结果{raw: AIMessage(content, additional_kwargs{refusal: None}, response_metadata{token_usage: {completion_tokens: 98, prompt_tokens: 338, total_tokens: 436, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 256}, prompt_cache_hit_tokens: 256, prompt_cache_miss_tokens: 82}, model_provider: deepseek, model_name: deepseek-v4-flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: fb56e059-74df-4c6c-91a8-727d02a08b81, finish_reason: tool_calls, logprobs: None}, idlc_run--01a03806-dd62-7281-b80e-a1a9e51b2a9f-0, tool_calls[{name: Joke, args: {setup: 一个小偷半夜潜入一家银行打开保险柜发现里面只有一张纸条..., punchline: 纸条上写着先生您来得太晚了我们已经下班了。, rating: 7}, id: call_00_UUDvbOhhKX00ALGrLEDx7210, type: tool_call}], invalid_tool_calls[], usage_metadata{input_tokens: 338, output_tokens: 98, total_tokens: 436, input_token_details: {cache_read: 256}, output_token_details: {}}), parsed: {setup: 一个小偷半夜潜入一家银行打开保险柜发现里面只有一张纸条..., punchline: 纸条上写着先生您来得太晚了我们已经下班了。, rating: 7}, parsing_error: None}其中三个顶层字段rawAIMessage 完整对象AIMessage(content, additional_kwargs{refusal: None}, response_metadata{token_usage: {completion_tokens: 98, prompt_tokens: 338, total_tokens: 436, completion_tokens_details: None, prompt_tokens_details: {audio_tokens: None, cache_write_tokens: None, cached_tokens: 256}, prompt_cache_hit_tokens: 256, prompt_cache_miss_tokens: 82}, model_provider: deepseek, model_name: deepseek‑v4‑flash, system_fingerprint: a26a7955944dc5c60445bff77fac9c8e, id: fb56e059‑74df‑4c6c‑91a8‑727d02a08b81, finish_reason: tool_calls, logprobs: None}, idlc_run‑‑01a03806‑dd62‑7281‑b80e‑a1a9e51b2a9f‑0, tool_calls[{name: Joke, args: {setup: 一个小偷半夜潜入一家银行打开保险柜发现里面只有一张纸条..., punchline: 纸条上写着先生您来得太晚了我们已经下班了。, rating: 7}, id: call_00_UUDvbOhhKX00ALGrLEDx7210, type: tool_call}], invalid_tool_calls[], usage_metadata{input_tokens: 338, output_tokens: 98, total_tokens: 436, input_token_details: {cache_read: 256}, output_token_details: {}})parsed解析后的业务字典{setup: 一个小偷半夜潜入一家银行打开保险柜发现里面只有一张纸条..., punchline: 纸条上写着先生您来得太晚了我们已经下班了。, rating: 7}parsing_errorNone1.1.3 返回JSON还可以让聊天模型直接返回 JSON只不过为了声明 JSON我们需要定义 JSON Schema如下所示from langchain_deepseek import ChatDeepSeek from pydantic.v1.schema import json_scheme modelChatDeepSeek(modeldeepseek-chat) json_scheme{ title: joke, description: 给用户讲一个笑话。, type: object, properties: { setup: { type: string, description: 这个笑话的开头, }, punchline: { type: string, description: 这个笑话的妙语, }, rating: { type: integer, description: 从1到10分给这个笑话评分, default: None, }, }, required: [setup, punchline], } model_with_structuredmodel.with_structured_output(json_scheme) print(model_with_structured.invoke(给我讲一个小偷的笑话))运行结果{setup: 一个小偷闯进一家银行用枪指着出纳员说把钱都交出来, punchline: 出纳员冷静地说对不起先生我们这里是图书馆不是银行。小偷不好意思地说哦那好吧把这本书借给我吧。, rating: 7}1.1.4 选择输出格式创建具有联合类型属性的父模式以使用 Pydantic 为例其他同理代码如下from typing import Optional,Union from langchain_deepseek import ChatDeepSeek from pydantic import BaseModel from pydantic import Field class Joke(BaseModel): 给用户讲一个笑话 setup:strField(...,description这是笑话的开篇) punchline:strField(...,description这是笑话的妙语) rating:Optional[int]Field(defaultNone,description为这个笑话打分) class Common(BaseModel): 给用户返回常规信息 response:strField(...,description这是回复的主要内容) class Structured(BaseModel): 这是所有格式的总汇 final_output:Union[Joke,Common] modelChatDeepSeek(modeldeepseek-chat) model_with_structuredmodel.with_structured_output(Structured) print(model_with_structured.invoke(给我讲一个小偷的笑话)) print(model_with_structured.invoke(你好呀))这里可以根据用户输入的问题自动选择两种结构化输出格式Joke和Common的其中之一运行结果final_outputJoke(setup一个警察抓到一个小偷问他你为什么要偷东西, punchline小偷说警官我没办法啊我爸爸是贼妈妈也是贼所以我从小就一直偷……偷懒, ratingNone) final_outputCommon(response你好呀很高兴见到你有什么我可以帮你的吗)1.2 实用场景作为信息提取器这里我们可以利用大模型结构化输出能力从一段自然语言文本里抽取人物信息直接返回 Pydantic 对象不用手动解析 JSON 字符串。输入文本史密斯是一个头发金黄的白人他的身高是6英尺输出结构化对象name史密斯 hair_color金黄 skin_color白人 height_in_meters1.83自动完成单位换算英尺转米from typing import Optional, List,Union from langchain_core.messages import SystemMessage, HumanMessage from langchain_deepseek import ChatDeepSeek from pydantic import BaseModel,Field class Person_Info(BaseModel): 获取一个人的重要信息 name:Optional[str] Field(...,description这个人的名字) hair_color:Optional[str]Field(...,description如果知道这个人的头发颜色) skin_color:Optional[str]Field(...,description如果知道这个人的肤色) height_in_meters:Optional[str]Field(...,description如果知道这个人的身高以米为单位) modelChatDeepSeek(modeldeepseek-chat) messages[ SystemMessage(你是⼀个提取信息的专家只从⽂本中提取相关信息。如果您不知道要提取的属性的值属性值返回null), HumanMessage(史密斯是一个头发金黄的白人他的身高是6英尺) ] model_with_structuredmodel.with_structured_output(Person_Info) print(model_with_structured.invoke(messages))运行结果name史密斯 hair_color金黄 skin_color白人 height_in_meters1.83with_structured_output会自动把 Pydantic 模型转成 JSON Schema 传给大模型要求模型输出 JSONLangChain 内部自动完成 JSON 解析、Pydantic 实例化开发者拿到直接可用的 Python 对象省去手写 JSON 解析、正则提取的工作。二、流式输出2.1 stream()同步传输在 LangChain 聊天模型中可以使用其.stream()方法来同步生成流式响应的效果。 聊天模型的.stream()方法返回一个迭代器该迭代器在生成输出时同步产生输出消息块。可以使用for循环实时处理每个块。代码如下from langchain_deepseek import ChatDeepSeek modelChatDeepSeek(modeldeepseek-chat) chunks[] for chunk in model.stream(写一个关于湖泊的故事要求100字): chunks.append(chunk) print(chunk.content,end ,flushTrue)运行结果湖水 托 举起 整个 天空 时 老 鱼 在 云 朵 里 游 过 。 树 影 把 涟漪 绣 成 银 线 垂 钓 者 收 竿 离去 留下一 枚 夕阳 当 鱼 饵 。 我 往 水里 丢 石子 涟漪 荡 开 那 人的 倒 影 —— 他说 湖 是 旧 年 留下的 镜子 照 见 所有 错 过的 春天 。 忽然 风 起 镜 面 碎了 千万 道 波 光 追逐 着 逃 向 天 边 。 原来 湖 一直在 收集 我们的 叹息 等 某个 傍晚 还给 天空 。 进程已结束退出代码为 0这里每个消息块的类型都是AIMessageChunk它代表AIMessage的一部分。消息块还可以直接相加来看效果print(\n) chunks_tempchunks[0]chunks[1]chunks[2] print(chunks_temp.content)2.2 astream () 异步传输对于流式传输通常我们可以选择异步调用。先来了解下异步相关知识。2.2.1 异步的相关概念想象一个场景你需要煮一壶水同时还要给朋友发一条短信。我们分别用同步传统和异步两种方式来完成以此对比并引入协程和事件循环的概念。同步阻塞方式做事必须一件一件来import time def boil(): print(开始烧水。。。) time.sleep(5) #烧水耗费5秒时间 print(烧水完成) def send_mess(): print(开始发消息。。。) time.sleep(2) #发消息耗费2秒时间 print(发消息完成) def main(): boil() send_mess() #同步总耗时7秒 main()在boil函数等待的 5 秒里CPU 完全空闲但却不能去做send_message任务效率低下。总耗费7秒的时间。异步方式我们我们可以使用asyncio、协程和事件循环。采用异步的方式充分利用CPU的资源让其在烧水等待的5秒时间内去完成发消息的操作大大提高了效率。什么是协程多进程通常利用的是多核 CPU 的优势同时执行多个计算任务。每个进程有自己独立的内存管理所以不同进程之间要进行数据通信比较麻烦。多线程是在一个 cpu 上创建多个子任务当某一个子任务休息的时候其他任务接着执行。多线程的控制是由 python 自己控制的。线程存在数据同步问题所以要有锁机制。协程的实现是在一个线程内实现的相当于流水线作业。由于线程切换的消耗比较大所以对于并发编程可以优先使用协程。图示进程、线程、协程之间的关系协程作为一种轻量级的并发编程模型可以被视为用户态的 “轻量级线程”。与传统线程相比协程的核心优势在于其调度完全由用户空间掌控避免了操作系统内核的频繁介入从而显著降低了上下文切换的开销。在诸如网络数据刷新、资源加载、用户界面更新、以及 I/O 读写等场景下如果并发任务的计算量相对较小、对系统资源占用较低则不必动用操作系统级别的线程。协程的切换则由程序员和编程语言控制程序员决定在何时暂停或恢复协程。协程是一个特殊的函数它可以在执行过程中暂停并在稍后恢复执行。它用async def定义并在需要暂停的地方使用await。在我们的例子里boil和send_mess就可以用两个协程来进行实现import asyncio async def boil(): print(开始烧水。。。) await asyncio.sleep(5) #等待烧水完成此时可以去执行其他任务 print(烧水完成) async def send_mess(): print(开始发消息。。。) await asyncio.sleep(2) #等待发消息完成 print(发消息完成) # 主程序也是一个协程 async def main(): # 创建两个任务并交给事件循环去调度 task1 asyncio.create_task(boil()) task2 asyncio.create_task(send_mess()) # 等待两个任务都完成 await task1 await task2 # 它负责创建事件循环并将第一个协程主程序放入其中运行。 asyncio.run(main())通过使用asyncio我们可以在单线程中同时处理多个任务。异步一共耗时5秒。总结一下什么是事件循环事件循环是 asyncioPython 标准库中的模块用于编写异步 I/O 操作的代码的核心你可以把它想象成一个总调度员或一个高效的待办事项 (To‑Do List) 管理员。它的工作流程非常简单它维护着一个任务列表比如煮水、发短信。它不断地循环检查每个任务 a. 如果任务处于 “等待 I/O” 状态比如等水开、等网络响应就暂停它立即去执行下一个已经 “就绪” 的任务。 b. 如果任务的等待时间到了或者 I/O 操作完成了事件循环就恢复执行这个任务。一个在单线程内调度和管理所有协程的核心机制就是事件循环。它不停地检查哪些协程可以执行哪些在等待。