一、聊天模型的核心能力

1. 定义聊天模型

大语言模型 （LLM） 在各种与语言相关的任务（例如文本生成、翻译、摘要、问答等）中表现出色。
现代 LLM 通常通过聊天模型接口访问，该接口将消息列表作为输入，并返回消息作为输出，而不是使用纯文本。

这里需要注意 LLM 与 LangChain 中 聊天模型 的关系：

• 在 LangChain 的官方文档中，认为 LLM 大多数是纯文本补全模型。这些纯文本模型封装的 API 接受一个字符串提示作为输入，并输出一个字符串补全结果（实际上 LLM 还包括多模态输入）。OpenAI 的 GPT-5 就是作为 LLM 来实现的。
• LangChain 中的 聊天模型 通常由 LLM 提供支持，但经过专门调整以用于对话。关键在于，它们不是接受单个字符串作为输入，而是接受聊天消息列表，并返回一条 AI 消息作为输出。

1.1 通过 API 定义聊天模型

1.1.1 方式1：ChatOpenAI

ChatOpenAI 定义聊天模型在快速上手模块中已经涉及。

class langchain_openai.chat_models.base.ChatOpenAI 是 LangChain 为 OpenAI 的聊天模型（如 gpt-5 , gpt-5-mini ）提供的具体实现类。
其继承了 class langchain_openai.chat_models.base.BaseChatOpenAI ，且 BaseChatOpenAI 实现了标准的 Runnable 接口。

ChatOpenAI 常用初始化参数说明

参数名	参数描述
model	要使用的 OpenAI 模型的名称
temperature	采样温度，温度值越高，AI 回答越天马行空；温度越低，回答越保守靠谱。
max_tokens	要生成的最大令牌数
timeout	请求超时时间
max_retries	最大重试次数
openai_api_key / api_key	OpenAI API 密钥。如果未传入，将从环境变量中读取 OPENAI_API_KEY 。
base_url	API 请求的基本 URL。
organization	OpenAI 组织 ID。如果未传入，将从 env var OPENAI_ORG_ID 中读取。

示例：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-5-mini",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    # api_key="...",
    # base_url="...",
    # organization="...",
    # other params...
)

若使用其它与 OpenAI 兼容的大模型，例如 DeepSeek，则可以使用以下定义方式：

import os
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key=OPENAI_API_KEY,
    model="deepseek-chat",
    # ...
)

参数说明：

• base_url ：出于与 OpenAI 兼容考虑，要将 base_url 设置为 https://api.deepseek.com/v1 来使用，但注意，此处 v1 与模型版本无关。
• openai_api_key ：需要单独申请 DeepSeek 的 API Key，然后重新进行环境变量配置。
  • DeepSeek API Key 申请地址：https://platform.deepseek.com/api_keys。

1.1.2 invoke() 调用

介绍 方式2 之前，需要先来了解一下关于 Runnable 接口中的 .invoke() 调用。该方法是将单个输入转换为对应的输出。例如对于聊天模型来说，就是根据用户的问题输入，输出相应的答案。

invoke() 方法定义：

abstractmethod invoke(
    input: Input,
    config: RunnableConfig | None = None,
    **kwargs: Any,
) → Output

请求参数：

• input ：输入一个 Runnable 实例
• config （默认空）：用于 Runnable 的配置。

返回值：

• 返回一个 Runnable 实例

class langchain_core.runnables.config.RunnableConfig 常用参数说明

参数名	参数描述
configurable	通过 .configurable_fields（）在此 Runnable 或子 Runnable 上配置的属性的运行时值。
run_id	针对此调用运行的跟踪器的唯一标识符。如果未提供，将生成新的 UUID。
run_name	此调用的跟踪器运行的名称。默认为类的名称。
metadata	此次调用和任何子调用的元数据。键是字符串，值是 JSON。类型：dict[str， Any]

具体示例，下面的 方式2 会用到~

1.1.3 方式2：init_chat_model

上面的 ChatOpenAI 用于明确创建 OpenAI 聊天模型的实例。而 init_chat_model() 是一个工厂函数，它可以初始化多种支持的聊天模型（如 OpenAI、Anthropic、FireworksAI 等），不仅仅是 OpenAI 的聊天模型。

init_chat_model() 函数定义

langchain.chat_models.base.init_chat_model(
    model: str,
    *,
    model_provider: str | None = None,
    configurable_fields: Literal[None] = None,
    config_prefix: str | None = None,
    **kwargs: Any,
) → BaseChatModel

init_chat_model() 常用参数说明

参数名	参数描述
model	要使用的模型的名称
model_provider	模型提供方。支持的 model_provider 值和相应的集成包有：• openai -> langchain-openai• anthropic -> langchain-anthropic• google_genai -> langchain-google-genai• ollama -> langchain-ollama• deepseek -> langchain-deepseek• …如果未指定，将尝试从模型推断 model_provider。
configurable_fields	设置哪些模型参数是可配置的。若配置为：• None: 没有可配置的字段。• 'any' ：所有字段都是可配置的，类似 api_key 、 base_url 等可以在运行时更改。• Union[List[str], Tuple[str, …]]：指定的字段是可配置的。
config_prefix	• 配置为非空字符串，则模型将在运行时通过查找 config["configurable"]["{config_prefix}_{param}"] 字段设置配置项 。• 配置为空字符串，那么模型将可以通过 config["configurable"]["{param}"] 字段设置配置项 。
temperature	采样温度，温度值越高，AI 回答越天马行空；温度越低，回答越保守靠谱。
max_tokens	要生成的最大令牌数
timeout	请求超时时间
max_retries	最大重试次数
openai_api_key / api_key	OpenAI API 密钥。如果未传入，将从环境变量中读取 OPENAI_API_KEY 。
base_url	API 请求的基本 URL。

init_chat_model() 返回值说明

函数返回一个与指定的 model_name 和 model_provider 相对应的 BaseChatModel （如 ChatOpenAI , ChatAnthropic 等）。注意要是模型可配置，则返回一个聊天模型模拟器，该模拟器在传入配置后，于运行时才会初始化底层模型。

示例1：基本用法

使用不同的模型提供方，需要安装为其各自包，与设置各自的 API Key 环境变量！例如：

• OpenAI 环境变量配置为：OPENAI_API_KEY=“your_openai_api_key”
  • 安装命令： pip install -U langchain-openai
• Anthropic 环境变量配置为：ANTHROPIC_API_KEY=“your_anthropic_api_key”
  • 安装命令： pip install -U langchain-anthropic
• DeepSeek 环境变量配置为：DEEPSEEK_API_KEY=“your_deepseek_api_key”
  • 安装命令： pip install -U langchain-deepseek
• Google VertexAI 环境变量配置为：GOOGLE_API_KEY=“your_google_api_key”
  • 安装命令： pip install -U langchain-google-vertexai
