RAG基础与环境LangChain 生态系统

第 3 章：LangChain 生态系统

上一章：RAG 核心概念深入 下一章：Milvus 索引机制与基本操作

---

本章导入

LangChain 的组件很多：Runnable、Prompt、Message、Parser、History、Loader、Splitter、VectorStore 都是常见概念。学习这些组件时，最重要的不是记住每个类名，而是看清它们在企业级 RAG 项目中分别出现在什么位置、解决什么工程问题。

本章只围绕三个问题：

1. LangChain 在本项目里到底是什么角色？
2. 一次在线问答中，哪些步骤用到了 LangChain？
3. 一次离线入库中，哪些步骤用到了 LangChain？

先记住一句话：

本项目没有把 LangChain 当成“一键 RAG 框架”，而是把它当成一组可靠的工程适配器：模型适配器、消息对象、结构化输出、历史记录、文档对象、加载器、切分器和向量库封装。

---

本章目标

学完本章后，需要能说清楚：

• 为什么项目用 ChatOpenAI 接入 DashScope，而不是直接绑定某个厂商 SDK。
• SystemMessage、HumanMessage、AIMessage 在多轮对话和 Prompt 中分别承担什么角色。
• with_structured_output() 为什么比让模型返回自由文本更适合查询变体生成。
• SQLChatMessageHistory、Document、TextSplitter、Milvus VectorStore 分别位于项目哪条链路。
• 为什么本项目没有直接使用 RetrievalQA、ConversationalRetrievalChain 或一条 LCEL 管道完成全部 RAG。

---

第一部分：LangChain 在项目中的角色

1.1 LangChain 不是完整业务流程

LangChain 不是模型，也不是知识库，更不是项目的业务大脑。它更像一组标准接口：

项目问题	LangChain 提供的抽象
不同 LLM 厂商 API 不一致	ChatOpenAI 等 ChatModel 统一接口
多轮对话消息结构容易混乱	SystemMessage / HumanMessage / AIMessage
LLM 输出格式不稳定	with_structured_output() + Pydantic
历史消息要持久化	SQLChatMessageHistory
文件格式不同	Document Loader
大文档需要切块	Text Splitter
向量库写入和检索 API 复杂	VectorStore

也就是说，LangChain 解决的是接口标准化和工程胶水问题。本项目真正的业务流程仍然由 qa_core 自己编排。

1.2 本项目的使用边界

flowchart TB
    subgraph Use["本项目重点使用"]
        A["ChatOpenAI<br/>统一 LLM 调用"]
        B["Message Types<br/>多轮对话消息"]
        C["Structured Output<br/>意图/变体结构化"]
        D["SQLChatMessageHistory<br/>MySQL 历史"]
        E["Document / Loader / Splitter<br/>离线入库"]
        F["Milvus VectorStore<br/>向量库封装"]
    end

    subgraph Light["本项目仅作说明，不作为主链路"]
        G["Runnable<br/>invoke / stream / batch"]
        H["LCEL<br/>prompt | model | parser"]
    end

    subgraph Avoid["本项目不直接采用"]
        I["RetrievalQA"]
        J["ConversationalRetrievalChain"]
        K["自由任务编排"]
    end

    Use --> Light --> Avoid

    style Use fill:#ECFDF5,stroke:#059669,stroke-width:2px
    style Light fill:#EFF6FF,stroke:#2563EB,stroke-width:2px
    style Avoid fill:#FEF2F2,stroke:#DC2626,stroke-width:2px

这里要特别区分：不用高层 Chain，不代表不用 LangChain。本项目用的是 LangChain 的底层稳定组件，把分支、阈值、追问改写、FAQ 直出、文档检索、Prompt 档位这些业务决策留在项目代码里。

---

第二部分：一张图看清两条主线

本项目中 LangChain 的使用分成两条线：

1. 在线问答链路：用户提问 -> 意图识别 -> 检索准备 -> Prompt -> LLM 流式生成。
2. 离线入库链路：业务文件 -> Document -> 切分 -> Milvus VectorStore 写入。

