LangChain – LarkdropPARA

LangChain 简介

LangChain 是一个用于开发大语言模型应用的框架。它提供了：

统一的模型调用接口（支持 OpenAI、通义千问等多种模型）
提示词模板管理
会话记忆管理
工具调用支持
文档加载、切分、向量检索（RAG）支持

安装依赖：

```
pip install langchain-core langchain-openai langchain-community redis
```

一轮对话

我们采取循序渐进的方式讲解 LangChain，先从最简单的一轮对话开始，逐步增强：

最基本的一轮对话
增加系统提示词
增加工具调用
结构化输出
实现流式输出

最基本的一轮对话

场景：用户输入一个问题，模型给出回答。

```
from langchain_openai import ChatOpenAI

# 创建模型客户端
# ChatOpenAI 是 LangChain 对 OpenAI 兼容接口的封装
# 通义千问提供了 OpenAI 兼容的接口，所以可以直接使用
llm = ChatOpenAI(
    model="qwen-plus",                                          # 模型名称
    api_key="你的API Key",                                      # API密钥
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 接口地址
)

# 发送消息并获取回答
#
response = llm.invoke("你好，请简单介绍一下你自己")

# response 是一个 AIMessage 对象，其content属性就是回复的字符串内容
print(response.content)
```

增加系统提示词和用户提示词

我们希望通过系统提示词指定模型扮演某个角色来回答问题。

```
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(
    model="qwen-plus",
    api_key="你的API Key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 定义提示词模板
# system: 系统提示词，定义模型的角色和行为边界
# human: 用户输入的消息，{question} 是占位符
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业的客服助手，请耐心、礼貌地回答用户问题。"),
    ("human", "{question}"),
])

# 使用prompt格式化消息，再传给模型
messages = prompt.invoke({"question": "你们的退货政策是什么？"})
response = llm.invoke(messages)
print(response.content)
```

增加工具调用

当模型回答需要依赖外部信息时（比如查天气、查数据库、调用API），我们可以给模型提供工具，让它在需要时调用工具拿到数据再回答。

下面用一个天气查询的例子，演示工具调用的完整闭环。

完整流程：

定义工具 + 把工具绑定到模型
把用户问题发给模型
模型判断需要调用工具 → 返回工具调用请求（不是最终答案）
我们执行工具函数，拿到结果
把工具结果作为 ToolMessage 回传给模型
模型根据工具结果生成最终的自然语言回答

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

# ============ 1. 定义工具 ============

class GetWeatherArgs(BaseModel):
    """工具的参数模型（类比Java中的DTO）"""
    city: str = Field(description="城市名称，例如：北京、上海")

@tool("get_weather", args_schema=GetWeatherArgs)
def get_weather(city: str) -> str:
    """查询指定城市的实时天气。当用户询问某个城市的天气时调用。"""
    # 这里模拟一个天气数据（实际项目中会调用真实的天气API）
    fake_data = {
        "北京": "晴，22°C，东南风3级",
        "上海": "多云，25°C，无风",
        "杭州": "小雨，20°C，南风2级",
    }
    return fake_data.get(city, f"{city} 暂无天气数据")

# ============ 2. 绑定工具到模型 ============

llm = ChatOpenAI(
    model="qwen-plus",
    api_key="你的API Key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
# bind_tools 告诉模型：你有这些工具可以调用
llm_with_tools = llm.bind_tools([get_weather])

# ============ 3. 第一次调用模型 ============

# 维护一个消息列表，记录对话过程（后续要把工具结果追加进去）
messages = [HumanMessage(content="北京今天天气怎么样？")]
response = llm_with_tools.invoke(messages)

# 模型不会直接回答，而是返回一个"我要调用工具"的请求
# response.content 通常为空字符串
# response.tool_calls 包含要调用的工具信息
print(response.tool_calls)
# [{'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_xxx'}]

# 把模型的这个"工具调用请求"也加入到消息列表
messages.append(response)

# ============ 4. 执行工具 ============

# 遍历模型要求调用的所有工具
for tool_call in response.tool_calls:
    # 取出工具名和参数
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]

    # 执行工具函数（这里我们只绑定了一个工具，直接调用）
    if tool_name == "get_weather":
        tool_result = get_weather.invoke(tool_args)  # "晴，22°C，东南风3级"

    # ============ 5. 把工具结果作为ToolMessage回传给模型 ============
    # ToolMessage 必须通过 tool_call_id 关联到对应的工具调用请求
    messages.append(ToolMessage(content=tool_result, tool_call_id=tool_call["id"]))

# ============ 6. 第二次调用模型，让它基于工具结果生成最终回答 ============