• 更多见这里。

from langchain.chat_models import init_chat_model

# 返回 langchain_openai.ChatOpenAI 实例
gpt_model = init_chat_model("gpt-5-mini", model_provider="openai",
temperature=0)

# 返回 langchain_deepseek.ChatDeepSeek 实例
deepseek_model = init_chat_model("deepseek-chat", model_provider="deepseek",
temperature=0)

# 由于所有模型集成都实现了ChatModel接口，因此可以以相同的方式使用它们。
print("gpt-5-mini: " + gpt_model.invoke("what's your name").content + "\n")
print("deepseek-chat: " + deepseek_model.invoke("what's your name").content +
"\n")

输出：

gpt-5-mini: I’m called ChatGPT. How can I assist you today?
deepseek-chat: I'm DeepSeek Chat! 😊 You can call me DeepSeek or just Chat if
you'd like. I'm here to help with anything you need—ask me anything! 🚀

示例2：创建可配置模型

class langchain_core.runnables.config.RunnableConfig 常用参数说明

参数名	参数描述
configurable	通过 .configurable_fields（）在此 Runnable 或子 Runnable 上配置的属性的运行时值。
run_id	针对此调用运行的跟踪器的唯一标识符。如果未提供，将生成新的 UUID。
run_name	此调用的跟踪器运行的名称。默认为类的名称。
metadata	此次调用和任何子调用的元数据。键是字符串，值是 JSON。类型：dict[str， Any]

# 可配置模型模拟器
configurable_model_1 = init_chat_model(temperature=0)

# 动态修改配置，初始化模型并调用
configurable_result_1 = configurable_model_1.invoke(
    "what's your name", config={"configurable": {"model": "gpt-5-mini"}}
)
print("configurable1: " + configurable_result_1.content + "\n")

输出：

configurable1: I’m called ChatGPT. How can I assist you today?

示例3：具有默认值的可配置模型

# 可配置模型模拟器
configurable_model_2 = init_chat_model(
    model="gpt-5-mini",
    temperature=0,
    configurable_fields=("model", "model_provider", "temperature",
    "max_tokens"),
    config_prefix="first",
)

# 动态修改配置，初始化模型并调用
configurable_result_2 = configurable_model_2.invoke(
    "what's your name?",
    config={
        "configurable": {
            "first_model": "deepseek-chat",
            "first_temperature": 0.5,
            "first_max_tokens": 100,
        }
    },
)
print("configurable2: " + configurable_result_2.content + "\n")

输出：

configurable2: My name is DeepSeek Chat! 😊 I'm here to help you with any
questions or topics you're curious about. How can I assist you today?

1.2 通过本地部署的 LLM 定义聊天模型

1.2.1 ChatOllama

若想使用 ChatOllama，需要先安装 Ollama 包：

pip install -U langchain_ollama

class langchain_ollama.chat_models.base.ChatOllama 是 LangChain 为通过 Ollama 部署的聊天模型提供的具体实现类。 ChatOllama 同样也实现了标准的 Runnable 接口。

ChatOllama 常用初始化参数说明

参数名	参数描述
model	要使用的 Ollama 模型的名称
temperature	采样温度，温度值越高，AI 回答越天马行空；温度越低，回答越保守靠谱。
timeout	请求超时时间
base_url	API 请求的基本 URL。
num_ctx	设置用于生成下一个令牌的上下文窗口的大小。（默认值：2048）
num_gpu	要使用的 GPU 数量。在 macOS 上，默认为 1 表示启用金属支持，默认为 0 表示禁用。

示例：

from langchain_ollama import ChatOllama

ollama_model = ChatOllama(model="deepseek-r1:70b",
base_url='http://192.168.100.220:11434')
result = ollama_model.invoke("what's your name?")
print(result)

输出：

Greetings! I'm DeepSeek-R1, an artificial intelligence assistant created by DeepSeek. I'm at your service and would be delighted to assist you with any inquiries or tasks you may have."
additional_kwargs={} response_metadata={'model': 'deepseek-r1:70b',
'created_at': '2025-08-20T06:10:54.9742632Z', 'done': True, 'done_reason':
'stop', 'total_duration': 9158375400, 'load_duration': 103256900,
'prompt_eval_count': 8, 'prompt_eval_duration': 1303997800, 'eval_count': 44,
'eval_duration': 7739543700, 'model_name': 'deepseek-r1:70b'} id='run--
c62716da-3c5c-4ddf-8e76-da27dc9291a7-0' usage_metadata={'input_tokens': 8,
'output_tokens': 44, 'total_tokens': 52}

2. 聊天模型–调用工具

工具调用根本作用是让大语言模型（LLM）具备与外部世界交互的能力。

LLM 本身是一个封闭的知识系统，其能力受限于其训练数据（存在滞后性）和内在的文本生成逻辑。它无法执行直接计算、查询实时信息、操作数据库或调用任何外部 API。工具调用打破了这层壁垒，其作用具体体现在：

1. 扩展能力边界：模型可以借助工具完成它自身无法完成的任务，如执行数学计算、搜索网络、查询数据库等。
2. 保证信息实时性：通过调用搜索工具或数据库查询工具，LLM 可以获取最新的、训练数据中不存在的信息，避免回答过时或“一本正经地胡说八道”。
3. 处理复杂任务：将一个复杂的用户请求（如“分析我上个月的消费趋势”）分解成多个步骤，并依次调用不同的工具（如“从数据库获取数据” -> “用 Python 进行数据分析” -> “生成图表”）来协同完成。协调这件事更体现在 Agent 智能体上。
4. 连接现有系统：可以将企业内部已有的系统、API 和数据库封装成工具，让 LLM 成为一个用自然语言驱动的统一接口，极大地提升了自动化和集成能力。

在 LangChain 中，聊天模型提供了额外的功能：工具调用。它能使 LLM 与外部服务、API 和数据库进行交互。工具调用还可用于从非结构化数据中提取结构化信息并执行各种其他任务。

例如，当我们希望获取当前天气情况时，由于 LLM 无法获取实时信息，此时我们就可以借助工具，通过外部服务进行搜索完成查询；
再例如，当我们希望获取数据库表中的数据时，由于 LLM 无法直接获取表数据，此时我们就可以借助工具，通过与数据库交互完成查询。

2.1 创建工具

2.1.1 使用 @tool 装饰器创建工具

