王炸！LangChain 1.0重磅更新！5分钟搞定一个流式聊天机器人，新手也能起飞！

为了让初学者更容易上手，在文章末尾，给出一个综合应用样例：**搭建流式响应的多轮问答机器人**，它结合了消息格式和流式响应，模拟真实聊天场景。

Python_金钱豹

392人浏览 · 2025-11-20 11:54:15

Python_金钱豹 · 2025-11-20 11:54:15 发布

为了让初学者更容易上手，在文章末尾，给出一个综合应用样例：搭建流式响应的多轮问答机器人，它结合了消息格式和流式响应，模拟真实聊天场景。

1. LangChain 1.0 的基础模型消息格式

在LangChain 1.0中，消息（Message）是模型输入和输出的核心构建块。它提供了一个跨模型统一的Message标准，无论你使用OpenAI、Anthropic、Gemini还是本地模型，都能保持一致的行为。这种统一抽象带来了三大好处：

好处	解释
兼容性强	不同模型的消息格式自动对齐，无需编写提供商特定的代码。
可扩展性高	支持多模态内容（如文本、图像、音频）、工具调用和自定义元数据，便于扩展新数据类型。
可追踪性好	通过元数据（如ID、token使用量、响应信息）为LangSmith等调试工具提供一致的上下文结构。

每个消息包含角色（role）（如system、user、assistant、tool）、内容（content）（文本或其他）和可选的元数据（metadata）。核心消息类型包括：

SystemMessage：设置助手的角色或行为。
HumanMessage：用户输入。
AIMessage：模型输出。

1.样例代码：构建消息并调用模型

下面是一个简单示例：创建一个系统消息定义助手角色，然后添加用户消息，调用模型生成响应。

from langchain_openai import ChatOpenAI  # 导入OpenAI聊天模型from langchain_core.messages import SystemMessage, HumanMessage, AIMessage# 初始化模型（替换为你的API密钥）model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)# 系统消息：定义助手角色system_msg = SystemMessage(content="你叫小智，是一名助人为乐的助手。")# 用户消息：用户输入human_msg = HumanMessage(content="你好，好久不见，请介绍下你自己。")# 消息列表：按顺序传递给模型messages = [system_msg, human_msg]# 调用模型response = model.invoke(messages)print(response)  # 输出：AIMessage(content='你好！我是小智，一名乐于助人的AI助手...', ...)

2.响应结果解释

模型的invoke()方法返回一个AIMessage对象，包含生成的文本和其他元数据。以下表格解释了其关键属性（基于官方文档）：

属性	类型	描述
`content`	str 或 list[dict]	原始内容负载（字符串或提供商原生块）。
`text`	str	人类可读的文本内容（简化版content）。
`content_blocks`	list[ContentBlock]	内容块的标准化视图（如推理块、图像）。
`tool_calls`	list[dict] 或 None	模型调用的工具（名称、参数、ID）。
`id`	str	唯一标识符（自动生成或来自提供商）。
`usage_metadata`	dict 或 None	Token使用量（如input_tokens、output_tokens）。
`response_metadata`	ResponseMetadata 或 None	提供商特定响应数据（如完成原因）。

运行上述代码后，你会看到类似输出：

AIMessage(content='你好！我是小智，一名乐于助人的AI助手。我可以帮你解答问题、生成代码或聊天。有什么我能帮忙的吗？', id='run-abc123', usage_metadata={'input_tokens': 25, 'output_tokens': 50, ...})

这让调试和追踪变得简单——例如，通过response.usage_metadata监控token消耗。

3.常见问题排查

错误：No API key provided：确保设置了OPENAI_API_KEY环境变量（export OPENAI_API_KEY=your_key）。
消息顺序混乱：始终按[system, human, ai, …]顺序构建列表，模型会记住历史。
多模态扩展：想加图像？用HumanMessage(content=[{"type": "text", "text": "描述"}, {"type": "image_url", "image_url": {"url": "img.jpg"}}])。

2. 模型的流式响应

流式响应（Streaming）是提升用户体验的关键：在LangChain 1.0中，通过stream()方法，模型不会一次性返回完整结果，而是返回一个迭代器（iterator）。每次迭代生成一个AIMessageChunk，包含部分内容（如单个token或文本块）。这允许实时打印输出，减少感知延迟，非常适合聊天界面。

启用流式需要设置streaming=True（在模型初始化时）。迭代器会逐步yield chunk，你可以处理每个chunk的content属性。

1.样例代码：实时打印流式输出

以下示例使用简单用户输入，逐步打印模型生成的token（以“|”分隔，模拟实时流）。

from langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage# 初始化模型，启用流式model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, streaming=True)# 用户消息messages = [HumanMessage(content="你好，好久不见。讲个关于AI的笑话吧。")]# 流式调用：返回迭代器for chunk in model.stream(messages):    if chunk.content:  # 确保chunk有内容        print(chunk.content, end="|", flush=True)  # 实时打印，无换行print("\n")  # 结束时换行

2.运行结果示例

运行后，你会看到类似实时输出：

为什么AI不怕冷？因为它有许多层！|哈哈，是不是很智能？|

每个chunk是一个AIMessageChunk对象，chunk.content是增量文本。flush=True确保立即刷新到控制台，而end="|"添加分隔符模拟流式效果。

提示：如果使用stream_mode="messages"（高级用法），chunk将代表完整消息块。但对于初学者，上述简单循环就够用。流式特别适合Web应用，能让用户看到“打字机”效果。

3.常见问题排查