final_response = llm_with_tools.invoke(messages)
print(final_response.content)
# 输出类似："北京今天天气晴朗，气温22°C，东南风3级，是个不错的天气。"
```

关键点：

工具调用是两次LLM调用：第一次让模型决定要不要调用工具，第二次让模型根据工具结果生成最终回答
中间的 ToolMessage 必须用 tool_call_id 和工具调用请求关联起来，模型才能知道”这是哪次工具调用的结果”
如果模型一次要调用多个工具（比如同时查北京和上海的天气），就会有多个 tool_call，需要全部执行后再回传

增加结构化输出

很多时候我们希望模型返回的不是一段自由的文本，而是结构化的数据（如JSON），方便后续代码直接使用。

```
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 定义输出结构（类比Java中的响应DTO）
class SentimentResult(BaseModel):
    sentiment: str = Field(description="情感倾向：正面、负面、中性")
    score: int = Field(description="强烈程度，1-10分")
    keywords: list[str] = Field(description="关键词列表")

llm = ChatOpenAI(
    model="qwen-plus",
    api_key="你的API Key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# with_structured_output 让模型按照指定的结构返回数据
structured_llm = llm.with_structured_output(SentimentResult)
```

这里有个很容易困惑的点：模型返回的不应该是字符串吗？为什么后面可以直接 result.sentiment 当对象访问？

with_structured_output(SentimentResult) 在背后默默做了两件事：

告诉模型按schema返回JSON：把 SentimentResult 这个 Pydantic 模型的字段定义（字段名+类型+description）转换成 JSON Schema，作为提示词的一部分发给模型，强制模型返回严格符合这个 schema 的 JSON
自动把JSON转成对象：收到模型返回的 JSON 字符串后，LangChain 自动调用 SentimentResult(**json_data)，把 JSON 反序列化为 Pydantic 对象

所以最终我们拿到的 result 已经是一个 SentimentResult 对象了，不是原始JSON字符串。这就和Java里的Jackson把JSON自动转成DTO对象是一回事。

```
result = structured_llm.invoke("这家餐厅的菜太难吃了，服务也差，再也不来了")

# result 是 SentimentResult 类的实例，可以直接用 . 访问属性
print(f"情感倾向: {result.sentiment}")    # 负面
print(f"强烈程度: {result.score}")        # 9
print(f"关键词: {result.keywords}")       # ['难吃', '服务差']

# 验证 result 的类型
print(type(result))  # <class '__main__.SentimentResult'>
```

实现流式输出

什么是SSE（Server-Sent Events）？

SSE是一种服务器向客户端单向推送数据的技术。想象一下你在看直播弹幕，弹幕不是一次性全部加载的，而是一条一条实时推送过来的。SSE就是这个原理——服务器可以持续不断地向客户端推送数据，而不需要客户端反复请求。

模型生成回答是逐字逐句的，如果等模型全部生成完再返回，用户需要等很久。使用SSE，模型每生成一小段文字就立刻推送给用户，用户体验就像”打字机”一样。

注意：SSE并未提升模型的输出效率，但它可以提升用户体验，让用户尽快得到第一次响应！

在讲解如何在LangChain中实现流式输出前，我们必须先理解 Python 中的 yield 关键字。

普通函数用 return 返回结果，函数执行完就结束了。而使用 yield 的函数叫做生成器函数，它可以”暂停”执行，每次调用时返回一个值，下次调用时从上次暂停的地方继续。

打个比方：return 就像一次性把整本书给你，yield 就像一页一页地翻给你看。

```
# 普通函数：一次性返回所有数据
def get_all_numbers():
    return [1, 2, 3, 4, 5]

# 生成器函数：每次yield一个数据
def generate_numbers():
    yield 1  # 第一次调用返回1，然后暂停
    yield 2  # 第二次调用返回2，然后暂停
    yield 3  # 第三次调用返回3，然后暂停

# 使用生成器
for num in generate_numbers():
    print(num)  # 依次打印 1, 2, 3
```

流式输出的实现：

```
from langchain_openai import ChatOpenAI
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

llm = ChatOpenAI(
    model="qwen-plus",
    api_key="你的API Key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    streaming=True,  # 开启流式模式
)

@app.post("/chat/stream")
def stream_chat(message: str):
    def event_stream():
        # llm.stream() 返回一个生成器，每次yield一个文本片段
        for chunk in llm.stream(message):
            text = chunk.content
            if text:
                # 按照SSE格式封装数据
                payload = json.dumps({"type": "delta", "text": text}, ensure_ascii=False)
                yield f"data: {payload}\n\n"
        # 发送结束标记
        yield f"data: {json.dumps({'type': 'done'})}\n\n"

    # StreamingResponse 会持续调用 event_stream() 生成器
    return StreamingResponse(event_stream(), media_type="text/event-stream")
```

—

链（Chain）

回顾一下前面”增加系统提示词”那一节的代码：

```
# 普通写法：手动调用prompt格式化，再手动调用llm
prompt = ChatPromptTemplate.from_messages([...])
llm = ChatOpenAI(...)

# 先格式化
messages = prompt.invoke({"question": "你好"})
# 再调用模型
response = llm.invoke(messages)
print(response.content)
```

这种写法没有问题，但当步骤变多时，代码会变得冗长。LangChain提供了一种更优雅的写法——链（Chain）。

什么是链？

链就是将多个处理步骤串联起来的管道。每个步骤的输出会自动成为下一个步骤的输入，最终得到处理结果。

类比Java： 链的概念类似于Java中的Stream流式处理：

```
// Java Stream：多个处理步骤串联
List<String> result = users.stream()
    .filter(u -> u.getAge() > 18)     // 步骤1：过滤
    .map(User::getName)               // 步骤2：转换
    .collect(Collectors.toList());    // 步骤3：收集
```

在LangChain中，使用 | 管道符将组件串联成链：

```
# 链的写法：一行代码串联多个步骤
chain = prompt | llm | output_parser
```

对比普通写法和链的写法：

```
# ===== 普通写法 =====
messages = prompt.invoke({"question": "你好"})
response = llm.invoke(messages)
result = parser.invoke(response)

# ===== 链的写法（等价） =====
chain = prompt | llm | parser
result = chain.invoke({"question": "你好"})
```

两种写法的结果完全一样，但链的写法更简洁，而且支持 .stream() 流式调用。

这套用 | 把各个组件串成链的写法，有个正式名字叫 LCEL（LangChain Expression Language，LangChain 表达式语言）。 它是 LangChain 推荐的、构建链的标准方式。本课件后面（包括多轮对话、RAG）都会用 LCEL 来组装链。下面先把 LCEL 里最常用的几个东西讲清楚：链的构造规则、RunnableLambda、RunnablePassthrough。

链的构造规则

并不是任意对象都能参与链的构造，有以下规则：

规则1：只有 Runnable 对象才能参与链的构造

LangChain中所有能参与链的组件都实现了 Runnable 接口（类比Java中的接口约束）。常见的Runnable组件有：

组件	作用
`ChatPromptTemplate`	格式化提示词
`ChatOpenAI` (LLM)	调用大模型
`StrOutputParser`	解析输出为字符串
`RunnableWithMessageHistory`	自动管理历史消息

普通的Python函数、字符串、字典等不能直接参与链的构造（除非用 RunnableLambda 包装）。

规则2：前一个组件的输出类型必须匹配后一个组件的输入类型

```
ChatPromptTemplate  →  输出：消息列表
ChatOpenAI          →  输入：消息列表，输出：AIMessage
StrOutputParser     →  输入：AIMessage，输出：字符串
```

所以正确的顺序是：prompt | llm | output_parser

如果写成 llm | prompt 就会报错，因为LLM输出的是AIMessage，而prompt期望的输入是字典。

规则3：顺序决定了数据的流向

链的执行顺序严格从左到右，数据像水流一样依次经过每个组件：

```
# 正确：先格式化提示词 → 再调用模型 → 最后解析输出
chain = prompt | llm | output_parser

# 错误：顺序反了，类型不匹配
chain = output_parser | prompt | llm  # ❌ 会报错
```

RunnableLambda

前面说过普通的Python函数不能直接参与链的构造，那如果我就是想在链中插入一段自定义的处理逻辑怎么办？用 RunnableLambda 把函数包装一下就行。

场景：用户的原始问题不能直接给prompt，需要先做预处理

比如我们想在用户问题前面统一加上”用户咨询：”的前缀，让模型识别得更清楚。如果不用链，可能要这样写：

```
# 没有链的写法
question = "你们的退货政策是什么？"
processed = f"用户咨询：{question}"                 # 第一步：手动加前缀
messages = prompt.invoke({"question": processed})  # 第二步：手动格式化
response = llm.invoke(messages)                     # 第三步：手动调模型
```

用 RunnableLambda 把”加前缀”这个普通函数包成 Runnable，就可以串到链里了：

```
from langchain_core.runnables import RunnableLambda

# 1. 定义一个普通的Python函数
def add_prefix(inputs: dict) -> dict:
    """给问题加上"用户咨询："前缀"""
    question = inputs["question"]
    return {"question": f"用户咨询：{question}"}

# 2. 用 RunnableLambda 把普通函数包装成 Runnable
preprocess = RunnableLambda(add_prefix)

# 3. 串到链中：预处理 → 格式化提示词 → 调用模型
chain = preprocess | prompt | llm

# 4. 调用链，预处理会自动在格式化提示词之前执行
result = chain.invoke({"question": "你们的退货政策是什么？"})
# 模型实际看到的问题是："用户咨询：你们的退货政策是什么？"
```

RunnableLambda 就是一个适配器，把任何普通函数都能转成可以参与链的 Runnable。常见用途：数据清洗、字段提取、格式转换等。

RunnablePassthrough

当链的输入是一个字典时，经常会遇到这样的需求：在不破坏原字典的前提下，给它补充一个新字段。RunnablePassthrough 就是干这个的。

先看它最基础的用法——RunnablePassthrough() 表示”原样传递”，输入什么就输出什么：

```
from langchain_core.runnables import RunnablePassthrough

chain = RunnablePassthrough()
print(chain.invoke({"question": "你好"}))   # 原样输出 {"question": "你好"}
```

光”原样传递”没什么用，它的威力在于 .assign() 方法——在原输入字典的基础上，新增一个键值对：

加到哪个字典：加到当前流经这条链的输入字典上（也就是 invoke 时传进来的那个）。
key（键）怎么定：就是你写在 assign(...) 里的那个参数名。比如 assign(total=...)，新增的 key 就是 "total"。
value（值）从哪来：把原输入字典整个交给等号右边的 Runnable 去运行，它返回什么，这个新 key 的 value 就是什么。
原字典里已有的字段原样保留，不受影响。

举个最小的例子：

```
from langchain_core.runnables import RunnablePassthrough

# 等号右边是一个 Runnable，这里用 lambda 包一个简单函数
# 它接收整个输入字典 x，返回 x["a"] + x["b"]
chain = RunnablePassthrough.assign(total=lambda x: x["a"] + x["b"])

print(chain.invoke({"a": 3, "b": 5}))
# 输出：{"a": 3, "b": 5, "total": 8}
#       ↑ 原有的 a、b 原样保留    ↑ 新增的 key="total"，value=右边函数的返回值
```

RunnablePassthrough.assign 在后面的 RAG 里会派上大用场——用它来给输入”自动补上检索到的内容”这个字段。

链的调用方式

```
chain = prompt | llm

# 方式1：invoke —— 同步调用，等待完整结果
result = chain.invoke({"question": "你好"})

# 方式2：stream —— 流式调用，逐步获取结果（用于SSE）
for chunk in chain.stream({"question": "你好"}):
    print(chunk)
```

—

四、多轮对话（会话记忆）

前面的对话都是”一锤子买卖”——模型回答完就忘了，下一次提问它完全不记得上一轮聊了什么。要实现多轮对话，核心就是会话记忆：把历史对话保存下来，每次提问时连同历史一起发给模型。

实现会话记忆要解决两个问题：

会话记忆的保存和隔离：历史消息存在哪？如何区分不同的会话？
上下文组织：在多轮对话的每一轮对话中，到底给模型发送哪些内容？

会话记忆存的保存和隔离

LangChain支持多种会话记忆的持久化方式，并提供了开箱即用的实现：

持久化方式	对应类	特点
内存	`ChatMessageHistory`	进程重启后丢失，仅适合测试
Redis	`RedisChatMessageHistory`	高性能，支持TTL自动过期，适合生产环境
PostgreSQL	`PostgresChatMessageHistory`	持久化到关系数据库
MongoDB	`MongoDBChatMessageHistory`	持久化到文档数据库
文件系统	`FileChatMessageHistory`	持久化到本地文件

在我们的项目中使用 Redis（RedisChatMessageHistory）

存储在RedisChatMessageHistory中的会话记忆怎么隔离？

RedisChatMessageHistory 通过一个 session_id 参数来隔离不同用户的会话——给每个会话分配一个唯一的 session_id，保存和读取时都带上它，不同会话的消息就分开存了。

```
from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_history(session_id: str) -> RedisChatMessageHistory:
    return RedisChatMessageHistory(
        session_id=session_id,          # 用 session_id 隔离不同会话
        url="redis://localhost:6379", # Redis连接地址
        key_prefix="chat:",             # key前缀，最终key为 chat:{session_id}
        ttl=604800,                     # 7天自动过期
    )
```

会话记忆的自动管理

有了 RedisChatMessageHistory，是不是每轮对话都要我们自己调它的 add_messages() 来保存用户输入和模型回答？

不需要。LangChain 提供了一个叫 RunnableWithMessageHistory 的封装，专门帮我们自动完成”读历史、拼历史、写历史”这些事。

RunnableWithMessageHistory 是一个链的包装器，它包装一个基础链，自动为这个链维护会话记忆。它的核心职责就三件事：

调用前：根据 session_id 调用 get_session_history 拿到这个会话的历史对象，读出历史消息
调用中：把历史消息交给基础链，让历史参与到本次模型调用中
调用后：把本轮的”用户消息”和”模型回答”追加写回历史

但第2步”历史消息怎么交给基础链”，分两种情况，取决于你的基础链接收的输入是什么——是单条消息，还是字典（dict）。这两种情况的写法差别很大，下面分别讲。

情况一：输入是单条消息（最简单）

如果基础链直接就是一个 llm（或 llm | parser），输入是用户的一条消息，那 RunnableWithMessageHistory 会把”历史消息 + 当前消息”拼成一个消息列表直接喂给模型。这种情况下不需要提示词模板，也不需要指定任何 key：

```
from langchain_openai import ChatOpenAI
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import RedisChatMessageHistory

llm = ChatOpenAI(model="qwen-plus", api_key="...", base_url="...")

# 基础链就是 llm 本身，输入/输出都是消息
chain_with_history = RunnableWithMessageHistory(
    llm,                          # 基础链：直接是 llm
    get_session_history=get_history
)

# 调用时直接传一条消息（字符串会被当成 HumanMessage）
result = chain_with_history.invoke(
    "我叫张三",
    config={"configurable": {"session_id": "abc-123"}},
)
```

这种模式下，RunnableWithMessageHistory 自动把历史消息列表和当前消息拼在一起作为模型输入，你什么都不用管。

但它的局限也很明显：没法加系统提示词、没法在历史前后插入别的内容——因为输入就是一串纯消息。实际项目里我们几乎总要加系统提示词（指定角色、注入知识等），所以更常用的是下面第二种。

情况二：输入是字典（需要提示词模板）

当基础链是 prompt | llm 这种带提示词模板的链时，链的输入是一个字典（比如 {"question": "..."}），由模板去填充。这种情况下有两个关键约定，必须自己处理：

基础链必须有 ChatPromptTemplate 模板，而且模板里要用 MessagesPlaceholder 显式占一个”历史消息”的位置——框架不会自动帮你把历史拼进去，它只负责把历史放到你指定的那个占位符，到底插在系统提示词后面还是哪里，由你在模板里决定。
必须告诉框架两个 key：

input_messages_key：输入字典里哪个 key 是”当前用户消息”（框架要把它写回历史）
history_messages_key：把读出来的历史塞到模板的哪个 MessagesPlaceholder（名字要对得上）

```
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import RedisChatMessageHistory

llm = ChatOpenAI(model="qwen-plus", api_key="...", base_url="...")

# 1. 提示词模板：必须自己用 MessagesPlaceholder 占好历史消息的位置
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位友好的助手"),
    MessagesPlaceholder("history"),   # ← 历史消息会被塞到这里（位置由你决定）
    ("human", "{question}"),           # 当前用户消息
])