在 LangChain 中，实现了一个@tool装饰器来创建工具，@tool装饰器是自定义工具的最简单方法。如下所示：

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers.
    Args:
        a: First integer
        b: Second integer
    """
    return a * b

print(multiply.invoke({"a": 3, "b": 4}))  # 输出：12
print(multiply.name)                      # 输出：multiply
print(multiply.description)               # 输出：Multiply two ...省略...b: Second integer
print(multiply.args)                      # 输出：{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}

可以看出，工具通过 @tool 加 Python函数 实现，其中：

• 该装饰器默认使用函数名称作为工具名称。
• 该装饰器将使用函数的文档字符串作为工具的描述。

因此，函数名、类型提示和文档字符串都是传递给工具 Schema 的一部分，不可缺失。定义好的描述是使模型良好运行的重要部分。

什么是 Schema ？

答：Schema 是描述其他数据结构的声明格式，用于规范化数据结构、自动校验数据合法性，不包含业务逻辑代码，仅用于定义数据规范。

可以通过两组 JSON 示例直观理解数据结构化的差异：

示例1（简易结构化数据）：

{
    "name": "李小明",
    "birthday": "1998年5月12日",
    "address": "浙江省杭州市西湖区"
}

示例2（标准结构化数据）：

{
    "surname": "李",
    "given_name": "小明",
    "birthday": "1998-05-12",
    "address": {
        "district": "西湖区",
        "city": "杭州市",
        "province": "浙江省",
        "country": "中国"
    }
}

两种数据表述均有效，但示例2结构更规范、可解析性更强。JSON Schema 就是用来定义这类规范数据结构的模板，用于统一数据格式、自动校验数据是否合规。

对应上述标准数据的 JSON Schema 模板如下：

{
    "type": "object",
    "properties": {
        "surname": { "type": "string" },
        "given_name": { "type": "string" },
        "birthday": { "type": "string", "format": "date" },
        "address": {
            "type": "object",
            "properties": {
                "district": { "type": "string" },
                "city": { "type": "string" },
                "province": { "type": "string" },
                "country": { "type" : "string" }
            }
        }
    }
}

使用该 Schema 可校验数据合法性，简易格式的示例1无法通过校验，标准化的示例2可完全通过校验。

核心特点： JSON Schema 仅为数据描述模板，非可执行程序，只能校验数据结构，无法校验数据之间的业务逻辑关系，复杂数据校验需结合编程语言实现语义校验。

工具 Schema 基于该原理，自动从工具的函数名、类型提示、文档字符串中提取信息，定义工具的名称、描述、入参、出参等核心规范，保障模型调用工具时参数合法、格式统一。

工具默认解析 Google 风格文档字符串 获取参数描述，该风格是 Python 通用注释规范，可读性极强，示例如下：

def fetch_data(url, retries=3):
    """从给定的URL获取数据。
    Args:
        url (str): 要从中获取数据的URL。
        retries (int, optional): 失败时重试的次数。默认为3。
    Returns:
        dict: 从URL解析的JSON响应。
    """
    # 函数业务逻辑实现
    pass

2.1.1.1 模式1：依赖 Pydantic 类定义工具

仅使用 @tool 装饰器时，若函数无文档字符串，会直接抛出参数异常：ValueError: Function must have a docstring if description not provided.。

此时可通过 Pydantic 类 定义工具入参 Schema，通过 Field 注解补充参数描述，实现无文档字符串也可正常定义工具，同时支持运行时数据校验。

完整示例代码：

# 导入pydantic数据校验工具
from pydantic import BaseModel, Field
# 导入工具装饰器
from langchain_core.tools import tool

# 定义加法工具入参Schema
class AddInput(BaseModel):
    """Add two integers."""
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")

# 定义乘法工具入参Schema
class MultiplyInput(BaseModel):
    """Multiply two integers."""
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")

# 绑定Schema定义工具，无需文档字符串
@tool(args_schema=AddInput)
def add(a: int, b: int) -> int:
    return a + b

@tool(args_schema=MultiplyInput)
def multiply(a: int, b: int) -> int:
    return a * b

通过 args_schema 参数绑定 Pydantic 类后，工具可自动完成参数校验、Schema 生成，规避无文档字符串的报错问题。

2.1.1.2 模式2：依赖 Annotated 定义工具

除 Pydantic 外，也可通过 Annotated 注解直接在函数参数中添加参数描述，配合文档字符串生成工具 Schema，写法更简洁直观。

示例代码：

from langchain_core.tools import tool
from typing_extensions import Annotated

@tool
def add(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Add two integers."""
    return a + b

