核心组件2

六 流式传输 (Streaming)

需要注意的是：

流式输出依赖于整个程序链路都支持“逐块处理”。如果程序中的某个环节必须等待完整输出（如需一次性写入数据库），则无法直接使用 Streaming；

LangChain 1.0 进一步优化了流式机制，引入 自动流式模式（Auto-streaming）。例如在Agent中，如果整体程序处于 streaming 模式，即便节点中调用 model.invoke()，LangChain 也会自动流式化模型调用。

# 使用.stream()方法进行流式传输forchunkinmodel.stream("用一段话描述大海。"):print(chunk.content,end="",flush=True)# 逐块打印

astream_events()
此外，LangChain 还支持通过 astream_events() 对语义事件进行异步流式监听，适合需要过滤不同事件类型的复杂场景。

你能看到 完整语义生命周期事件,包括：

on_chain_start

on_prompt_start / on_prompt_end

on_llm_start

on_llm_stream（逐 Token）

on_llm_end

on_chain_end

非常适合：

调试 LLM 推理过程

了解 LangChain pipeline 的执行顺序

构建 UI（如 web 前端的逐 token streaming）

实现日志、可观测性、监控系统

importasynciofromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate# 1. 构建最简单的 Prompt + LLMprompt=ChatPromptTemplate.from_messages([("system","你是一个专业的 AI 助手。"),("human","{question}")])# 2. 初始化 ChatOpenAI 实例，指定使用 gpt-3.5-turbo 模型llm=ChatOpenAI(model="gpt-3.5-turbo",)# 3. 使用管道符将 prompt 模板与 llm 连接，构建可运行的链chain=prompt|llm# 4. 使用 astream_events() 监听所有语义事件events=chain.astream_events({"question":"请用一句话介绍一下 LangChain 1.0 的核心思想。"},version="v1",# 必须指明版本，v1 才有语义事件)asyncforeventinevents:# 打印事件类型print(f"""[Event] type={event["event"]}""")# 展示关键字段if"data"inevent:print(" data:",event["data"])print("-----------------------------")

七 结构化输出解析

• 很多时候，我们需要模型返回结构化的数据（如JSON），以便程序后续处理。输出解析器 (Output Parsers) 正是为此而生。

• 最强大的是 StructuredOutputParser，它可以与 Zod（TypeScript）或 Pydantic（Python）等模式定义工具结合使用，确保输出符合预定格式。

  • 目标：让大模型返回可程序解析的数据

  • 任务：学习Pydantic模型，使用with_structured_output()

  • 产出：一个信息抽取器（提取电影信息/新闻摘要）

  • 关键点：ToolStrategy兼容所有模型，ProviderStrategy更可靠

with_structured_output()

使用 Pydantic 的 BaseModel 定义一个严格的数据结构。每个字段都明确了类型（如 str、int、float），并用 Field(…, description=“…”) 提供语义描述。据此，模型回复时，LangChain 会要求 LLM 的输出必须能填充这些字段。然后使用with_structured_output即可引导模型进行结构化输出。

fromtypingimportListfromlangchain_core.utils.pydanticimportBaseModel,Fieldfromlangchain_openaiimportChatOpenAI# 1. 定义期望的输出结构 (Pydantic 模型)classPerson(BaseModel):"""Information about a person."""name:str=Field(description="人的姓名")age:int=Field(description="人的年龄")high:int=Field(description="人的身高")hobbies:List[str]=Field(description="人的爱好列表")# 2. 初始化模型并绑定结构化输出格式llm=ChatOpenAI(model="gpt-4o",temperature=0)structured_llm=llm.with_structured_output(Person)# 3. 调用模型并获取 Pydantic 对象，构造提示：要求提取约翰·多伊的姓名、年龄和兴趣爱好prompt="提取名为约翰·多伊的人的信息，提取不到的数据就为空值。他30岁，喜欢阅读、远足和弹吉他."result=structured_llm.invoke(prompt)# 4. 验证结果print(f"Type of result:{type(result)}")print(f"Result object:{result}")# 5.判断result是否属于Person类assertisinstance(result,Person)

Type of result: <class '__main__.Person'> Result object: name='约翰·多伊' age=30 high=0 hobbies=['阅读', '远足', '弹吉他']

• 而如果想要获得模型的完整回复，则可以设置include_raw=True

# 1. 配置结构化输出：指定返回 Pydantic 模型 Person，并保留原始响应structured_llm=llm.with_structured_output(Person,include_raw=True)

agent中结构化输出

frompydanticimportBaseModel,Field,field_validatorfromtypingimportLiteralfromlangchain.agentsimportcreate_agentfromlangchain_openaiimportChatOpenAI# 1. 定义天气结构化输出模型classWeatherForecast(BaseModel):"""天气预报结构化输出"""city:str=Field(description="城市名称")temperature:int=Field(description="温度(摄氏度)")condition:Literal["晴","雨","多云","雪"]=Field(description="天气状况")# 2. 加载模型model=ChatOpenAI(model="deepseek-ai/DeepSeek-V3.2",temperature=0.5,# 温度参数，用于控制模型的随机性，值越小则随机性越小max_tokens=512,# 最大生成token数timeout=30,# 超时时间，单位秒base_url='https://api.siliconflow.cn/v1',api_key='sk-xxx')# 3. 创建智能体agent=create_agent(model=model,# 加载的模型tools=[],# 工具列表，这里为空response_format=WeatherForecast# 指定结构化输出格式)# 4. 调用智能体解析天气描述result=agent.invoke({"messages":[{"role":"user","content":"北京今天阳光明媚，温度10度"}]})# 5. 提取并打印结果forecast=result["structured_response"]print(f"{forecast.city}天气:{forecast.condition},{forecast.temperature}°C")

