LangChain 能力详解

0.1 裸调模型的痛点（为什么要这层抽象）#

• 幻觉、上下文窗口有限、输出不定型、无法算数、无法访问库表与实时数据 → 都属于「模型是封闭知识系统」的连带后果。
• 工具调用解决的正是最后两条；RAG 解决的是知识陈旧与私有数据。

0.2 RAG 知识地图（后面所有组件都挂在这张图上）#

• 六个环节：加载 → 分割 → 嵌入 → 存储 → 检索 → 生成。
• 离线那三步是一次性的（建库）；在线那两步是每次提问都跑的（查询）。
• 我不把它当流程图背，而是当排错顺序用：答案不对，先看切分粒度，再看检索命中，最后才怪模型。

0.3 组件速查#

0.4 环境与依赖#

运行环境：Python 3.13 + 一套 requirements.txt（pip install -r requirements.txt 批量装）。

第一批（前面章节够用）

第二批（学到向量存储后半段才需要）：pinecone==7.3.0、langchain-pinecone==0.2.12。

坑（很重要，我踩过）：langchain-pinecone 会连带装上 simsimd，装上之后 Redis 向量存储的 MMR 搜索会失效，官方还没修。所以顺序上先把 Redis 部分跑完再装 Pinecone；两条路线别在同一环境里混着验收。

坑：早期 0.2.3 版对 Python 3.13 支持可能失败，因为它的 numpy 版本区间（>=1.26.4,<2.0.0）与 3.13 不兼容。

密钥类环境变量

0.5 全篇路线图#

内容与目标#

• LangChain 是围绕大模型构建应用的框架：把一条 AI 应用流程拆成标准化组件，让你自由组合。
• 目标：跑通「定义模型 → 组织消息 → 调用 → 解析」这条最短链路，并理解 | 的含义。

详细过程#

六步走完一次完整调用，顺序别乱。

步骤1：申请 API key 并配置环境变量#

• 先申请密钥，再落到系统环境变量，不要硬编码进代码（避免泄露 + 方便换号）。
• 变量名用 OPENAI_API_KEY。
• 创建密钥的界面操作：填 Name（示例里写 LC）→ Project 保持默认 → 点 Create secret key → 立刻复制保存。
• 注意：密钥只在创建时完整显示一次；丢了只能重新生成。

步骤2：定义大模型#

1
pip install -U langchain-openai

1
from langchain_openai import ChatOpenAI
2

3
model = ChatOpenAI(model="gpt-5-mini")

• 装好对应包，传个模型名，就拿到模型实例。

步骤3：定义消息列表#

1
from langchain_core.messages import HumanMessage, SystemMessage
2

3
messages = [
4
    SystemMessage(content="把下面这句英文翻译成中文"),
5
    HumanMessage(content="hi!"),
6
]

• SystemMessage 管行为设定，放列表首位；HumanMessage 装用户输入。
• 合起来是一个消息列表 —— 聊天接口吃的是列表，不是单串文字（这点后面回灌工具结果时会再撞上）。

步骤4：调用大模型#

1
result = model.invoke(messages)

• 模型是 Runnable 实例，invoke 就完事。
• 返回 AIMessage，不是字符串。

参数说明：AIMessage 的四个字段

• 示例结果：content='你好!'，finish_reason 为 stop，合计 22 token（入 20、出 2）。

步骤5：输出解析#

1
from langchain_core.output_parsers import StrOutputParser
2

3
parser = StrOutputParser()
4
print(parser.invoke(result))   # 你好!

• StrOutputParser 干的事：把消息对象降级成纯字符串。
• 注意它属于 langchain_core.output_parsers。

步骤6：链式执行#

1
chain = model | parser
2
print(chain.invoke(messages))

• 先定义组件，再链接，一次 invoke 走完全程。
• 结果与「先 model.invoke 再 parser.invoke」完全一致。

引出 LangChain 相关概念#

Runnable 接口

Runnable 是所有组件的基础接口，规定一个节点至少能做到五件事：

• 聊天模型是 Runnable，输出解析器也是 Runnable，所以两者都能 invoke。
• 关键点：组合出来的链本身还是 Runnable，所以能无限套下去。

LangChain Expression Language

• LCEL 是声明式的写法：用已有 Runnable 拼出一个新的 Runnable。
• model | parser、RunnableSequence(first=model, last=parser)、model.pipe(parser) 三种写法等价。

1
from langchain_core.runnables import RunnableSequence
2

3
chain = model | parser                                  # ① 重载 |
4
chain = RunnableSequence(first=model, last=parser)      # ② 显式构造
5
chain = model.pipe(parser)                              # ③ 方法调用

• 规则只有一条：上一个节点的输出 = 下一个节点的输入。
• 拼出来的对象类型叫 RunnableSequence，也就是「链」；它自己仍是 Runnable。
• 这套手感迁移自 Unix 管道：小工具各做一件事，靠管道拼出复杂行为。