@tool
def multiply(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Multiply two integers."""
    return a * b

2.1.2 使用 StructuredTool 类创建工具

StructuredTool 是 LangChain 提供的原生工具类，通过其 from_function 类方法可灵活初始化工具，支持自定义工具名称、描述、入参 Schema、响应格式，适配复杂工具场景。

核心方法定义：

classmethod from_function(
    func: Callable | None = None,
    coroutine: Callable[[...], Awaitable[Any]] | None = None,
    name: str | None = None,
    description: str | None = None,
    return_direct: bool = False,
    args_schema: type[BaseModel] | dict[str, Any] | None = None,
    infer_schema: bool = True,
    *,
    response_format: Literal['content', 'content_and_artifact'] = 'content',
    parse_docstring: bool = False,
    error_on_invalid_docstring: bool = False,
    **kwargs: Any,
) -> StructuredTool

关键参数说明：

• func：待封装的同步工具函数
• coroutine：待封装的异步工具函数
• name：自定义工具名称，默认取函数名
• description：自定义工具描述，默认取函数文档字符串
• args_schema：工具入参校验 Schema，适配 Pydantic 类
• response_format：工具响应格式，支持 content（纯文本）和 content_and_artifact（文本+原始数据）

2.1.2.1 示例1：常规基础用法

from langchain_core.tools import StructuredTool

# 定义工具函数
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

# 初始化工具
calculator_tool = StructuredTool.from_function(func=multiply)
# 调用工具
print(calculator_tool.invoke({"a": 3, "b": 4}))  # 输出：12

2.1.2.2 示例2：结合 Pydantic 自定义配置

可脱离函数文档字符串，通过 Pydantic 定义参数规范，手动指定工具名称和描述，灵活性更高。

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

# 定义入参Schema
class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

# 纯业务函数，无注释描述
def multiply(a: int, b: int) -> int:
    return a * b

# 自定义工具配置
calculator_tool = StructuredTool.from_function(
    func=multiply,
    name="Calculator",
    description="两数相乘计算工具",
    args_schema=CalculatorInput,
)

print(calculator_tool.invoke({"a": 3, "b": 4}))  # 输出：12
print(calculator_tool.name)                      # 输出：Calculator
print(calculator_tool.description)               # 输出：两数相乘计算工具

2.1.2.3 示例3：配置 response_format 区分内容与原始数据

content 是给大模型读取的结构化文本结果，artifact 是供程序后续处理的原始结构化数据，不直接输入给大模型，适用于日志记录、二次分析、问题排查等场景。

示例场景：天气搜索工具调用后，content 为精简天气总结，artifact 为接口返回的完整原始数据。

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
from typing import List, Tuple

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")

# 返回 文本结果 + 原始数据 二元组
def multiply(a: int, b: int) -> Tuple[str, List[int]]:
    nums = [a, b]
    content = f"{nums}相乘的计算结果为：{a * b}"
    return content, nums

# 配置双响应格式
calculator_tool = StructuredTool.from_function(
    func=multiply,
    name="Calculator",
    description="两数相乘计算工具",
    args_schema=CalculatorInput,
    response_format="content_and_artifact"
)

# 普通调用：仅返回content文本
print(calculator_tool.invoke({"a": 3, "b": 4}))

# 模型式调用：返回完整ToolMessage（包含content、artifact、调用ID）
print(calculator_tool.invoke(
    {
        "name": "Calculator",
        "args": {"a": 3, "b": 4},
        "id": "tool_001",
        "type": "tool_call",
    }
))

核心小结： content 服务于大模型推理，artifact 服务于程序后续自动化处理、数据溯源。

2.2 绑定工具到聊天模型

自定义工具后，需通过聊天模型的 bind_tools() 方法将工具绑定到模型，让模型具备工具调用能力，绑定后返回可运行的 Runnable 实例。

基础绑定示例：

from langchain_openai import ChatOpenAI

# 初始化兼容DeepSeek的聊天模型
model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 绑定自定义工具
tools = [add, multiply]
model_with_tools = model.bind_tools(tools)

bind_tools 核心参数：

• tools：待绑定的工具列表，支持函数、Pydantic类、BaseTool等多种格式
• tool_choice：工具调用策略，auto自动选择、none不调用、any强制调用至少一个工具
• strict：是否严格校验工具入参出参Schema
• parallel_tool_calls：是否允许并行调用多个工具

2.3 基础工具调用流程

模型绑定工具后，通过 invoke() 方法接收用户指令，自动判断是否需要调用工具，返回包含工具调用信息的 AIMessage。

完整调用示例：

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing_extensions import Annotated

# 初始化模型
model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 定义工具
@tool
def add(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Add two integers."""
    return a + b

@tool
def multiply(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Multiply two integers."""
    return a * b

# 绑定工具
tools = [add, multiply]
model_with_tools = model.bind_tools(tools)

# 触发工具调用
result = model_with_tools.invoke("9乘6等于多少？")
print(result)

当用户提问无需工具即可回答（如普通问候），模型会直接返回文本结果，不触发工具调用。

2.4 强制工具调用

通过 tool_choice="any" 可强制模型调用工具，无论用户输入是否需要工具能力，适用于固定工具调用的业务场景。

# 强制调用工具
model_with_tools = model.bind_tools(tools, tool_choice="any")
result = model_with_tools.invoke("你好！")
print(result.tool_calls)

2.5 工具调用核心属性

模型触发工具调用后，返回的 AIMessage 会自带 tool_calls 属性，包含工具名称、入参、唯一调用ID等核心信息，可用于解析和执行工具。

result = model_with_tools.invoke("9乘6等于多少？")
# 打印工具调用详情
print(result.tool_calls)

2.6 工具结果回传模型生成最终答案

单次工具调用仅能获取工具执行指令和结果，需将 HumanMessage、AIMessage、ToolMessage 完整消息链回传给模型，才能生成最终的自然语言答案。

完整闭环示例：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from typing_extensions import Annotated

# 初始化模型
model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 定义工具
@tool
def add(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Add two integers."""
    return a + b

@tool
def multiply(
    a: Annotated[int, ..., "First integer"],
    b: Annotated[int, ..., "Second integer"]
) -> int:
    """Multiply two integers."""
    return a * b

# 绑定工具
tools = [add, multiply]
model_with_tools = model.bind_tools(tools)

# 构建消息链
messages = [HumanMessage("9乘6等于多少？5加3等于多少？")]
# 第一次调用：模型生成工具调用指令
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)

# 遍历执行所有工具调用
tool_map = {"add": add, "multiply": multiply}
for tool_call in ai_msg.tool_calls:
    tool = tool_map[tool_call["name"].lower()]
    tool_msg = tool.invoke(tool_call)
    messages.append(tool_msg)

# 第二次调用：模型结合工具结果生成最终答案
final_result = model.invoke(messages)
print(final_result.content)

完整流程总结： 用户提问 -> 模型生成工具调用指令 -> 程序执行工具 -> 工具结果回传模型 -> 模型输出最终答案。

2.7 LangChain 官方内置工具

LangChain 无需全部手动自定义工具，官方提供了大量现成工具、工具包（Toolkit），涵盖搜索、数据库、网页解析、文件处理等场景，所有内置工具均继承 BaseTool/BaseToolkit，开箱即用。

2.7.1 TavilySearch 智能搜索工具

Tavily 是专为 AI 推理设计的搜索引擎工具，可精准检索实时网络信息，返回结构化、高相关性的搜索结果，适配天气查询、资讯检索、知识问答等实时场景。

工具特性：检索精度高于传统搜索引擎、结构化返回结果、适配大模型推理场景。

使用步骤：

1. 安装依赖包

pip install -U langchain-tavily

2. 配置环境变量：在系统环境中配置 TAVILY_API_KEY（官网申请）
3. 完整调用示例

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_tavily import TavilySearch

# 初始化模型
model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 初始化搜索工具，设置最大返回结果数
tool = TavilySearch(max_results=4)
model_with_tools = model.bind_tools([tool])

# 构建提问
messages = [HumanMessage("杭州今日天气情况如何？")]
# 模型生成搜索指令
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)

# 执行搜索工具
for tool_call in ai_msg.tool_calls:
    tool_msg = tool.invoke(tool_call)
    messages.append(tool_msg)

# 模型整合搜索结果输出答案
result = model_with_tools.invoke(messages)
print(result.content)

调用输出示例：

杭州今日天气情况如下：
- 天气：晴天转多云
- 温度：24℃~33℃
- 提示：夜间局部有短时阵雨，建议出行携带薄外套。
详细气象数据可参考中国气象局官方天气预报平台。

3. 聊天模型–结构化输出

在 LangChain 中，聊天模型提供了结构化输出能力，这是一种让聊天模型按照指定结构化格式（典型如 JSON）返回响应的核心技术。在实际业务开发中，我们常常需要将模型输出落地存储到数据库、对接后端接口、完成数据解析，这就要求模型输出必须符合固定的数据范式，结构化输出能力正是为解决该场景需求而生。

结构化输出的核心价值是实现了从「字符串文本」到「结构化对象」的范式转换。在无该能力时，调用聊天模型返回的 AIMessage 内容是纯字符串，对人类阅读友好，但对程序极不友好。若需要从自由文本中提取指定字段（如公司名、股价、时间、数值等），必须手动编写正则表达式、字符串分割等解析代码，不仅开发效率低，还极易因文本格式差异出现解析报错、数据丢失等问题。