# 2. 基础链：带模板
base_chain = prompt | llm

# 3. 包装时，必须指定两个 key
chain_with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history=lambda session_id: RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379",
        key_prefix="chat:"
    ),
    input_messages_key="question",     # 输入字典里"用户问题"对应的 key
    history_messages_key="history",     # 历史塞到模板里名为 history 的占位符
)

# 4. 调用时传字典（key 要和模板对得上），并通过 config 传入 session_id
result = chain_with_history.invoke(
    {"question": "我叫张三"},
    config={"configurable": {"session_id": "abc-123"}},
)
# 下一次用同一个 session_id 提问，模型就能记得"你叫张三"
result = chain_with_history.invoke(
    {"question": "我叫什么名字？"},
    config={"configurable": {"session_id": "abc-123"}},
)
# 模型回答："你叫张三"
```

两种情况对比：

	情况一：输入是消息	情况二：输入是字典
基础链	直接 `llm`	`prompt	llm`（必须有模板）
提示词模板	不需要	必须，且要有 `MessagesPlaceholder`
历史拼接	框架自动拼到消息列表	框架塞到你指定的占位符，拼接位置由你的模板决定
需要指定 key	不需要	必须指定 `input_messages_key` 和 `history_messages_key`
能否加系统提示词	不能	能（在模板里写）

实际项目里几乎都用情况二——因为我们总要加系统提示词、注入知识等，离不开提示词模板。下面的内容也都以情况二为基础。

会话记忆的裁剪

RunnableWithMessageHistory 默认会读取该 session_id 下的全部历史消息。但把所有历史都拼进提示词是不行的：

模型有上下文窗口限制（token数量有上限）
消息越多，调用成本越高（按token计费）
过多的历史消息可能引入噪音，反而降低回答质量

所以通常采用窗口策略：只保留最近 N 条消息。

最简单的做法是写一个普通函数对历史列表做切片，再用 RunnableLambda 把它串进链里：

```
from langchain_core.runnables import RunnableLambda