结论：理解了「输出即输入」，后面所有组合方式（RunnablePassthrough 让输入原样穿过、字典字面量把同一输入并行分发给多分支）都是这句话的推论。

概念说明：Runnable 不是某个具体类，而是一份契约 —— 谁实现了它，谁就能进管道。聊天模型、提示模板、输出解析器、检索器、工具包装，都是在履行这份契约。

我踩过的坑：| 两侧必须都是 Runnable。普通函数直接塞进去会报错，得先包成 RunnableLambda 或者用 @chain 装饰。

1. 定义聊天模型#

先分清两个概念，不然看文档会飘：

• 现在绝大多数模型都走聊天接口：进消息列表、出消息对象，而不是裸文本。

1.1 通过 API 定义聊天模型#

方式1：ChatOpenAI#

1
from langchain_openai import ChatOpenAI
2

3
model = ChatOpenAI(
4
    model="gpt-5-mini",
5
    temperature=0,
6
    max_tokens=None,
7
    timeout=None,
8
    max_retries=2,
9
    # api_key=..., base_url=..., organization=...
10
)

ChatOpenAI 是 OpenAI 聊天模型的具体实现，父类是 BaseChatOpenAI。

ChatOpenAI 常用初始化参数说明

• 想用国产模型：只要对方兼容 OpenAI 协议，改 base_url + 换密钥即可，ChatOpenAI 不用换。

1
import os
2
from langchain_openai import ChatOpenAI
3

4
model = ChatOpenAI(
5
    model="deepseek-chat",
6
    base_url="https://api.deepseek.com/v1",
7
    openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
8
)

坑：base_url 里的 /v1 是接口版本路径，与模型版本无关；写错前缀直接 404。另外 DeepSeek 的密钥要单独申请、单独配变量。

1.1.2 invoke() 调用#

1
abstractmethod invoke(input: Input, config: RunnableConfig | None = None, **kwargs) -> Output

请求参数

RunnableConfig 里值得记的字段：

• configurable：运行时可改的属性（键值对），动态换模型就靠它。
• run_id：本次运行标识，缺省自动生成 UUID。
• run_name：默认取类名，日志里用它区分。
• metadata：字符串键、JSON 值的字典，附加到这次运行上。

返回值：AIMessage（或其它 BaseMessage 子类）。

方式2：init_chat_model#

1
from langchain.chat_models import init_chat_model
2

3
gpt = init_chat_model("gpt-5-mini", model_provider="openai", temperature=0)
4
ds = init_chat_model("deepseek-chat", model_provider="deepseek", temperature=0)
5

6
print(gpt.invoke("what's your name").content)     # ChatGPT
7
print(ds.invoke("what's your name").content)      # DeepSeek Chat

init_chat_model() 常用参数说明

• 支持的 model_provider 常见取值：openai、anthropic、google_genai、ollama、deepseek（openai 对应 langchain-openai 包）。
• 因为所有集成实现的都是同一套 ChatModel 接口，所以换厂商只是换字符串。

init_chat_model() 返回值说明

• 返回对应 provider 的 BaseChatModel 子类实例：
  • model_provider="openai" → langchain_openai.ChatOpenAI
  • model_provider="deepseek" → langchain_deepseek.ChatDeepSeek
  • 其它厂商同理（如 ChatAnthropic）。
• 开了 configurable_fields 时，先拿到的是一个「可配置模型模拟器」，传入配置后才真正建模型。

示例1：基本用法

1
gpt_model = init_chat_model("gpt-5-mini", model_provider="openai", temperature=0)
2
deepseek_model = init_chat_model("deepseek-chat", model_provider="deepseek", temperature=0)

问它是谁：一个答 ChatGPT，一个答 DeepSeek Chat。

示例2：创建可配置模型

1
configurable_model_1 = init_chat_model(temperature=0)
2

3
configurable_model_1.invoke(
4
    "what's your name",
5
    config={"configurable": {"model": "gpt-5-mini"}},
6
)

• 初始化时只给 temperature，模型名留到运行时用 configurable.model 指定。

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

1
configurable_model_2 = init_chat_model(
2
    model="gpt-5-mini",
3
    temperature=0,
4
    configurable_fields=("model", "model_provider", "temperature", "max_tokens"),
5
    config_prefix="first",
6
)
7

8
configurable_model_2.invoke(
9
    "what's your name?",
10
    config={"configurable": {
11
        "first_model": "deepseek-chat",
12
        "first_temperature": 0.5,
13
        "first_max_tokens": 100,
14
    }},
15
)

• 有了 config_prefix="first"，运行时键名就变成 first_model / first_temperature / first_max_tokens。
• 实用价值：一套代码里放多个可切换的模型位（快模型/强模型），不必改代码。

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

1.2.1 ChatOllama#

1
from langchain_ollama import ChatOllama
2

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

ChatOllama 常用初始化参数说明