而通过 LangChain 提供的结构化输出方法，可预先定义标准化数据结构，强制大模型严格按照该结构返回内容，彻底规避手动解析的弊端，实现模型输出可直接被程序识别、调用、存储。

3.1 with_structured_output() 核心方法

with_structured_output() 是 LangChain 实现结构化输出最简单、最可靠的原生方法。核心使用逻辑为：先自定义输出数据结构，再通过该方法绑定结构，生成支持结构化返回的 Runnable 实例，最终调用实例即可获取标准化结构化结果。

基础使用伪代码如下：

# 1. 定义自定义输出结构
schema = {"foo": "bar"}
# 2. 绑定结构，生成结构化输出模型实例
model_with_structure = model.with_structured_output(schema)
# 3. 调用模型，直接获取结构化结果
structured_output = model_with_structure.invoke(user_input)

该方法支持多种主流结构定义方式，适配不同开发场景：可传入 Pydantic 类、TypedDict 类、JSON Schema。其中 Pydantic 结构返回 Pydantic 对象，TypedDict、JSON Schema 统一返回字典对象，所有结果均可直接用于程序逻辑处理。

3.1.1 方法完整定义与参数详解

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 Schema 字典、TypedDict 类、Pydantic 类、OpenAI 工具函数等多种格式，是结构化输出的核心依据。
• method：指定大模型生成结构化数据的底层实现方式，默认值为 json_schema：
  • json_schema：调用 OpenAI 官方结构化输出 API，稳定性、规范性最优，为首选方式；
  • function_calling：基于传统工具调用（函数调用）能力实现结构化输出；
  • json_mode：开启模型原生 JSON 输出模式，需手动在提示词中声明 Schema 规范，否则输出可能不达标。
• include_raw：是否返回原始模型响应，默认 False：
  • False：仅返回解析完成的结构化数据，解析失败时直接抛出异常；
  • True：返回包含 raw、parsed、parsing_error 的完整字典，raw 为原始 AIMessage、parsed 为结构化结果、parsing_error 为解析异常信息，便于问题排查。
• strict：严格校验模式，默认 None：
  • True：强制模型输出与 Schema 完全匹配，同时校验输入 Schema 的合法性；
  • False：不校验输入 Schema 和模型输出，兼容性强但规范性弱；
  • None：不向模型传递严格校验参数，使用默认规则。
• tools：绑定配套工具列表，仅在 method=json_schema、strict=True、include_raw=True 时生效，原始响应中会携带工具调用信息。
• kwargs：额外扩展参数，直接透传给模型 bind() 方法。

返回值规则： 返回 Runnable 可运行实例，调用后输出规则如下：

• include_raw=False 时：Pydantic 结构返回 Pydantic 对象，其余结构返回字典；
• include_raw=True 时：统一返回字典，包含 raw（原始消息）、parsed（结构化结果）、parsing_error（异常信息）三个固定字段。

3.1.2 返回 Pydantic 对象（常用）

通过 Pydantic 类定义输出结构，支持字段类型校验、默认值、字段描述、嵌套结构，是企业级开发中最常用的结构化输出方式。LangChain 会自动解析模型返回的 JSON 数据，校验格式后封装为 Pydantic 对象，支持直接调用对象属性，代码可读性和健壮性极强。

基础单层结构示例：

from langchain_openai import ChatOpenAI
from typing import Optional
from pydantic import BaseModel, Field

# 初始化兼容DeepSeek的聊天模型
model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 定义结构化输出结构
class Joke(BaseModel):
    """生成一则趣味笑话"""
    setup: str = Field(description="笑话的开头铺垫内容")
    punchline: str = Field(description="笑话的核心妙语、笑点")
    rating: Optional[int] = Field(default=None, description="笑话评分，范围1-10分")

# 绑定结构，生成结构化输出模型
structured_model = model.with_structured_output(Joke)
# 调用模型获取结构化结果
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)
# 可直接调用属性
print(f"笑话开头：{result.setup}，评分：{result.rating}")

输出结果：

setup='为什么歌手总是喜欢在洗手间里唱歌？' punchline='因为那里有很好的回音和灵感！' rating=7
笑话开头：为什么歌手总是喜欢在洗手间里唱歌？，评分：7

嵌套结构示例（支持复杂多层数据）：

from langchain_openai import ChatOpenAI
from typing import Optional, List
from pydantic import BaseModel, Field

model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 子结构：单条笑话
class Joke(BaseModel):
    """生成一则趣味笑话"""
    setup: str = Field(description="笑话的开头铺垫内容")
    punchline: str = Field(description="笑话的核心妙语、笑点")
    rating: Optional[int] = Field(default=None, description="笑话评分，范围1-10分")

# 父结构：嵌套多条笑话
class JokeData(BaseModel):
    """批量笑话数据"""
    jokes: List[Joke] = Field(description="笑话列表，可包含多条不同主题笑话")

structured_model = model.with_structured_output(JokeData)
result = structured_model.invoke("分别讲一个关于唱歌和跳舞的笑话")
print(result)

输出结果：

jokes=[
Joke(setup='为什么唱歌的人总是很快乐？', punchline="因为他们总是'音'乐满满！", rating=8),
Joke(setup='一个跳舞的牛走进俱乐部，为什么大家都不理它？', punchline="因为它总是'踏'错节拍！", rating=7)
]

3.1.3 返回 TypedDict 字典

TypedDict 用于为 Python 字典提供精准的类型约束，可指定字典键名、对应值类型，能在开发阶段捕获键名拼写错误、类型不匹配问题，轻量简洁，适合简单结构化场景。通过该方式定义的结构，模型最终返回标准字典对象。

基础使用示例：

from langchain_openai import ChatOpenAI
from typing import Optional
from typing_extensions import Annotated, TypedDict

model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 定义TypedDict结构化输出规则
class Joke(TypedDict):
    """生成一则趣味笑话"""
    setup: Annotated[str, ..., "笑话的开头铺垫内容"]
    punchline: Annotated[str, ..., "笑话的核心妙语、笑点"]
    rating: Annotated[Optional[int], None, "笑话评分，范围1-10分"]

structured_model = model.with_structured_output(Joke)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)

输出结果（标准字典）：

{'setup': '为什么歌手总是带一把伞？', 'punchline': '因为他们怕下雨时会错过音调！', 'rating': 7}

开启 include_raw 完整返回模式示例：

# 开启原始数据返回
structured_model = model.with_structured_output(Joke, include_raw=True)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)

输出结果（包含原始消息、结构化结果、异常信息）：

{
'raw': AIMessage(content='{"setup":"你知道为什么歌手总是喜欢在吃饭的时候唱歌吗？","punchline":"因为他们想要增加自己的‘调味’！","rating":7}', ...),
'parsed': {'setup': '你知道为什么歌手总是喜欢在吃饭的时候唱歌吗？', 'punchline': '因为他们想要增加自己的‘调味’！', 'rating': 7},
'parsing_error': None
}