WINDOW = 20  # 保留最近20条

def trim_history(inputs: dict) -> dict:
    """裁剪输入字典中的历史消息，只保留最近 WINDOW 条"""
    values = dict(inputs)
    history = values.get("history")
    if isinstance(history, list):
        values["history"] = history[-WINDOW:]   # 列表切片，保留最后N条
    return values

# 把裁剪函数串在 prompt 之前
# 数据流：读历史(自动) → 裁剪(我们写的函数) → 拼接提示词 → 调用模型
base_chain = RunnableLambda(trim_history) | prompt | llm

chain_with_history = RunnableWithMessageHistory(
    base_chain,
    get_session_history=get_history,
    input_messages_key="question",
    history_messages_key="history",
)
```

为什么 inputs 里会有 history 字段？ 因为 RunnableWithMessageHistory 在链执行前会自动把历史消息读出来塞进 inputs 字典，key 就是配置的 history_messages_key="history"。所以等 trim_history 执行时，inputs 里已经有 {"question": "...", "history": [历史消息列表]} 了。

完整数据流：

```
chain_with_history.invoke({"question": "..."}, config={"session_id": "abc"})
   ↓
RunnableWithMessageHistory 读取存储中 abc 会话的全部历史
   ↓ 在 inputs 中添加 "history": [全部历史]