• 示例输出里 content 是空的（思考内容走独立字段）、usage_metadata 显示 8/44/52。
• 场景：内网、离线、数据不外出的环境，用它顶替 ChatOpenAI；它同样实现 Runnable 接口，所以链的写法一字不改。

2. 聊天模型—调用工具#

为什么需要工具：

• 模型是封闭知识系统：训练数据有截止日、参数不能改，让它「算 9×6」或「查昨天的天气」都不可靠。
• 收益四条：扩能力边界、保实时性、拆解复杂任务、接入既有系统。

机制一句话：模型不执行任何东西，它只产出「要调用哪个工具、参数是什么」；执行由你的代码完成。

• 天气例子：模型不直接答，而是给出 tool_calls，其中 name 为 get_weather、args 含 location 为 Hawaii、id 为 call_abc123、type 为 tool_call。
• 查库例子四步：建工具 → 绑工具 → 模型给出调用参数（min_price=10.0、max_price=50.0、category="electronics"）→ 代码执行取结果。

2.1 创建工具#

三种造法，从最省事的开始。

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

1
from langchain_core.tools import tool
2

3
@tool
4
def multiply(a: int, b: int) -> int:
5
    """Multiply two integers.
6

7
    Args:
8
        a: First integer
9
        b: Second integer
10
    """
11
    return a * b
12

13
multiply.invoke({"a": 2, "b": 3})   # 6
14
multiply.name          # 'multiply'
15
multiply.description   # 取自文档字符串
16
multiply.args          # {'a': {'title': 'A', 'type': 'integer'}, ...}

• @tool 是最省事的做法：函数名即工具名，文档字符串即描述。
• 进 schema 的三样东西：函数名 + 类型提示 + 文档字符串，缺一不可；描述写得含糊，模型选错工具的概率立刻上升。
• 参数必须需要指定变量名和变量类型（类型提示），否则 schema 推断不出来。
• 简单工具靠解析 Google 风格文档字符串取参数描述，关键字是 Args: 与 Returns:。
• 实际打印出来的 args 是 JSON Schema 片段，形如 {'a': {'title': 'A', 'type': 'integer'}}。
• 天气类的工具示例里，返回字段可以细到「天气 / 最高气温 / 最低气温」这一级。

什么是 Schema？

• Schema = 描述其它数据结构的声明格式。同一份人物数据可以有两种组织方式（扁平 vs 分组），差异就是 schema 差异。
• 更正式的例子：把「姓名」拆成姓与名，「地址」再拆成省/市/区/国，日期统一成 ISO 格式 —— 这些都是 schema 层面的约定。
• JSON Schema 的能力边界：能描述表面结构并自动校验（type / properties / format），但表达不了元素之间的约束关系。
• 所以复杂格式的校验常分两阶段：结构级（schema 干）+ 语义级（通用编程语言干）。

1
from pydantic import BaseModel, Field
2
from langchain_core.tools import tool
3

4
class MultiplyInput(BaseModel):
5
    """Multiply two integers."""
6
    a: int = Field(..., description="First integer")
7
    b: int = Field(..., description="Second integer")
8

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

1
from typing_extensions import Annotated
2
from langchain_core.tools import tool
3

4
@tool
5
def add(
6
    a: Annotated[int, ..., "First integer"],
7
    b: Annotated[int, ..., "Second integer"],
8
) -> int:
9
    """Add two integers."""
10
    return a + b

2.1.2 使用 StructuredTool 类提供的函数创建工具#

1
from langchain_core.tools import StructuredTool
2

3
calculator_tool = StructuredTool.from_function(func=multiply)
4
calculator_tool.invoke({"a": 2, "b": 3})   # 6

• StructuredTool.from_function 是类方法：吃一个普通函数，吐一个工具对象。
• 关键参数说明：

• **kwargs 继续向后透传。

示例1：常规用法

• 最小示例：函数直接交给 func，工具名与描述从函数自动继承；invoke 收字典入参，两数同乘打印 6。

示例2：加入配置，依赖 Pydantic 类

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

5
calculator_tool = StructuredTool.from_function(
6
    func=multiply,
7
    name="Calculator",
8
    description="两数相乘",
9
    args_schema=CalculatorInput,
10
)

• 函数没有说明时，用 name、description、args_schema 三项补齐元信息。
• tool.args 返回的是 JSON Schema 片段，含 description、title、type 键。

示例3：加入 response_format 配置

1
from typing import List, Tuple
2

3
def multiply(a: int, b: int) -> Tuple[str, List[int]]:
4
    nums = [a, b]
5
    return f"{nums}相乘的结果是{a * b}", nums
6

7
calculator_tool = StructuredTool.from_function(
8
    func=multiply, name="Calculator", response_format="content_and_artifact",
9
)

• 默认 'content'：工具结果只落在 ToolMessage.content。
• 配成 'content_and_artifact'：函数必须返回二元组，分别对应 content 与 artifact。
• 两者的分工：