flowchart LR
    subgraph Online["在线问答链路"]
        Q["用户问题"]
        M1["Message<br/>历史上下文"]
        I["Structured Output<br/>意图/查询变体"]
        P["Prompt Profile<br/>System + Human"]
        L["ChatOpenAI.stream()<br/>流式生成"]
    end

    subgraph Offline["离线入库链路"]
        F["PDF / Word / MD / Excel"]
        D["Document Loader<br/>统一成 Document"]
        S["Text Splitter<br/>父子块切分"]
        V["Milvus VectorStore<br/>add_documents"]
    end

    subgraph Store["Milvus"]
        FAQ[("FAQ 集合")]
        DOC[("文档集合")]
    end

    Q --> M1 --> I --> P --> L
    F --> D --> S --> V --> DOC
    I --> FAQ
    P --> DOC

    style Online fill:#EFF6FF,stroke:#2563EB,stroke-width:2px
    style Offline fill:#ECFDF5,stroke:#059669,stroke-width:2px
    style Store fill:#FFFBEB,stroke:#D97706,stroke-width:2px

这张图就是本章主线。后面所有组件都放回这两条线里说明。

---

第三部分：在线问答链路中的 LangChain

这部分回答“用户提问之后，LangChain 在哪里参与在线回答”。先看项目代码位置，再看组件作用。

3.1 ChatOpenAI：统一模型调用入口

项目文件：qa_core/llm/client.py

本项目实际使用 DashScope 的 OpenAI-compatible 接口，但代码里不直接写 DashScope SDK，而是统一用 LangChain 的 ChatOpenAI：

from functools import lru_cache

from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

from qa_core.config.settings import get_settings

@lru_cache(maxsize=2)
def get_chat_model(streaming: bool = False) -> ChatOpenAI:
    settings = get_settings()
    return ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=settings.llm_temperature,
        timeout=settings.llm_timeout,
        streaming=streaming,
    )

这里的设计点：

• base_url 可以指向 DashScope、DeepSeek 或其他 OpenAI-compatible 服务。
• streaming=False 用于查询改写和查询变体结构化输出。
• streaming=True 用于最终答案生成，方便 WebSocket 逐 token 推送。
• @lru_cache(maxsize=2) 只缓存两个客户端实例，不缓存模型答案。

这里要区分两个“缓存”：@lru_cache 缓存的是 ChatOpenAI 客户端对象，只是避免每次请求重建连接；它不会跳过模型调用，也不会复用上一次生成的文本。

当前 V1 也没有必要立刻做最终答案缓存：FAQ 精确命中已经会直接返回，查询 embedding 和 FAQ/Doc 检索结果也已缓存。普通 RAG 的最终文本还要结合历史追问、权限域、知识库版本、Prompt Profile 和本次检索证据生成，生成完成后还要做引用补强和生成后核验。因此当前的边界是：缓存稳定的计算和证据，最终答案每次根据当前上下文生成并核验。

3.2 Message：让多轮对话结构稳定

LangChain 的消息对象对应 OpenAI API 的角色格式：

LangChain 对象	API role	项目用途
SystemMessage	system	设定助手身份、回答边界、风险约束
HumanMessage	user	用户问题、改写请求、检索上下文问题
AIMessage	assistant	历史回答，进入多轮上下文

项目文件：qa_core/memory/history.py

from langchain_core.messages import AIMessage, HumanMessage, SystemMessage

def add_turn(self, session_id: str, question: str, answer: str) -> None:
    history = self.for_session(session_id)
    history.add_messages([
        HumanMessage(content=question),
        AIMessage(content=answer),
    ])

def get_context_messages(self, session_id: str):
    recent = self.get_messages(session_id, limit=self.settings.history_recent_messages)
    summary = self.get_summary(session_id)
    if summary:
        return [SystemMessage(content=f"历史摘要：{summary}")] + recent
    return recent

需要理解：LLM 本身没有会话记忆。所谓“多轮对话”，本质是每次调用模型时，把必要的历史消息重新发给模型。

sequenceDiagram
    participant U as 用户
    participant API as QAService
    participant H as SQLChatMessageHistory
    participant LLM as ChatOpenAI

    U->>API: 入职流程有哪些步骤？
    API->>LLM: [System, Human]
    LLM-->>API: AIMessage
    API->>H: 保存 Human + AI

    U->>API: 那审批需要多久？
    API->>H: 读取最近历史
    API->>LLM: [System, Human, AI, Human]
    LLM-->>API: 能理解“审批”指入职审批

3.2.1 History 不等于模型自带记忆

LangChain 中常说的 Memory，不能理解成“模型自己记住了用户”。它的真实过程是：

本轮请求
  -> 从存储读取 session_id 对应的历史消息
  -> 组装成 SystemMessage / HumanMessage / AIMessage
  -> 再次发送给模型

SQLChatMessageHistory 负责把 LangChain 消息对象映射到关系型数据库。它解决的是消息的读写适配，不负责判断哪些历史与当前问题相关，也不负责检索知识库。