3.1.4 返回原生 JSON 格式

直接通过 JSON Schema 字典定义输出规范，无需依赖 Pydantic、TypedDict 依赖，轻量化适配所有场景，最终返回标准 JSON 字典，可直接对接接口、存储数据库。

使用示例：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 自定义JSON Schema结构
json_schema = {
    "title": "JokeSchema",
    "description": "生成一则趣味唱歌主题笑话",
    "type": "object",
    "properties": {
        "setup": {
            "type": "string",
            "description": "笑话的开头铺垫内容"
        },
        "punchline": {
            "type": "string",
            "description": "笑话的核心妙语、笑点"
        },
        "rating": {
            "type": "integer",
            "description": "笑话评分，范围1-10分",
            "default": None
        }
    },
    "required": ["setup", "punchline"]
}

structured_model = model.with_structured_output(json_schema)
result = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(result)

输出结果：

{'setup': '为什么唱歌的人总是很开心？', 'punchline': '因为他们总是有很多音符可供选择！', 'rating': 7}

3.1.5 多格式联合输出（Union 类型）

通过 Union 联合类型定义多模式输出结构，可让模型根据用户输入自动匹配对应的输出格式，适配多场景自适应响应需求。以下以 Pydantic 多结构联合为例实现：

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Optional, Union

model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 场景1：笑话输出结构
class Joke(BaseModel):
    """趣味笑话生成结构"""
    setup: str = Field(description="笑话开头铺垫")
    punchline: str = Field(description="笑话核心笑点")
    rating: Optional[int] = Field(default=None, description="1-10分笑话评分")

# 场景2：普通对话输出结构
class ChatResponse(BaseModel):
    """通用对话响应结构"""
    response: str = Field(description="友好的自然语言回复")

# 联合多结构，实现自适应输出
class FinalResponse(BaseModel):
    final_output: Union[Joke, ChatResponse]

structured_model = model.with_structured_output(FinalResponse)

# 场景1：触发笑话输出
res1 = structured_model.invoke("给我讲一个关于唱歌的笑话")
print(res1)

# 场景2：触发普通对话输出
res2 = structured_model.invoke("你好，有什么可以帮你的？")
print(res2)

输出结果：

final_output=Joke(setup='为什么歌手总是带着梯子？', punchline='因为他们想要达到更高的音调！', rating=7)
final_output=ChatResponse(response='你好！有什么我可以帮助你的吗？')

3.2 结构化输出核心实用场景

3.2.1 场景一：通用信息提取器（高频场景）

结构化输出最核心的用途是从自由文本中精准、自动提取结构化信息，无需手动正则解析，支持字段可选、自动容错，适配信息抽取、数据清洗、文本结构化转换等业务。

示例：从自然文本中提取人物信息

from langchain_openai import ChatOpenAI
from typing import Optional
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage, SystemMessage

model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 定义人物信息提取结构
class PersonInfo(BaseModel):
    """人物基础信息结构化提取"""
    name: Optional[str] = Field(default=None, description="人物姓名，未知则返回null")
    hair_color: Optional[str] = Field(default=None, description="人物头发颜色，未知则返回null")
    skin_color: Optional[str] = Field(default=None, description="人物肤色，未知则返回null")
    height_in_meters: Optional[str] = Field(default=None, description="人物身高（单位：米），未知则返回null")

structured_model = model.with_structured_output(schema=PersonInfo)

# 构建提示词，精准提取文本信息
messages = [
    SystemMessage(content="你是专业的文本信息提取专家，仅从输入文本中提取指定字段信息，未知字段统一返回null，不编造内容"),
    HumanMessage(content="史密斯身高6英尺，金发，无其他外貌信息")
]

result = structured_model.invoke(messages)
print(result)

输出结果：

name='史密斯' hair_color='金发' skin_color=None height_in_meters='1.83'

3.2.2 场景二：与工具调用结合使用

结构化输出可与 LangChain 工具调用能力结合，实现「工具执行+结果结构化整理」闭环。需要注意：原生 with_structured_output 不会自动执行工具，仅能识别工具调用指令，需手动执行工具并回传结果，最终生成结构化数据，复杂场景建议后续使用 LangGraph Agent 优化。

标准可用实现方案（先绑定工具、后结构化输出）：

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

model = ChatOpenAI(
    base_url="https://api.deepseek.com/v1",
    openai_api_key="你的DEEPSEEK_API_KEY",
    model="deepseek-chat",
    temperature=0
)

# 定义搜索结果结构化输出结构
class SearchResult(BaseModel):
    """搜索结果结构化数据"""
    query: str = Field(description="用户原始搜索查询")
    findings: str = Field(description="搜索结果核心摘要信息")

# 定义搜索工具
@tool
def web_search(query: str) -> str:
    """联网搜索工具，用于获取实时信息
    Args:
        query: 用户搜索关键词
    """
    # 模拟实时搜索返回结果
    return "西安今天多云转小雨，气温18-23度，东南风2级，空气质量良好"

# 1. 先绑定工具，生成工具调用模型
model_with_tool = model.bind_tools([web_search])
messages = [HumanMessage("搜索当前最新的西安的天气")]

# 2. 模型生成工具调用指令，手动执行工具
ai_msg = model_with_tool.invoke(messages)
messages.append(ai_msg)
for tool_call in ai_msg.tool_calls:
    tool_result = web_search.invoke(tool_call)
    messages.append(tool_result)

# 3. 绑定结构化输出，整合工具结果生成标准化数据
structured_model = model_with_tool.with_structured_output(SearchResult)
result = structured_model.invoke(messages)
print(result)

输出结果：

query='西安天气' findings='西安今天多云转小雨，气温18-23度，东南风2级，空气质量良好。'

3.3 结构化输出核心总结

1. 核心优势： 彻底解决大模型自由文本输出难以被程序解析的痛点，实现文本数据标准化，适配数据库存储、接口对接、自动化数据处理等工程化场景。
2. 三种主流结构选型： Pydantic 适合复杂、需要数据校验的企业级场景；TypedDict 适合轻量字典结构化场景；JSON Schema 适合无依赖、纯标准化输出场景。
3. 工具结合要点： 结构化输出不会自动执行工具，需手动完成工具调用、结果回传流程，复杂多工具、多轮对话场景可通过 LangGraph 简化开发。

4. 聊天模型–流式传输

流式处理是LLM应用开发中提升用户体验的核心能力。传统非流式调用会等待模型生成完整响应后一次性返回，若生成长文本耗时较久，用户会面临长时间空白等待的问题，体验极差。而流式传输可让模型逐字、逐段逐步输出内容，在完整响应生成前持续推送内容，大幅降低用户等待感知，是各类聊天、问答类AI应用的标配能力。