• 只留 content 的代价：丢掉来源链接、原始返回和排查线索，也不方便留档分析。
• 想拿到元组，得把请求伪装成模型的调用形态（name / args / id / type 四项，id 与 type 不可省）；返回体形如 content、name、tool_call_id、artifact 四字段并列。
• @tool 装饰器同样认 response_format 这个开关。

2.2 绑定工具#

1
from langchain_openai import ChatOpenAI
2

3
model = ChatOpenAI(model="gpt-4o-mini")
4
model_with_tools = model.bind_tools([multiply, add])

• bind_tools 的 tools 可以是字典、Pydantic 类、普通函数或 BaseTool 实例。
• 返回值是一个 Runnable：可以接 PromptValue、字符串或消息列表；输出是 AIMessage（其中可能夹带工具调用）。
• 关键参数：

2.3 工具调用#

1
result = model_with_tools.invoke("9乘6等于多少？")
2
result.tool_calls
3
# [{'name': 'multiply', 'args': {'a': 9, 'b': 6}, 'id': 'call_...', 'type': 'tool_call'}]

• 典型回包：content 为空，调用信息藏在 additional_kwargs 与 tool_calls 里。
• finish_reason 为 tool_calls（表示「我要调工具」）；response_metadata 里带 prompt/completion/total 三个 token 数。
• 换一句无关的话（hello world!）：模型正常闲聊，finish_reason 变成 stop —— 说明调不调工具由相关性决定，不是绑了就必调。

2.4 强制模型调用工具#

1
model_with_tools = model.bind_tools([add, multiply], tool_choice="any")

• 同一句问候在强制模式下也产出了 add 调用，参数被模型自拟为 1 与 2。
• 用途：做「必须走工具」的确定性流程，或者调试工具 schema 是否可用。

2.5 工具属性#

• 只要发生调用，返回对象上就有 tool_calls，里面是名字、入参、调用编号（name / args / id / type）。
• 它是纯 Python 数据（列表套字典），可以直接循环取用，不需要解析文本 —— 这就是结构化工具调用相对「让模型输出 JSON 再正则抠」的最大好处。

2.6 将工具输出传递给聊天模型#

拿到调用请求只是半程；答案要把工具产物回灌给模型才生成。

1
from langchain_core.messages import HumanMessage
2

3
messages = [HumanMessage("9乘6等于多少？5加3等于多少？")]
4
ai_msg = model_with_tools.invoke(messages)
5
messages.append(ai_msg)
6

7
for tool_call in ai_msg.tool_calls:
8
    selected_tool = {"add": add, "multiply": multiply}[tool_call["name"].lower()]
9
    messages.append(selected_tool.invoke(tool_call))   # @tool 工具可直接吃整个 tool_call 字典
10

11
result = model.invoke(messages)     # 54 与 8

要点：

1. 聊天接口吃消息列表，所以工具产物必须先包成 ToolMessage。
2. 一轮对话里需要凑齐三类消息：用户提问（HumanMessage）、带调用的 AI 回复（AIMessage）、工具结果（ToolMessage）。
3. 按名字挑函数时做小写化比较 → 对大小写不敏感。
4. @tool 造出的工具支持把整个 tool_call 字典丢进 invoke，直接得到 ToolMessage（内部自动补 tool_call_id）。
5. 示例里 ai_msg 携带两条调用（9×6 与 5+3），于是产生两条 ToolMessage。
6. 两趟的 token 统计可对照：第二趟输入明显变长，输出很短。

关键字段对照

2.7 LangChain 提供的工具#

• 官方既备了单件工具，也备了成套工具包（Toolkit）；搜索、数据库、浏览器等方向都有。
• 所有工具都落在 BaseTool 与 BaseToolkit 两个基类之下，找类名可以按这两个基类全局检索。
• 分类覆盖：Code Interpreter、Productivity、Web Browsing、Database、Finance 等。
• 同是搜索类的返回字段粒度不同：多数给 URL / Snippet / Title，SerpApi 只给 Answer，Jina 还带正文；免费与收费并存（例如 DuckDuckGoSearch 免费）。
• 集成包命名有规律：单个集成常落在 langchain_community.tools 下（如 tavily_crawl.py、azure_ai_services.py 这类模块名），也有独立包（如 langchain-tavily）。

2.7.1 TavilySearch#

• 定位：面向智能体检索场景的搜索工具，返回结构化结果，便于直接进入推理与生成环节。
• 申请：官网注册后拿密钥，免费额度按积分计（每月 1000），key 常以 tvly-dev 开头；控制台带 Playground 可试搜，也有预置案例（Research Assistant / Chat / Crawl2RAG 等）。

1
from langchain_tavily import TavilySearch
2

3
tool = TavilySearch(max_results=4)      # 最多取回 4 条
4
model_with_tools = model.bind_tools([tool])
5

6
ai_msg = model_with_tools.invoke("西安今天天气怎么样")
7
tool_msg = tool.invoke(ai_msg.tool_calls[0])
8
result = model_with_tools.invoke([*messages, ai_msg, tool_msg])

参数说明