无输出或卡住：检查streaming=True是否设置；如果chunk为空，可能是模型不支持或网络问题。
性能瓶颈：流式会增加少量开销；生产中，用asyncio异步处理以支持并发。
历史消息：流式也支持消息列表，确保传入完整历史以保持上下文。

3. 模型的结构化输出方法

结构化输出让模型返回可预测的格式（如JSON对象或Pydantic模型），避免解析自然语言的麻烦。在LangChain 1.0中，支持多种Schema：

Pydantic：最丰富，支持字段校验、描述、默认值和嵌套结构。生产首选。
TypedDict：轻量类型约束，适合简单场景。
JSON Schema：通用，与前后端/跨语言接口兼容。

这些通过with_structured_output()方法绑定到模型，确保输出符合Schema。Pydantic提供验证（如类型检查、范围约束），如果输出无效会自动重试。

1.样例代码：使用Pydantic生成电影信息

以下示例定义一个电影Schema，绑定到模型，然后输入提示，输出结构化对象。

from pydantic import BaseModel, Fieldfrom langchain_openai import ChatOpenAI# 定义Pydantic模型：电影详情class Movie(BaseModel):    """A movie with details."""    title: str = Field(..., description="The title of the movie")    year: int = Field(..., description="The year the movie was released")    director: str = Field(..., description="The director of the movie")    rating: float = Field(..., description="The movie's rating out of 10")# 初始化模型model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)# 绑定结构化输出：返回Movie实例model_with_structure = model.with_structured_output(Movie)# 输入提示response = model_with_structure.invoke("Provide details about the movie Inception")print(response)  # 输出：Movie(title='Inception', year=2010, director='Christopher Nolan', rating=8.8)

2.解释与好处

绑定过程：with_structured_output(Movie)创建了一个新模型变体，内部使用工具调用或提供商原生功能强制Schema。
输出：response直接是Movie实例（Pydantic对象），便于访问属性如response.title。无需手动解析JSON！
好处：减少错误（自动校验）、提高可靠性（模型“知道”输出格式）、易集成（直接序列化为JSON）。

扩展提示：对于TypedDict，使用typing.TypedDict定义；对于JSON Schema，直接传dict。生产中，结合错误处理（如method_kwargs={"handle_errors": True}）以重试无效输出。

3.常见问题排查

校验失败：模型可能生成不匹配数据；用include_raw=True保留原始输出调试。
嵌套结构：Pydantic支持类内嵌套类，如actors: List[Actor]（需导入List）。
兼容性：非OpenAI模型可能需额外配置（如mode="tool_calling"）。

4. 综合应用：搭建流式响应的多轮问答机器人

现在，让我们将前两部分知识结合：使用消息格式维护对话历史，并启用流式响应构建一个交互式聊天机器人。这个机器人支持多轮对话、实时输出，并自动限制历史长度以防token超限。适合初学者快速原型化聊天应用。

1.优化说明

模型选择：采用OpenAI（更通用），但你可轻松替换为其他模型。
改进点：添加了异常处理、消息长度限制（最近50条，避免token爆炸）、退出机制优化。使用messages列表维护状态，支持上下文记忆。
运行方式：在终端运行，输入exit退出。流式输出模拟“打字机”效果。

2.完整样例代码

from langchain_openai import ChatOpenAI  # 使用OpenAI作为示例（可替换为其他模型）from langchain_core.messages import SystemMessage, HumanMessage, AIMessage# 1️⃣ 初始化模型（LangChain 1.0 接口，启用流式）model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7, streaming=True)# 2️⃣ 初始化系统提示词（System Prompt）system_message = SystemMessage(    content="你叫小智，是一名乐于助人的智能助手。请在对话中保持温和、有耐心的语气，并记住之前的上下文。")# 3️⃣ 初始化消息历史messages = [system_message]print("🔹 欢迎使用小智聊天机器人！输入 'exit' 退出对话\n")# 4️⃣ 主循环（支持多轮对话 + 流式输出）while True:    try:        user_input = input("👤 你：").strip()        if user_input.lower() in {"exit", "quit", ""}:            print("🧩 对话结束，再见！")            break        # 追加用户消息到历史        messages.append(HumanMessage(content=user_input))        # 实时输出模型生成内容        print("🤖 小智：", end="", flush=True)        full_reply = ""  # 收集完整回复，用于添加到历史        # ✅ LangChain 1.0 标准写法：流式输出（传入完整消息历史）        for chunk in model.stream(messages):            if chunk.content:                print(chunk.content, end="", flush=True)                full_reply += chunk.content        print("\n" + "-" * 50)  # 分隔线，提升可读性        # 追加 AI 回复消息到历史        messages.append(AIMessage(content=full_reply))        # 保持消息长度（只保留最近50轮，避免token超限）        if len(messages) > 50:            messages = [messages[0]] + messages[-49:]  # 保留系统消息 + 最近49条    except KeyboardInterrupt:        print("\n🧩 对话被中断，再见！")        break    except Exception as e:        print(f"\n❌ 发生错误：{e}。请重试。")        continue

3.运行示例

启动后：

👤 你：你好，小智！今天天气怎么样？🤖 小智：你好！作为AI，我无法直接查看实时天气，但如果你告诉我你的位置，我可以帮你查询或建议。有什么具体城市吗？--------------------------------------------------👤 你：在北京。🤖 小智：北京今天晴朗，温度约15-20°C。记得带件外套哦！有什么其他想聊的？--------------------------------------------------👤 你：exit🧩 对话结束，再见！

机器人记住上下文（如位置），流式输出让响应更自然。

扩展建议：集成LangSmith追踪（pip install langsmith），或用Streamlit/Flask转为Web app。结合结构化输出，可让机器人返回JSON回复（如天气数据）。