此前使用的 invoke() 调用属于非流式传输，为一次性全量返回模式，伪代码示例如下：

# 非流式传输：等待全部内容生成后一次性返回
model = ChatOpenAI()
model.invoke("讲一个1000字的笑话")
# 等待20s后，一次性返回完整长文本结果
# 从前有一个小镇，...... ，毕竟，生活中总需要一些笑声来调剂。

主流大模型客户端（如DeepSeek客户端）默认采用流式返回，实时逐字展示内容，彻底解决长时间等待问题。LangChain 所有聊天模型均原生支持流式传输，提供同步流式和异步流式两种实现方案。

4.1 stream() 同步流式传输

LangChain 聊天模型提供 stream() 同步流式方法，调用后返回一个迭代器，会在模型生成每一段消息块（Token块）时同步产出，通过 for 循环即可实时捕获、打印、处理每一段输出内容。

流式传输返回的最小单元为 AIMessageChunk（消息块），是完整 AIMessage 的片段，支持拼接合并，可完整还原最终响应内容。

同步流式完整示例：

from langchain_openai import ChatOpenAI

# 初始化大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 流式输出：逐块获取并打印内容
chunks = []
for chunk in model.stream("讲一个50字的笑话"):
    chunks.append(chunk)
    # 实时打印，| 分隔token，flush=True 强制刷新输出
    print(chunk.content, end="|", flush=True)

实时打印结果：

|有|一天|，|兔|子|和|乌|龟|比赛|跑|步|。|兔|子|跑|得|飞|快|，|没|想到|中|途|睡|着|了|。|等|它|醒|来|，|发现|乌|龟|快|到|终|点|了|，|兔|子|急|了|：“|怎么|可能|！”|乌|龟|笑|着|说|：“|慢|就是|快|，我|在|享|受|风|景|！”||

AIMessageChunk 拼接特性： 所有消息块支持直接相加合并，可还原完整响应内容，示例如下：

# 拼接多个消息块，还原完整内容
print("\n合并完整内容：")
print(chunks[0] + chunks[1] + chunks[2] + chunks[3] + chunks[4])

拼接结果：

content='有一天，兔' additional_kwargs={} response_metadata={} id='run--e619cbc3-9ee9-4ae7-a73e-32edb166d401'

4.2 astream() 异步流式传输

同步流式 stream() 为阻塞式调用，单线程下同一时间只能处理一个任务。而 astream() 是专为非阻塞、高并发场景设计的异步流式方法，基于 Python 异步协程实现，可在单线程内并发处理多个流式任务，大幅提升程序运行效率。

4.2.1 异步核心概念

为理解异步流式优势，通过「煮水+发短信」场景对比同步阻塞与异步并发的差异，同时引入协程、事件循环核心概念。

同步阻塞模式（串行执行，效率低）： 任务必须逐一对接执行，前一个任务阻塞等待时，CPU 完全空闲，无法处理其他任务。

import time

def boil_water():
    print("开始煮水...")
    time.sleep(5) # 模拟阻塞等待5秒
    print("水开了！")

def send_message():
    print("开始发短信...")
    time.sleep(2) # 模拟阻塞等待2秒
    print("短信发送成功！")

# 主程序：串行执行
def main():
    boil_water() # 先花5秒煮水，全程阻塞
    send_message() # 水开后再花2秒发短信

main()
# 总耗时：7秒

异步并发模式（基于协程+事件循环）： 核心依托 asyncio 库，通过协程实现单线程并发调度。

核心概念解析：

• 协程： 轻量级并发编程模型，可理解为用户态轻量级线程。通过async def 定义，await 关键字实现暂停让出CPU资源，等待IO操作完成后恢复执行，上下文切换开销远低于线程、进程。
• 事件循环： asyncio 核心调度器，维护任务列表，自动检测任务状态，IO阻塞时切换就绪任务，实现单线程并发调度。

异步并发示例：

import asyncio

# 定义异步协程任务
async def boil_water_async():
    print("开始煮水...")
    await asyncio.sleep(5) # 阻塞等待时让出控制权，执行其他任务
    print("水开了！")

async def send_message_async():
    print("开始发短信...")
    await asyncio.sleep(2) # 非阻塞等待，让出控制权
    print("短信发送成功！")

# 主协程：调度多个异步任务
async def main():
    # 创建任务，交由事件循环调度
    task1 = asyncio.create_task(boil_water_async())
    task2 = asyncio.create_task(send_message_async())
    # 等待所有任务执行完成
    await task1
    await task2

# 启动事件循环，运行异步程序
asyncio.run(main())
# 总耗时：5秒（任务并发执行，无冗余等待）

异步执行输出结果：

开始煮水...
开始发短信...
（等待约2秒）
短信发送成功！
（等待约3秒）
水开了！

异步核心总结： 协程是可暂停、可恢复的异步函数；事件循环负责统一调度所有协程任务，实现单线程高并发，完美适配IO密集型的模型流式调用场景。

4.2.2 astream() 异步流式使用示例

astream() 为异步迭代器，需在异步函数中通过 async for 循环遍历，实现非阻塞流式输出。

from langchain_openai import ChatOpenAI
import asyncio

# 初始化大模型
model = ChatOpenAI(model="gpt-4o-mini")

# 异步流式调用函数
async def async_stream():
    print("=== 异步流式调用 ===")
    # 异步迭代获取流式输出块
    async for chunk in model.astream("讲一个50字的笑话"):
        print(chunk.content, end="|", flush=True)

# 执行异步任务
asyncio.run(async_stream())

异步流式输出结果：

=== 异步调用 ===
|有|一天|，一|只|鸭|子|走|进|药|店|，|问|：“|你|们|有|口|红|吗|？”|药|剂|师|说|：“|没|有|。”|鸭|子|失|望|地|摇|摇|头|，|转|身|离|开|。|第二|天|，|鸭|子|又|来|：“|你|们|有|口|红|吗|？”|药|剂|师|说|：“|我|说|过|没有|！”|鸭|子|说|：“|那|你|们|为什么|不|进|货|呢|？”||

4.3 结合 StrOutputParser 解析流式输出

聊天模型、输出解析器、LCEL 链均实现 LangChain Runnable 接口，天然支持流式传输、批量调用、组合复用等能力。需要注意：并非所有组件都支持流式，例如检索器（Retrievers）无流式能力。

模型原生流式返回AIMessageChunk 对象，通过 StrOutputParser 可直接提取消息文本内容，简化流式数据处理，适配 LCEL 链式编程。

链式流式解析示例：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 初始化模型与解析器
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# LCEL 构建流式链路：模型输出 + 文本解析
chain = model | parser