• 接入三步：装包（langchain-tavily）→ 配环境变量 TAVILY_API_KEY（Windows 下写进系统变量）→ 代码里实例化并 bind_tools。
• 示例结果：问西安天气，工具跑完后模型用中文汇总，并附上气象局链接。
• 注意：回灌之后仍用带工具的实例发第二次请求；这与 2.6 里用裸模型发第二趟略有差别（带工具时理论上还能继续发起新调用）。
• 也可以用官方客户端 tavily.TavilyClient 直接调 tavily_client.search(query) —— 适合不需要模型决策、写死流程的场景。

3. 聊天模型—结构化输出#

要解决的问题：从「字符串」到「对象」的范式转换。

• 没有它时：AIMessage.content 是一段人读着舒服的文本，程序读着难受；想抠出「公司名」和「股价变化」，只能上正则，脆弱又费事。
• 有它之后：先声明期望结构，再要求模型必须按这个结构返回。
• 典型动机：输出要落库，就得对齐库表结构。

3.1 with_structured_output()#

三步走：

1
# 1. 定义输出结构
2
class Person(BaseModel):
3
    name: str
4
    age: int
5

6
# 2. 绑定 schema，其实是生成一个支持结构化返回的 Runnable
7
structured_model = model.with_structured_output(Person)
8

9
# 3. 执行
10
result = structured_model.invoke("张小红今年 18 岁")
11
# Person(name='张小红', age=18)

• 这是获得结构化输出最简单、最可靠的方法。
• 输出结构可以是 TypedDict、JSON Schema 或 Pydantic 类：
  • TypedDict / JSON Schema → 返回字典
  • Pydantic 类 → 返回 Pydantic 对象

请求参数：

method 三选一

返回值： 返回一个 Runnable 实例。执行时：

• include_raw=False：Pydantic schema → Pydantic 对象；其它 schema → 字典。解析失败直接抛错。
• include_raw=True：返回带三个键的字典 ——

3.1.1 返回 Pydantic 对象#

1
from pydantic import BaseModel, Field
2

3
class Movie(BaseModel):
4
    """一部电影的信息。"""
5
    title: str = Field(description="电影标题")
6
    year: int = Field(description="上映年份")
7
    director: "Director" = Field(description="导演")
8

9
class Director(BaseModel):
10
    name: str
11
    country: str
12

13
structured_model = model.with_structured_output(Movie)

• 内部流程：抽出代表 Pydantic 参数的 JSON → 用 Pydantic 模型解析与校验 → 转成可用的对象实例返回。
• 支持嵌套输出（上面的 Movie 里套 Director），也支持列表、联合类型。

3.1.2 返回 TypedDict#

1
from typing_extensions import TypedDict
2

3
class Person(TypedDict):
4
    name: str
5
    age: int
6

7
structured_model = model.with_structured_output(Person)
8
structured_model.invoke("张小红 18 岁")    # {'name': '张小红', 'age': 18}

• TypedDict 的价值：给字典补上精确的类型提示 —— 键名拼错、类型写错，类型检查器当场就报，不必等到运行时。
• 想同时看原始响应，就加 include_raw=True，拿到的就是 raw / parsed / parsing_error 三键字典。

3.1.3 返回 JSON#

1
json_schema = {
2
    "title": "Person",
3
    "type": "object",
4
    "properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
5
    "required": ["name", "age"],
6
}
7

8
structured_model = model.with_structured_output(json_schema)

• 直接给 JSON Schema 也能跑，适合 schema 由外部配置下发的场景。

3.1.4 选择输出格式#

• 做法：建一个带联合类型属性的父模式，让模型自己选其中一种。

1
from typing import Literal, Union
2

3
class Action(BaseModel):
4
    action: Literal["respond", "search"]
5
    response: Union[Response, SearchQuery]
6

7
structured_model = model.with_structured_output(Action)

• 价值：把「要不要检索」「回不回答」这类分支交给模型判，后面接链时按 action 路由即可。

3.2 实用场景#

结构化输出真正落地的地方就这三处，逐个过一遍。

3.2.1 场景1：作为信息提取器#

1
class Person(BaseModel):
2
    name: str
3
    age: int
4

5
structured_model = model.with_structured_output(Person)
6
structured_model.invoke("张小红今年 18 岁，在杭州读大学")

• 最常用的一类：把非结构化文本（简历、工单、聊天记录）洗成字段。
• 和 RAG 结合时，它常常是「抽元数据」的那一环。

3.2.2 场景2：使用”少样本提示”来增强信息提取能力#

• 光给 schema，模型对边界情况（多值、缺值、「无」怎么写）理解会飘。
• 加几个「输入 → 期望输出」的示例对之后，抽取稳定很多。
• 具体写法见 少样本提示 一节的案例二。

3.2.3 场景3：与工具结合使用#

注意：把聊天模型的原生工具和结构化输出凑一块用，体验很差；这里只看能力边界。

方式1：使用 with_structured_output() 的 tools 参数#