RunnableLambda(trim_history) 把 history 切片为最近20条
   ↓ inputs 变为 {"question": "...", "history": [最近20条]}
prompt 把20条历史 + 当前问题 拼成完整消息列表
   ↓
llm 调用模型生成回答
   ↓
RunnableWithMessageHistory 把"当前用户消息"和"模型回答"追加写回存储
```

关键点： 存储里始终保留完整历史，每次调用模型时才临时裁剪到20条。这样存储的数据始终完整，而模型不会被超长上下文淹没。

—

RAG（检索增强生成）

下面我们用一份《员工请假管理制度》文档，从零实现一个最基础的 RAG。

需要额外安装的依赖：

```
pip install langchain-text-splitters langchain-chroma chromadb
```

假设我们有一份 员工请假管理制度.md，内容节选如下：

```
## 第三章 年假
第五条 年假天数根据累计工作年限确定：累计工作满1年不满10年的，年假5天；
满10年不满20年的，年假10天；满20年以上的，年假15天。

## 第五章 事假
第十三条 事假每月累计不得超过3天，全年累计不得超过15天。超过限额的，公司有权不予批准。

## 第八章 请假审批流程
第二十条 审批权限如下：请假1天以内（含1天）由直属主管审批；请假2天至3天由部门负责人审批；
请假3天以上由部门负责人和人力资源部共同审批；请假7天以上需报总经理审批。
```

我们的目标：让模型能准确回答”事假一年最多请几天””请5天假谁来审批”这类问题。

加载文档

RAG 的第一步是把文档读进程序。LangChain 提供了各种 Loader（加载器）来读取不同格式的文件（txt、md、pdf、word……）。

这里我们的文档是 Markdown 文本文件，用 TextLoader 即可：

```
from langchain_community.document_loaders import TextLoader