本项目的生产存储关系如下：

层次	实现	职责
LangChain 适配器	SQLChatMessageHistory	把 BaseMessage 读写到 SQL 表
项目存储封装	ChatHistoryStore	绑定 MySQL、session、最近消息和摘要
持久化表	chat_messages	保存完整会话消息，支持重启后恢复和审计
派生表	chat_session_summaries	保存旧消息压缩后的摘要
RAG 上下文	历史摘要 + 最近消息	只把必要上下文传给改写和回答链路

chat_messages 的核心字段可以简化为：

CREATE TABLE chat_messages (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    session_id VARCHAR(191) NOT NULL,
    message LONGTEXT NOT NULL,
    INDEX idx_chat_messages_session_id (session_id)
);

其中 session_id 是会话隔离边界，message 保存 LangChain 序列化后的消息内容。相同用户开启不同会话时，只要 session_id 不同，读取到的历史就不会混在一起。

下面的演示使用临时 SQLite 文件模拟同一套 LangChain 存储适配器。它验证四件事：写入两条消息、重新创建适配器后仍能读到、另一个 session 读不到、清空后消息消失：

from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage
from sqlalchemy import create_engine

engine = create_engine("sqlite:///history-demo.db")
history = SQLChatMessageHistory(
    session_id="session-a",
    connection=engine,
    table_name="demo_chat_messages",
)
history.add_messages([
    HumanMessage(content="新人入职流程有哪些？"),
    AIMessage(content="包括材料提交、合同签署和账号开通。"),
])

# 模拟进程重启：重新创建适配器，仍然能从数据库读出消息
reloaded = SQLChatMessageHistory(
    session_id="session-a",
    connection=engine,
    table_name="demo_chat_messages",
)
print(len(reloaded.messages))  # 2

项目实际代码的对应关系是：

ChatHistoryStore.for_session()
  -> SQLChatMessageHistory(session_id, connection=self.engine)
  -> MySQL chat_messages
  -> get_context_messages()
  -> 传给查询改写或最终回答 Prompt

本章 demo 为了离线可运行使用 SQLite；正式服务使用 MySQL，配置和表初始化由项目启动前置流程负责。历史摘要、最近消息窗口和追问改写的业务策略放在第 07 章展开，本章只建立“消息如何落库并重新装配”的基础认知。

3.3 Structured Output：把 LLM 输出变成业务对象

项目文件：qa_core/pipeline/query_variants.py

查询变体生成不能让模型自由发挥。自由文本会出现这些情况：

可以改写为：新人入职流程、入职办理步骤、入职 SOP。
["新人入职流程", "入职办理步骤"]
variants = 新人入职流程; 入职办理步骤

这些格式都不稳定。项目里使用 Pydantic 结构约束：

from pydantic import BaseModel, Field

class QueryVariants(BaseModel):
    variants: list[str] = Field(description="查询变体列表")

model = get_chat_model(streaming=False).with_structured_output(QueryVariants)
decision = model.invoke([
    SystemMessage(content="请为用户问题生成 2-3 个等价查询变体。"),
    HumanMessage(content="用户问题：入职流程有哪些步骤？"),
])

返回值不是字符串，而是一个 Pydantic 对象：

decision.variants         # ["入职流程有哪些步骤？", "新人入职流程", ...]

这一步是 LangChain 在项目中非常关键的价值：让 LLM 的输出进入可校验、可分支、可记录的工程世界。

在主项目中，意图识别的高频路径主要走确定性规则；结构化输出重点用于查询变体、改写和后续更复杂的模型结构化任务。这样能让主入口稳定，模型能力集中在更适合它的位置。

3.4 Prompt Profile：不是一个 Prompt 走天下

项目文件：

• qa_core/prompts/profiles.py
• qa_core/prompts/selector.py

项目没有把所有问题都塞进同一个 Prompt，而是按意图和风险类别选择不同模板：

PROMPT_PROFILES = {
    "FAQ_QUERY": PromptProfile(
        name="faq_answer",
        system_template=FAQ_ANSWER_SYSTEM_PROMPT,
        user_template=FAQ_ANSWER_USER_TEMPLATE,
        reason="FAQ 类问题优先复用标准答案，控制回答长度和业务口径。",
    ),
    "KNOWLEDGE_QUERY": PromptProfile(
        name="knowledge_answer",
        system_template=KNOWLEDGE_ANSWER_SYSTEM_PROMPT,
        user_template=KNOWLEDGE_ANSWER_USER_TEMPLATE,
        reason="业务知识咨询需要整合文档资料。",
    ),
    "FOLLOW_UP": PromptProfile(
        name="follow_up",
        system_template=FOLLOW_UP_ANSWER_SYSTEM_PROMPT,
        user_template=FOLLOW_UP_ANSWER_USER_TEMPLATE,
        reason="追问需要结合历史理解指代。",
    ),
}