1
structured_model = model.with_structured_output(
2
    SearchResult, method="json_schema", strict=True, include_raw=True, tools=[search_tool],
3
)
4
out = structured_model.invoke("搜索 LangChain 的最新进展")

• 结果：它不会直接给你想要的结构化结果，而是只返回一个带工具调用信息的 AIMessage。
• 原因：它只负责把「有哪些工具可用」告诉模型，执行与否它一概不管。
• 要拿到「工具执行结果 + 最终结构化输出」，只能手动处理工具调用，再把结果回灌给模型 —— 也就是下一种做法。

方式2：拆解能力，单独依次执行#

顺序不能弄反：① 先绑定工具 → ② 再添加结构化输出。

1
model_with_tools = model.bind_tools([search_tool])           # ① 先工具
2
structured_model = model.with_structured_output(SearchResult) # ② 再结构化输出
3

4
messages = [HumanMessage("搜索并总结 LangChain 的最新进展")]
5
ai_msg = model_with_tools.invoke(messages)     # 第一趟：回来的是调用请求
6
messages.append(ai_msg)
7
for call in ai_msg.tool_calls:
8
    messages.append(search_tool.invoke(call))  # 执行工具，手动把结果加入消息列表
9
result = structured_model.invoke(messages)     # 第二趟：拿结构化结果

• 若不回灌工具结果，模型只会凭自己直接给答案（示例里它真这么干了）。
• 这么写实际调用了两次模型：第一次产工具调用，第二次带工具结果产结构化输出。
• 想「一次调用拿到结构化结果 + 工具执行」→ 那是 Agent 的活（LangGraph 部分）。

4. 聊天模型—流式传输#

• invoke 是非流式：模型想得越久，用户干等越久（示例里一个笑话等了约 20 秒）。
• 流式 = 结果没写完就先往外吐，等待感立刻下降。

4.1 stream() 同步传输#

1
from langchain_openai import ChatOpenAI
2

3
model = ChatOpenAI(model="gpt-4o-mini")
4
chunks = []
5
for chunk in model.stream("讲一个 50 字的笑话"):
6
    chunks.append(chunk)
7
    print(chunk.content, end="|", flush=True)

• .stream() 返回迭代器，可以 for 遍历；flush=True 保证即时输出。
• 块对象叫 AIMessageChunk（完整消息的一个片段）。
• 分隔符会暴露切块粒度：有时一个字一块，有时几个字一块。
• 返回结构与 AIMessage 一致：content 是文本，additional_kwargs 与 response_metadata 默认空字典；id 形如 run--…。
• 块可以相加：chunks[0] + chunks[1] + … 能合并成一条消息。

坑：末块常常是空字符串，拼接时会多出一个分隔符。

4.2 astream() 异步传输#

流式场景优先用异步：主流程别被 I/O 堵住。

4.2.1 异步相关概念#

什么是协程？

• 用 async def 定义，在需要停顿的地方 await，稍后从原地恢复。
• 切换由程序与语言掌控，属于用户态协作式调度。

什么是事件循环？

• asyncio 的核心，相当于一个维护待办清单的调度员。
• 机制：任务等 I/O 时先挂起 → 转去跑就绪任务 → 条件满足再恢复。

如何运行？

1
import asyncio, time
2

3
def boil_water():              time.sleep(5)          # 同步：烧水 5 秒
4
def send_message():            time.sleep(2)          # 同步：发消息 2 秒
5
async def boil_water_async():  await asyncio.sleep(5)
6
async def send_message_async(): await asyncio.sleep(2)
7

8
async def main():
9
    task1 = asyncio.create_task(boil_water_async())
10
    task2 = asyncio.create_task(send_message_async())
11
    await task1
12
    await task2
13

14
asyncio.run(main())     # 总耗时 5 秒，而不是 7 秒

• 同步版：烧水 + 发消息串行 = 7 秒，等待期间 CPU 完全空闲。
• 异步版：两段等待重叠 = 5 秒。
• create_task 把协程包成任务交给循环，再逐个 await 收尾；asyncio.run(main()) 负责建循环并把主协程投进去。
• 这段示例放在脚本里跑（比如 xxx.py）时，才能看出同步与异步的耗时差别。

坑：await 只是让出控制权，不是开线程。CPU 密集任务照样会卡住整个循环。

4.2.2 使用#

1
import asyncio
2
from langchain_openai import ChatOpenAI
3

4
model = ChatOpenAI(model="gpt-4o-mini")
5

6
async def async_stream():
7
    async for chunk in model.astream("讲一个 50 字的笑话"):
8
        print(chunk.content, end="|", flush=True)
9

10
asyncio.run(async_stream())

• 行为与同步流式一致，只是消费方式变成 async for，且必须写在 async 函数里。
• 坑：异步迭代器不能直接用普通 for；漏写 async 关键字会直接报错。

4.3 使用 StrOutputParser 解析模型的输出#