# 创建加载器，指定文件路径和编码
loader = TextLoader("员工请假管理制度.md", encoding="utf-8")

# load() 返回一个 Document 列表
documents = loader.load()

# 整个文件会被读成 1 个 Document
# Document 有两个核心属性：
#   - page_content: 文档的文本内容
#   - metadata:     元数据（如来源文件路径等）
print(documents[0].page_content[:50])   # 打印前50个字
print(documents[0].metadata)            # {'source': '员工请假管理制度.md'}
```

关键概念 Document： LangChain 用 Document 对象统一表示一段文档，它包含 page_content（正文）和 metadata（元数据）。后面切分、向量化、检索，操作的都是 Document。

切分文档

LangChain 用 RecursiveCharacterTextSplitter（递归字符切分器）来切分。它会按照预设的分隔符（先按段落、再按句子、再按字……）递归地把长文本切成大小合适的块。

```
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 创建切分器
splitter = RecursiveCharacterTextSplitter(
    chunk_size=200,        # 每个块的目标大小（字符数）
    chunk_overlap=30,      # 相邻块之间的重叠字符数
    separators=["\n\n", "\n", "。", "；", "，", ""],  # 按优先级尝试的分隔符
)

# split_documents 接收 Document 列表，返回切分后的 Document 列表
chunks = splitter.split_documents(documents)

print(f"切分成了 {len(chunks)} 个块")
for chunk in chunks[:2]:
    print("---")
    print(chunk.page_content)