# 链式流式输出，直接获取纯文本token
for chunk in chain.stream("写一段关于爱情的歌词，需要5句话"):
    print(chunk, end="|", flush=True)

解析后流式输出结果：

|在|星|空|下|许|下|心|愿|，|
|你的|笑|容|如|晨|光|温|暖|，|
|手|握|手|走|过|每|段|光|阴|，|
|无|论|风|雨|依|然|不|离|不|弃|，|
|爱|是|永|恒|，|心|与|心|相|连|。| ||

4.4 自定义流式输出解析器

原生流式为逐字输出，可通过生成器函数自定义流式切割规则，实现按句子、段落等自定义粒度输出，同时保留流式实时特性。核心原理：基于迭代器转换 Iterator[str] -> Iterator[List[str]]，缓冲区缓存文本，匹配分隔符后批量产出。

以「句号分割单句流式输出」为例，自定义解析器完整示例：

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from typing import Iterator, List

# 初始化模型与基础解析器
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# 自定义生成器：按句号分割句子，逐句流式输出
def split_into_list(input: Iterator[str]) -> Iterator[List[str]]:
    buffer = ""
    for chunk in input:
        buffer += chunk
        # 匹配句号，切割完整句子并产出
        while "。" in buffer:
            stop_index = buffer.index("。")
            yield [buffer[:stop_index].strip()]
            buffer = buffer[stop_index + 1 :]
    # 输出剩余文本
    if buffer.strip():
        yield [buffer.strip()]

# 构建自定义流式链路
chain = model | parser | split_into_list

# 执行自定义流式输出
for chunk in chain.stream("写一份关于爱情的歌词，需要5句话，每句话用句号分割"):
    print(chunk, end="|", flush=True)

自定义流式输出结果：

['在星空下许下承诺的誓言']|['你的笑容如同晨曦，温暖了我的心']|['无论时光如何流转，我愿与你携手共行']|['爱情是我们心中永恒的旋律']|['每一次相拥，都是一场甜蜜的重逢']|

4.5 流式传输底层原理深度解析

4.5.1 SSE 协议核心介绍

HTTP 协议默认是无状态单向请求-响应模式，默认无法实现服务端主动推送。SSE（Server-Sent Events，服务器发送事件）是基于 HTTP 的轻量级实时流式通信协议，是LLM流式传输的底层核心协议。

SSE 核心特点：

• 基于标准 HTTP/HTTPS 协议，无需额外端口，部署简单、兼容性强；
• 单向通信：仅支持服务端向客户端持续推送数据流，客户端单次建立长连接；
• 自带自动断线重连机制，保障流式传输稳定性；
• 支持自定义事件类型、消息ID、重连间隔等配置。

SSE 固定请求头：服务端开启流式传输必须配置以下响应头：

Content-Type: text/event-stream;charset=utf-8
Connection: keep-alive

SSE 数据格式：每条消息由固定字段组成，多条消息以\n\n 分隔，核心字段：

• data（必填）：流式推送的核心数据内容；
• event（可选）：自定义事件类型，默认 message；
• id（可选）：消息唯一标识，用于断点续传；
• retry（可选）：客户端断线重连间隔时间。

SSE 数据示例：

event: foo
data: a foo event

data: an unnamed event

event: end
data: a bar event

4.5.2 LangChain 流式传输完整流程

LangChain 不自定义底层传输协议，而是依托大模型服务商SSE能力 + 自身消息格式封装实现统一流式体验，兼容OpenAI、DeepSeek等所有主流模型。

4.5.2.1 核心源码流程（以OpenAI为例）

LangChain 通过 BaseChatOpenAI 类的_stream() 方法实现流式请求与数据处理，核心流程如下：

def _stream(
    self,
    messages: list[BaseMessage],
    stop: Optional[list[str]] = None,
    run_manager: Optional[CallbackManagerForLLMRun] = None,
    *,
    stream_usage: Optional[bool] = None,
    **kwargs: Any,
) -> Iterator[ChatGenerationChunk]:
    # 1. 强制开启流式模式
    kwargs["stream"] = True
    stream_usage = self._should_stream_usage(stream_usage,** kwargs)

    # 2. 构建API请求参数
    payload = self._get_request_payload(messages, stop=stop, **kwargs)
    default_chunk_class: type[BaseMessageChunk] = AIMessageChunk
    base_generation_info = {}

    # 3. 发起流式API请求
    if "response_format" in payload:
        response_stream = self.root_client.beta.chat.completions.stream(**payload)
        context_manager = response_stream
    else:
        response = self.client.create(**payload)
        context_manager = response

    # 4. 逐块处理SSE响应数据
    try:
        with context_manager as response:
            for chunk in response:
                # 原始API数据转换为LangChain统一消息块
                generation_chunk = self._convert_chunk_to_generation_chunk(
                    chunk,
                    default_chunk_class,
                    base_generation_info if is_first_chunk else {},
                )
                if generation_chunk is None:
                    continue
                # 触发token生成回调
                if run_manager:
                    run_manager.on_llm_new_token(
                        generation_chunk.text,
                        chunk=generation_chunk,
                        logprobs=logprobs,
                    )
                # 逐块产出流式数据
                yield generation_chunk
    # 5. 捕获处理API异常
    except openai.BadRequestError as e:
        _handle_openai_bad_request(e)

4.5.2.2 底层协议与数据转换原理

1. 请求协议： LangChain 基于 OpenAI 官方 Python SDK 的 HTTP 客户端发起标准 HTTP 请求，无自定义私有协议，兼容性极强。
2. 流式开启逻辑： 请求参数中强制设置 stream=True，告知模型服务端启用 SSE 流式推送，服务端保持长连接并逐段返回数据块。
3. 原始SSE数据格式： 模型服务端返回标准 SSE 结构化数据，示例片段：

data:{
"id": "chatcmpl-123", 
"object": "chat.completion.chunk", 
"created": 1717500000, 
"model": "gpt-4o-mini",
"choices": [{"index": 0, "delta": {"role": "assistant", "content": "你好"},
"finish_reason": null}]
}

4. 数据统一封装： 通过 _convert_chunk_to_generation_chunk()、_convert_delta_to_message_chunk() 方法，将不同模型厂商的原生 SSE 数据，统一转换为 LangChain 标准的AIMessageChunk 消息块，实现多模型流式接口统一。

4.5.3 流式传输核心总结

1. 底层依赖： 依托模型厂商 SSE 协议实现流式推送，LangChain 仅做请求封装与数据标准化；
2. 核心方法： stream() 同步阻塞流式、astream() 异步非阻塞流式，适配不同业务场景；
3. 数据规范： 统一输出 AIMessageChunk 消息块，支持拼接、解析、自定义切割；
4. 链路扩展： 结合 LCEL、StrOutputParser、自定义生成器，可灵活实现各类定制化流式输出效果。