• 流式不是聊天模型专属：任何 Runnable 都支持。
• 但部分组件不做流式，例如检索器就是同步阻塞型的。
• 块的类型随被流式的组件而变：聊天模型吐的是消息块，其它组件可能是别的。
• StrOutputParser 从消息块里抽出 content，于是链上流动的就是纯字符串令牌。

1
from langchain_core.output_parsers import StrOutputParser
2

3
parser = StrOutputParser()
4
chain = model | parser
5
for token in chain.stream("写一段关于爱情的歌词，需要 5 句话"):
6
    print(token, end="|", flush=True)

• 链本身还是 Runnable，所以能直接 .stream()。

4.4 自定义流式输出解析器#

目标：把「逐字」输出改成「逐句」输出，同时保留流式。

1
from typing import Iterator, List
2

3
def split_into_list(input: Iterator[str]) -> Iterator[List[str]]:
4
    buffer = ""
5
    for chunk in input:
6
        buffer += chunk
7
        while "。" in buffer:
8
            stop_index = buffer.index("。")
9
            yield [buffer[:stop_index].strip()]
10
            buffer = buffer[stop_index + 1:]
11
    yield [buffer.strip()]        # 收尾：别漏
12

13
chain = model | parser | split_into_list

坑：循环结束后残留缓冲必须再 yield 一次，否则最后一句丢失。 坑：缓冲末尾会多出一个空字符串块；模型不按句号断句时，切分规则直接失效。

4.5 深度探索流式传输#

4.5.1 SSE 协议介绍#

• HTTP 是无状态「请求—响应」模式，服务端本来没法主动推消息。
• SSE 让服务端先声明「后面发的是流消息」，客户端保持连接等新数据。
• 浏览器用内置 EventSource 接收，逐条触发 onmessage。

核心特点

数据格式

1
Content-Type: text/event-stream
2
Connection: keep-alive
3

4
data: 第一段
5
event: message
6
id: 1
7
retry: 3000
8

9
data: 第二段

• 报文由若干 message 组成，message 之间用空行分隔，每行写一个 [field]: value。
• 以冒号开头的行按注释处理。

坑：字段名与值之间是「冒号 + 空格」，漏换行或多写空行都会让解析失败。

4.5.2 LangChain 流式传输流程分析#

• LangChain 不定义底层传输协议；协议来自模型供应商与所用 Web 框架，LangChain 只负责接收并整理成消息块。

1
def _stream(self, messages, stop=None, run_manager=None, *,
2
            stream_usage=None, **kwargs) -> Iterator[ChatGenerationChunk]:
3
    kwargs["stream"] = True                      # ① 强制开流式
4
    stream_usage = self._should_stream_usage(stream_usage, **kwargs)
5
    payload = self._get_request_payload(messages, stop=stop, **kwargs)   # ② 组装请求体
6
    default_chunk_class = AIMessageChunk
7
    response_stream = self.root_client.beta.chat.completions.stream(**payload)  # ③a 结构化分支
8
    response = self.client.create(**payload)                                    # ③b 普通分支
9
    generation_chunk = self._convert_chunk_to_generation_chunk(                  # ④ 转生成块
10
        chunk, default_chunk_class, base_generation_info)
11
    run_manager.on_llm_new_token(generation_chunk.text, chunk=generation_chunk, logprobs=logprobs)
12
    yield generation_chunk

1
def _convert_chunk_to_generation_chunk(self, chunk, default_chunk_class, base_generation_info):
2
    choices = chunk.get("choices", []) or chunk.get("chunk", {}).get("choices", [])
3
    message_chunk = _convert_delta_to_message_chunk(choice["delta"], default_chunk_class)
4
    return ChatGenerationChunk(message=message_chunk, generation_info=None)

5. 使用 LangSmith 跟踪 LLM 应用#

• 定位：为生产级 LLM 应用提供监控与评估的平台；与框架无关，能配 langchain / langgraph，也能独立用。
• 触发场景：链、代理、模型调用一多，就必须能看每一步的输入输出与耗时。

申请 API Key（界面步骤）

打开跟踪

1
LANGSMITH_TRACING="true"
2
LANGSMITH_API_KEY="<申请到的 Key>"

• Windows 下写进系统变量。
• 坑：环境变量改完要重启终端或 IDE，旧进程读不到新值。

能看到什么

• 常见步骤类型：RunnableSequence（就是链）、ChatOpenAI（模型调用）、RunnableLambda（函数包装）。
• 坑：顶层步骤不一定叫「序列」，名称随实际调用结构变化。

结论：业务代码一行不改，只配环境变量，就能拿到全链路监控。

1. 消息（Messages）#

1.1 LLM消息结构#

每条消息大致由三部分组成：角色 + 内容 + 附加元数据；元数据字段各家厂商不一样。

• content 可以是纯文本，也可以是图像/音频/视频等字典列表（多模态支持仍有限）。
• 元数据四项：ID、Name、Metadata、Tool Calls。
  • Name 用于区分同角色的不同实体，仅部分模型支持。