可以这样理解：

• Prompt 不是一段固定文案，而是回答策略配置。
• system_template 控制助手身份、边界、风险口径。
• user_template 注入历史、检索上下文、用户问题。
• reason 进入调试信息，帮助解释为什么选择这个模板。

3.5 最终答案：ChatOpenAI.stream() 推给前端

项目文件：qa_core/pipeline/steps.py

from langchain_core.messages import HumanMessage, SystemMessage

def stream_llm_answer(system_prompt: str, user_prompt: str):
    llm = get_chat_model(streaming=True)
    return llm.stream([
        SystemMessage(content=system_prompt),
        HumanMessage(content=user_prompt),
    ])

项目主流程会把每个 chunk 转成 WebSocket token 事件：

for chunk in stream_llm_answer(answer_prepared.system_prompt, answer_prepared.user_prompt):
    token = str(getattr(chunk, "content", "") or "")
    if not token:
        continue
    yield build_token_event(token, context.session_id)

注意：这里不是 LangChain 替我们完成整个 RAG。LangChain 只负责模型流式调用；FAQ 直出、文档检索、上下文筛选、引用补强、写历史这些仍由项目代码控制。

---

第四部分：离线入库链路中的 LangChain

第 03 章只说明 LangChain 组件提供的统一接口，不展开项目入库实现。真实 Loader 注册表、Docling、metadata 标准化和 split_documents() 放在第 16 章；Milvus VectorStore 初始化与检索放在第 08 章。

flowchart LR
    A["业务文件"] --> B["Loader"]
    B --> C["Document"]
    C --> D["Splitter"]
    D --> E["Document chunks"]
    E --> F["VectorStore"]

4.1 Document：统一数据结构

from langchain_core.documents import Document

doc = Document(
    page_content="入职流程包括提交材料、签订合同和账号开通。",
    metadata={"source": "hr", "file_name": "入职制度.md"},
)

• page_content 保存用于切分、Embedding、BM25 和回答上下文的正文。
• metadata 保存来源、版本、权限、页码等治理字段。

4.2 Loader、Splitter 与 VectorStore 的接口边界

组件	输入	输出	本项目实现章节
Document Loader	PDF、Word、Markdown、表格等文件	list[Document]	第 16 章 Loader 注册表与 Docling
Text Splitter	list[Document]	更小的 list[Document]	第 16 章 split_documents()，附录 G 说明策略原理
VectorStore	Document 与查询文本	写入结果或检索候选	第 08 章 MilvusHybridStore

组件的价值是统一接口：上游文件格式可以不同，但进入切分前统一为 Document；切分策略可以变化，但写入 VectorStore 的对象仍然是 Document。LangChain 负责这些生态接口，项目代码负责版本、权限、质量门禁和业务编排。

---

第五部分：Runnable 和 LCEL 只作为统一接口理解

5.1 Runnable 的意义

Runnable 是 LangChain 的统一调用协议。无论 Prompt、Model、Parser，核心调用都收敛到三个方法：

方法	语义	本项目对应
invoke()	一个输入，返回完整结果	查询改写、查询变体结构化输出
stream()	一个输入，持续返回片段	最终答案逐 token 输出
batch()	多个输入，批量处理	批量测试、批量解析、离线评测可用

response = model.invoke([HumanMessage(content="入职流程有哪些步骤？")])

for chunk in model.stream([HumanMessage(content="入职流程有哪些步骤？")]):
    print(chunk.content, end="")

results = parser.batch([message_a, message_b, message_c])

5.2 LCEL 是线性链路语法，不是项目主流程

LCEL 可以把 Prompt、Model、Parser 串成线性管道：

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

prompt = ChatPromptTemplate.from_template("用一句话回答：{question}")
chain = prompt | model | StrOutputParser()

answer = chain.invoke({"question": "入职流程有哪些步骤？"})

它适合简单线性任务：

输入变量 -> Prompt -> Model -> Parser -> 输出

但本项目的 RAG 主流程不是一条直线：