北京天气: 晴, 10°C

带判断的结构

frompydanticimportBaseModelfromlangchain_openaiimportChatOpenAI# 1. 定义年龄模型，限制范围 0-150classAgeProfile(BaseModel):name:strage:int=Field(ge=0,le=150)# 年龄必须在0-150之间# 2. 定义模型model=ChatOpenAI(model="deepseek-ai/DeepSeek-V3.2",temperature=0.5,# 温度参数，用于控制模型的随机性，值越小则随机性越小max_tokens=512,# 最大生成token数timeout=30,# 超时时间，单位秒base_url='https://api.siliconflow.cn/v1',api_key='sk-xxx')# 3. 创建智能体agentagent=create_agent(model=model,tools=[],response_format=AgeProfile)# 4. 模型返回age=999（非法值）result=agent.invoke({"messages":[{"role":"user","content":"张三的年龄是999岁"# 明显不合理的数据}]})# LangChain会自动：# 1. 捕获ValidationError# 2. 在ToolMessage中反馈错误详情# 3. 让模型重新生成# 最终返回合法值print(result["structured_response"])

name='张三' age=150

JsonOutputParser

fromlangchain_core.output_parsersimportJsonOutputParserimportjsonfrompydanticimportBaseModel,Fieldfromlangchain_openaiimportChatOpenAI# 1. 定义输出结构classWeatherInfo(BaseModel):"""天气信息"""city:str=Field(description="城市名称")temperature:int=Field(description="温度（摄氏度）")condition:str=Field(description="天气状况")# 2. 创建 JSON 输出解析器json_parser=JsonOutputParser(pydantic_object=WeatherInfo)# 3. 创建提示模板（关键：必须包含 "json" 这个词）prompt=ChatPromptTemplate.from_template("""请根据以下信息提取天气数据，并以 JSON 格式返回。 信息：{weather_info} 请返回包含以下字段的 JSON： - city: 城市名称 - temperature: 温度（摄氏度） - condition: 天气状况 必须返回以下 JSON 格式（不要包含任何其他文本）： {{"city": "城市名称", "temperature": 温度数字, "condition": "天气状况"}} 例如：{{"city": "北京", "temperature": 25, "condition": "晴"}} JSON 格式： """)# 4. 定义模型model=ChatOpenAI(model="deepseek-ai/DeepSeek-V3.2",temperature=0.5,# 温度参数，用于控制模型的随机性，值越小则随机性越小max_tokens=512,# 最大生成token数timeout=30,# 超时时间，单位秒base_url='https://api.siliconflow.cn/v1',api_key='sk-xxx')# 5. 构建链runnable=prompt|model|json_parser# 6. 调用result=runnable.invoke({"weather_info":"北京今天晴，温度25度"})print(result)print(result["city"])

{'city': '北京', 'temperature': 25, 'condition': '晴'} 北京

结构化输出关键要点：

1. 输出json格式提示词必须包含 “json” 关键词

   • 有些大模型 要求提示词中包含 “json” 这个词，
   • 否则会报错：Prompt must contain the word 'json'

2. 推荐方案对比

   • 方案 1 (JsonOutputParser)：最简洁，推荐使用
   • 方案 2 (with_structured_output)：需要提示词包含 “json”
   • 方案 3 (可选手动 JSON 解析)：最稳定，适合关键应用

3. 配置建议

   • 设置temperature=0.0获得更稳定的输出
   • 最好提供清晰的 JSON 格式示例

4. 常见错误

   • 提示词中没有 “json” 关键词
   • 没有设置低温度参数
   • 没有提供 JSON 格式示例
   • 没有处理解析异常

简单问答机器人

用标准接口实现一个支持多轮消息与流式输出的问答机器人

fromlangchain_openaiimportChatOpenAIfromlangchain.messagesimportHumanMessage,SystemMessage,AIMessagefromlangchain_core.promptsimportChatPromptTemplate model=ChatOpenAI(model="deepseek-ai/DeepSeek-V3.2",temperature=0.5,# 温度参数，用于控制模型的随机性，值越小则随机性越小max_tokens=512,# 最大生成token数timeout=30,# 超时时间，单位秒base_url='https://api.siliconflow.cn/v1',api_key='sk-xxxx')defchat():msgs=[SystemMessage(content="你是一个智能助手"),]whileTrue:prompt=input("请输入:")ifprompt=="exit":breakhuman=HumanMessage(content=prompt)msgs.append(human)# ret=model.invoke(msgs)# msgs.append(AIMessage(content=ret.content))# print(ret.content)full_reply=""forchunkinmodel.stream(msgs):print(chunk.content,end="",flush=True)full_reply+=chunk.content msgs.append(AIMessage(content=full_reply))print("\n"+"-"*40)# 分隔线if__name__=='__main__':chat()