```

三个关键参数：

chunk_size：每个块的目标字符数。太大检索不精准，太小又会丢失上下文。请假制度这种条款型文档，每条几十到一两百字，设 200 比较合适。
chunk_overlap：相邻两个块重叠的字符数。为什么要重叠？防止一句话正好被切断，导致两个块都不完整。让相邻块共享一部分内容，能保证语义连续。
separators：分隔符列表，按优先级排列。切分器先用第一个（\n\n 段落）切，如果切出来的块还是太大，再用下一个（\n 换行）继续切，依此类推。这样能尽量在”自然边界”（段落、句号）处切断，而不是从字中间硬切。

切分后，每个 chunk 仍然是一个 Document，并且自动继承了原文档的 metadata。

向量化并存入向量数据库

我们用 Chroma 作为向量数据库来存储向量。我们这里连接的是部署在远程服务器上的 Chroma 服务，所以要通过 host 和 port 指定它的地址

```
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# 1. 创建嵌入模型（负责把文本转成向量）
embeddings = OpenAIEmbeddings(
    model="text-embedding-v3",     # 嵌入模型名
    api_key="你的API Key",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 2. 把所有 chunk 向量化并存入 Chroma
#    from_documents 内部会：① 对每个chunk调用embeddings转成向量
#                          ② 把(向量, 文本, metadata)一起存进Chroma
vector_store = Chroma.from_documents(
    documents=chunks,              # 上一步切分好的块
    embedding=embeddings,          # 嵌入模型
    collection_name="leave_policy",# 集合名（类似数据库的表名）
    host="192.168.1.100",          # 远程 Chroma 服务的地址
    port=8000,                     # 远程 Chroma 服务的端口
)

print("知识库构建完成！")
```

执行完后，整份请假制度就被切成若干小块、转成向量、存进远程的 Chroma 服务里了。构建阶段到此结束——这一步是离线的，只需做一次。

检索

构建好知识库后，就可以根据用户问题检索相关内容了。这是最基础的检索——相似度检索：把用户问题也转成向量，在 Chroma 里找向量最接近的几个块。

```
# 连接到已构建好的向量库（如果是新进程，重新连一次）
vector_store = Chroma(
    collection_name="leave_policy",
    embedding_function=embeddings,
    host="192.168.1.100",          # 远程 Chroma 服务的地址
    port=8000,                     # 远程 Chroma 服务的端口
)

# 用 similarity_search 做相似度检索
# k=2 表示返回最相关的 2 个块
question = "事假一年最多能请多少天？"
results = vector_store.similarity_search(question, k=2)

for doc in results:
    print("---")
    print(doc.page_content)
# 会检索出包含"第十三条 事假每月累计不得超过3天，全年累计不得超过15天"的那个块
```

similarity_search(question, k=2) 做的事：

把 question 用嵌入模型转成向量
在 Chroma 中找出向量距离最近的 k 个 chunk
返回这些 chunk（Document 列表）

更常用的写法：把向量库变成”检索器（Retriever）”

为了后面能把检索接入链（Chain），通常把向量库转成一个 retriever：

```
# as_retriever 把向量库包装成一个 Runnable，可以参与链
retriever = vector_store.as_retriever(search_kwargs={"k": 2})

# retriever 也有 invoke 方法，传入问题，返回相关 Document 列表
docs = retriever.invoke("事假一年最多能请多少天？")
```

上下文注入

检索拿到相关内容后，下一步是把这些内容”注入”到提示词里发给模型——这一步叫上下文注入。

我们先在提示词里留一个 {context} 占位符，专门用来放检索到的内容：

```
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 提示词：检索到的内容放进 {context}，用户问题放进 {question}
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是公司HR助手，请严格根据下面提供的制度内容回答员工问题。
如果制度内容里没有相关规定，就如实说"制度中未提及"，不要编造。

制度内容：
{context}"""),
    ("human", "{question}"),
])

llm = ChatOpenAI(model="qwen-plus", api_key="...", base_url="...")
```

这里有个细节要注意：retriever.invoke(question) 返回的是一个 Document 列表，而提示词里的 {context} 需要的是一段纯文本。所以中间要自己写一步”把 Document 列表拼成文本”：

```
# 手动写一个函数，把 Document 列表拼成纯文本
def format_docs(docs) -> str:
    # 取出每个 Document 的正文 page_content，用空行拼接起来
    return "\n\n".join(doc.page_content for doc in docs)

def answer(question: str) -> str:
    # ① 检索：拿到 Document 列表
    docs = retriever.invoke(question)
    # ② 注入：把 Document 列表拼成文本，填进 {context}
    context = format_docs(docs)
    # ③ 回答：调用链
    chain = prompt | llm | StrOutputParser()
    return chain.invoke({"context": context, "question": question})

# 测试
print(answer("事假一年最多能请多少天？"))
# 输出类似："根据制度第十三条，事假每月累计不得超过3天，全年累计不得超过15天。"
```

这样整个 RAG 闭环就跑通了：检索 → 拼成文本注入 → 模型回答。

但这里有两处是我们”手动”在做的：

手动调用 retriever.invoke(question) 检索
手动写 format_docs 把 Document 列表拼成文本

下一节我们用 LangChain 提供的工具，把这两步都自动化，串成一条完整的链。

自动检索与注入

create_stuff_documents_chain 自动拼接文档

LangChain 提供了一个现成的工具把 Document 列表拼成文本再填进提示词的 {context}，这个工具就是——create_stuff_documents_chain。

create_stuff_documents_chain 会创建一个链，它的作用是：接收一个 Document 列表，自动把它们拼成文本，填进提示词的占位符，再调用模型。（”stuff” 就是”把所有文档一股脑塞进上下文”的意思。）

```
from langchain.chains.combine_documents import create_stuff_documents_chain

# 创建文档链
document_chain = create_stuff_documents_chain(
    llm,                              # 用哪个模型来回答
    prompt,                           # 提示词模板
    document_variable_name="context", # 见下方说明
)

# 调用：context 直接传 Document 列表（不用自己拼文本了），它会自动拼接并注入
docs = retriever.invoke("事假一年最多能请多少天？")   # 手动检索，拿到 Document 列表
result = document_chain.invoke({
    "context": docs          # 直接传 Document 列表！
})
print(result)
```

对比上一节：之前 {context} 要填我们 format_docs 拼好的字符串，现在 context 直接传 Document 列表，拼接交给 create_stuff_documents_chain 自动完成，format_docs 不用我们自己写了。

重点解释 document_variable_name 这个参数。 结合上面的调用代码 invoke({"context": docs, ...}) 来看，它其实同时承担了两个作用（用的是同一个名字 context）：

读取：告诉 create_stuff_documents_chain 从输入字典的哪个 key 里取出 Document 列表。我们设的是 "context"，所以它就去 invoke 传入的字典里找 context 这个 key，把它的值（docs，一个 Document 列表）拿出来。
写入：把拿到的 Document 列表拼成文本后，填到提示词里同名的占位符 {context}。

RunnablePassthrough.assign 自动检索

虽然 document_chain 帮我们自动拼接了文档，但 docs = retriever.invoke(...) 这步检索还是手动的

我们希望的是：调用时只传用户问题，检索这一步也自动完成。这就要用到前面”链”小节讲过的 RunnablePassthrough.assign——它能在原输入字典的基础上新增一个字段（key 是参数名，value 是右边 Runnable 的运行结果，原有字段原样保留）。

回到 RAG，我们正好可以用它给输入字典自动补上 context 字段：

```
from langchain_core.runnables import RunnablePassthrough

# 检索器接收的是字符串问题，但我们的输入是 {"question": "..."} 字典
# 所以先从字典里取出 question 再交给 retriever
retrieve_docs = (lambda x: x["question"]) | retriever

# assign(context=retrieve_docs) 的含义：
#   - 新增的 key 是 "context"（等号左边的参数名）
#   - value 是 retrieve_docs 运行后的返回值（即检索到的 Document 列表）
#   - 原有的 question 字段原样保留
rag_chain = RunnablePassthrough.assign(context=retrieve_docs) | document_chain
```

这条链的数据流是这样的：

```
调用 rag_chain.invoke({"question": "事假一年最多能请多少天？"})
   ↓
RunnablePassthrough.assign(context=retrieve_docs)
   ↓ 取出 question → 交给 retriever 检索 → 得到 Document 列表
   ↓ 把 Document 列表作为 context 加进字典，question 原样保留
   ↓ 字典变为 {"question": "...", "context": [Document, Document]}
document_chain
   ↓ 自动把 context（Document列表）拼成文本填进 {context} 占位符
   ↓ 调用模型回答
最终输出：模型的回答
```

注意 assign 的精妙之处：它保留了原有的 question 字段，同时新增了 context 字段。这样最终传给 document_chain 的字典里，question 和 context 都齐了——而我们调用时只传了 question，context 是自动检索补上的。

测试效果：

```
# 调用时只传 question，检索和注入全自动
print(rag_chain.invoke({"question": "事假一年最多能请多少天？"}))
# 输出类似："根据制度第十三条，事假每月累计不得超过3天，全年累计不得超过15天。"

print(rag_chain.invoke({"question": "请5天假需要谁审批？"}))
# 输出类似："根据制度第二十条，请假3天以上由部门负责人和人力资源部共同审批。"

print(rag_chain.invoke({"question": "公司有没有团建经费？"}))
# 输出类似："制度中未提及。"（因为请假制度里确实没有这条，模型不会瞎编）
```

至此，整个 RAG 被串成了一条完整的链：传入问题 → 自动检索 → 自动拼接注入上下文 → 模型回答。

回顾整个流程：

阶段	步骤	用到的组件
构建（离线）	加载文档	`TextLoader`
构建（离线）	切分文档	`RecursiveCharacterTextSplitter`
构建（离线）	向量化入库	`OpenAIEmbeddings` + `Chroma`
检索（在线）	相似度检索	`vector_store.as_retriever()`
检索（在线）	文档拼接+注入	`create_stuff_documents_chain`
检索（在线）	自动检索串联	`RunnablePassthrough.assign`

这就是 RAG 最核心的骨架。实际项目中还会有很多增强（查询重写、重排序、混合检索等），但都是在这个基础骨架上做的扩展。