flowchart TD
    Q["用户问题"] --> R["Stage 1 查询路由"]
    R -->|问候/越界/转人工| Direct["直接返回"]
    R -->|FAQ 精确命中| FAQDirect["FAQ 标准答案直出"]
    R -->|需要检索| Prep["意图识别 + 改写 + 检索计划"]
    Prep --> FAQ["FAQ 检索"]
    FAQ -->|高分命中| FAQAnswer["FAQ 直出"]
    FAQ -->|未直出| Doc["文档检索"]
    Doc --> Context["上下文构建"]
    Context -->|无可靠上下文| Insufficient["信息不足"]
    Context -->|有上下文| LLM["ChatOpenAI.stream"]
    LLM --> Save["写历史 + Trace"]

    style Direct fill:#FEF2F2,stroke:#DC2626
    style FAQAnswer fill:#ECFDF5,stroke:#059669
    style LLM fill:#EFF6FF,stroke:#2563EB

所以本项目选择显式编排，而不是高层 Chain：

route = decide_route(context)
if route.answer:
    return direct_answer

prepared = prepare_retrieval(context)
faq_result = search_faq(context, prepared)
doc_result = search_doc(context, prepared)
answer_prepared = prepare_answer(context, prepared, faq_result, doc_result)

for chunk in stream_llm_answer(system_prompt, user_prompt):
    yield token_event(chunk)

可以这样总结：

LCEL 很适合教“组件怎么串起来”，但企业 RAG 主链路有大量分支、阈值、提前退出和诊断信息，所以本项目不用一条 LCEL 链包到底。

---

第六部分：自研与生态的分工

6.1 交给 LangChain 的部分

能力	原因
LLM 客户端	多厂商 OpenAI-compatible 统一接口
Message 类型	对话历史结构稳定
结构化输出	Pydantic 约束，减少自由文本解析
SQLChatMessageHistory	省掉历史消息 CRUD
Document / Loader	文件解析后统一结构
Text Splitter	成熟切分策略，减少手写边界问题
VectorStore	统一向量库写入和检索入口

6.2 项目自己实现的部分

能力	为什么自己做
查询路由	要优先处理问候、转人工、越界、FAQ 精确命中
意图决策网关	高频确定场景由规则处理，检索类长尾由 BERT 模型增强，避免所有请求都调用 LLM
检索计划	不同意图、风险类别、source 需要不同参数
FAQ 直出	标准答案命中后不需要 LLM 生成
上下文筛选	要做分数阈值、重排、来源整理
Prompt Profile	不同业务类别要有不同口径
引用补强	企业 RAG 必须给出可追溯来源
Trace 和诊断	调试、验收和生产排障都需要可解释过程

这就是本项目的工程取舍：底层组件用生态，业务编排自己掌控。

---

第七部分：常见误区

7.1 误区一：用了 LangChain 就应该用 RetrievalQA

不对。RetrievalQA 适合快速 demo，但企业级 RAG 需要：

• FAQ 标准答案直出
• 意图识别
• 追问改写
• 多场景 source 过滤
• 知识库版本隔离
• 风险 Prompt Profile
• 引用来源补强
• Trace 和诊断面板

这些都很难塞进一个黑盒 Chain 里。

7.2 误区二：LCEL 越多越工程化

LCEL 适合表达线性链路，但不是所有流程都应该写成管道。复杂业务分支用显式函数更清楚，也更容易调试。

7.3 误区三：SemanticChunker 一定比 RecursiveCharacterTextSplitter 更好

不一定。企业 RAG 里的切分要可控、稳定、便宜、可复现。RecursiveCharacterTextSplitter 更适合作为默认方案；SemanticChunker 可以作为特殊文档的增强策略，而不是主链路默认切分器。

7.4 误区四：VectorStore 就是 Milvus

不是。VectorStore 是 LangChain 的抽象，Milvus 是具体后端。本项目选择 Milvus，是因为需要服务端 BM25、混合检索、多集合、多版本和企业级部署能力。

---

重点掌握

优先级	内容	要求
必会	LangChain 在本项目中的定位	生态适配器，不是业务编排核心
必会	ChatOpenAI 工厂	知道 streaming=True/False 分别用于哪里
必会	Message 类型	能解释多轮对话为什么要带历史消息
必会	with_structured_output()	能解释它如何让 LLM 输出变成业务对象
必会	Document / Loader / Splitter / VectorStore	能放回离线入库链路
理解	Runnable 的 invoke/stream/batch	知道统一调用协议
理解	LCEL 的适用边界	适合线性任务，不适合本项目完整 RAG 主链路
了解	SQLChatMessageHistory	知道它负责历史持久化，不负责检索

---

返回笔记开头 ↑