• 输入可以直接用 OpenAI 风格的 dict 列表，invoke 同样吃：

1
model.invoke([{"role": "user", "content": "Hello, how are you?"}])

1.2 LangChain消息#

• 提供跨厂商统一的消息格式：换模型不用改消息写法。
• 用 init_chat_model + model_provider 就能切厂商：

1
from langchain.chat_models import init_chat_model
2

3
init_chat_model("gpt-4o-mini", model_provider="openai")
4
init_chat_model("claude-3-5-sonnet-latest", model_provider="anthropic")
5
init_chat_model("deepseek-chat", model_provider="deepseek")
6
init_chat_model("gemini-2.5-flash", model_provider="google_genai")

五类消息

• 五类都是 BaseMessage（langchain_core.messages.base）的子类，同时充当模型的入参与出参。

1.2.1 BaseMessage抽象消息类#

1.2.2 对话模式#

• 常规序列：System 打头，之后 Human → AI 成对循环。
• 带工具的一轮：Human → AI（发起调用）→ Tool（回结果）→ AI（最终答复）。

坑：顺序非法会让模型理解偏差 —— ToolMessage 必须紧跟在发起该调用的 AI 消息之后。

1.3 缓存历史消息#

1.3.1 多轮对话#

1
from langchain_core.messages import HumanMessage, AIMessage
2

3
model.invoke([HumanMessage("Hi! I'm Bob")])
4
model.invoke([
5
    HumanMessage("Hi! I'm Bob"),
6
    AIMessage("Hello Bob! How can I assist you today?"),
7
    HumanMessage("What's my name?"),
8
])

• 只发一条 HumanMessage，再问名字时模型答不上来。
• 把先前的 AI 回复和新问题一起送过去，就有多轮效果。
• 结论：模型不保存状态，历史列表必须由调用方自己维护。

1.3.2 内存缓存#

RunnableWithMessageHistory 把某个 Runnable 包起来，自动跟踪输入输出并落到某个存储里。

1
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
2
from langchain_core.runnables.history import RunnableWithMessageHistory
3

4
store = {}
5

6
def get_session_history(session_id: str) -> BaseChatMessageHistory:
7
    if session_id not in store:
8
        store[session_id] = InMemoryChatMessageHistory()
9
    return store[session_id]
10

11
with_message_history = RunnableWithMessageHistory(model, get_session_history)
12
config = {"configurable": {"session_id": "1"}}
13
with_message_history.invoke([HumanMessage("Hi! I'm Bob")], config=config)

1.4 管理历史消息#

所谓「管理」，本质是对消息做增删改查。

1.4.1 前置概念#

1.4.2 消息裁剪#

• 多轮输入 = 系统消息 + 历史 + 最新提问；模型没有记忆，每轮都把整段上下文重新送进去。
• trim_messages 能把历史压到指定 token 数或指定条数。

1
from langchain_core.messages import trim_messages
2

3
trimmer = trim_messages(
4
    max_tokens=65, strategy="last", token_counter=model,
5
    include_system=True, allow_partial=False, start_on="human",
6
)
7
trimmer.invoke(messages)
8
(trimmer | model).invoke(messages)

参数说明

裁剪合法性四条（不满足会报错）：

1. 必须以 Human / System 起头；
2. 必须以 Human / Tool 收尾；
3. Tool 必须跟在那条发起调用的 AI 消息之后；
4. 历史里有 System，就必须保留它。

坑：早期历史被裁掉后信息就丢了 —— 示例里模型又答不出名字了。

1
trim_messages(max_tokens=11, strategy="last", token_counter=len,
2
              include_system=True, allow_partial=False, start_on="human")

1.4.3 消息过滤#

适用场景：只想把整段历史里的某一部分交给模型。

1
from langchain_core.messages import filter_messages
2

3
filter_messages(messages, include_types="human")
4
filter_messages(messages, include_types=[HumanMessage, AIMessage], exclude_ids=["3"])
5
filter_messages(include_types="human").invoke(messages)     # 与直接调用等价

1.4.4 消息合并#

起因：有些厂商拒绝同角色消息连续出现。

1
from langchain_core.messages import merge_message_runs
2

3
merged = merge_message_runs(messages)     # 方式一：直接调用
4
merger = merge_message_runs()             # 方式二：先建 merger
5
(merger | model).invoke(messages)

• 把相邻同类消息并成一条；合并后内容以换行符拼接，元数据保留在合并结果上。

坑：方式一取回 merged 之后，别再把原始 messages 传给模型 —— 那等于合并白做了。

2. 提示词模板（Prompt Template）#

2.1 概念#

• 定义：可复用的提示蓝图，用占位符在运行时产出最终提示。
• 机制：固定文本 + 输入变量名清单，引擎逐个替换占位符。

四条价值

• 两条路径：字符串模板填空得文本；消息模板填空得消息对象列表。

2.2 用法#

2.2.1 字符串模板#