大模型技术之LangChain系统实战：从入门到RAG与Agent全栈通关

第一章 LangChain概述

1.1 从大模型到大模型应用：差的不只是一个API

近两年大语言模型爆发，几乎每个开发者都试过调用GPT、通义千问这类模型的API写个聊天Demo。但当你真的要做一个落地的AI应用——比如企业内部知识库、能查订单的智能客服、自动做数据分析的助手——就会发现：只靠裸调API远远不够。

你要处理多轮对话记忆、接入私有数据、调用外部工具、编排复杂业务流程……这些重复的底层工作，每个项目从头写一遍既低效又容易出错。LangChain就是为解决这个问题而生的：它是目前生态最成熟、应用最广的大语言模型应用开发框架，也是AI应用开发者的必备技能。

1.1.1 大模型与大语言模型

先厘清两个基础概念：

• 大模型(基础模型 Foundation Model)：具备海量参数与复杂结构的机器学习模型，能力覆盖自然语言处理、计算机视觉、语音识别等多个领域，是AI能力的“底座”。
• 大语言模型(Large Language Model, LLM)：大模型中专注于自然语言处理的分支，比如GPT系列、豆包、DeepSeek等。它通过海量文本训练，具备理解人类语言、生成连贯文本、逻辑推理的能力，也是当前AI应用最主流的载体。

我们常说的“大模型应用开发”，绝大多数都是基于大语言模型做上层应用，而LangChain就是连接“大模型能力”和“业务需求”的桥梁。

1.1.2 大模型行业的岗位分工

整个大模型产业链可以分为三层，对应不同的技术岗位：

1. 数据层：负责海量数据的清洗、标注、治理，对应数据开发工程师，偏数据工程方向。
2. 通用大模型层：做模型基座的训练、优化、对齐，对应大模型基座开发工程师，门槛极高，通常是算法博士、资深研究员的赛道。
3. 应用大模型层：基于现成的大模型基座，开发面向垂直场景的AI应用，对应大模型应用开发工程师，是目前行业需求最大、开发者最容易切入的方向。

我们学习LangChain，瞄准的就是大模型应用开发这个方向——不用从零训练模型，站在巨人的肩膀上做落地，性价比最高。

1.2 LangChain 到底是什么？

1.2.1 项目背景

LangChain 由哈佛大学的 Harrison Chase 于 2022 年 10 月发起开源，比 ChatGPT 正式发布还要早一个月，精准踩中了大模型应用爆发的风口，迅速成为GitHub上的顶流项目。

项目核心信息：

• GitHub 地址：https://github.com/langchain-ai/langchain
• 官方网站：https://www.langchain.com/langchain
• 官方文档：https://docs.langchain.com/oss/python/langchain/overview

1.2.2 一句话定位

LangChain ≠ 大模型。 如果用程序员熟悉的技术类比：

LangChain 之于 LLM，就像 Spring 之于 Java，Django 之于 Python。

它本身不生产大模型，而是一个“粘合剂”框架——把大模型、外部数据源、第三方工具、业务逻辑像链条一样串联起来，让你快速搭建出完整的AI应用，比如智能问答系统、Agent 助手、对话机器人、文档检索系统等等。

名字里的“Lang”对应 Language(语言模型)，“Chain”对应“链”，核心就是通过链式编排，把零散的能力组装成完整的应用。

1.3 主流大模型应用开发框架对比

LangChain不是唯一的选择，不同语言、不同场景下各有更适配的框架：

框架	语言	定位与适用场景
LangChain	Python	出现最早、生态最成熟，组件最丰富，适合复杂任务分解、单Agent应用开发，是最主流的选型
LlamaIndex	Python	深度聚焦索引与检索能力，在RAG(检索增强生成)场景下更专精，常和LangChain搭配使用
LangChain4J	Java	LangChain的Java生态版本，核心功能完整，适合Java技术栈的团队
Spring AI / Spring AI Alibaba	Java	Spring官方推出的AI开发套件，目前还在快速迭代中，对Spring体系友好但成熟度一般
Semantic Kernel	C##	微软官方推出，C#技术栈的首选，和Azure生态深度打通

如果你是Python开发者，想做通用的大模型应用开发，LangChain是性价比最高的入门选择。

1.4 为什么我们需要 LangChain？

很多人会问：我直接用HTTP请求调大模型API不也能做应用吗？为什么要额外学一个框架？

我们可以分三层来理解它的价值：

1.4.1 解决“模型切换成本高”的问题

不同厂商的大模型API接口规范、参数命名、调用方式各有差异，如果你的代码直接耦合某一家的SDK，后续想换模型就要重写一大片逻辑。

LangChain做了一层统一抽象：无论你用OpenAI、通义千问、DeepSeek还是本地开源模型，上层代码几乎不用改，只需要切换模型配置即可。这就像Java里的JDBC——不用管底层是MySQL还是Oracle，用一套接口就能操作。

1.4.2 解决“重复造轮子”的问题

做AI应用时，很多能力是通用的：提示词模板管理、对话记忆、文本分块、向量检索、工具调用、多步推理……每个项目都从零实现一遍，既耗时又容易出bug。

LangChain把这些通用能力都封装成了开箱即用的组件，你只需要关注业务逻辑，底层细节交给框架处理，开发效率能提升一个量级。

1.4.3 解决“复杂逻辑难编排”的问题

简单的问答是“一问一答”，但真实应用往往是多步骤的：比如先检索知识库、再调用工具、最后结合历史对话生成答案。这种多步骤流程如果用裸代码写，逻辑会非常混乱，难以维护和扩展。

LangChain的“链式”思想，让复杂流程变得结构化、可组合、可扩展，就像搭乐高一样，把组件拼起来就是完整应用。

总结下来，LangChain的核心价值就是：降低开发难度、统一模型接口、提供现成组件、规范业务流程。

1.5 LangChain 典型应用场景

LangChain能做的事情非常多，下面是工业界最常见的6类落地场景：

项目场景	核心技术点
文档问答助手	Prompt + Embedding + RetrievalQA
智能日程规划助手	Agent + Tool + Memory
LLM+数据库问答(NL2SQL)	SQLDatabaseToolkit + Agent
多模型路由对话系统	RouterChain + 多LLM
互联网智能客服	ConversationChain + RAG + Agent
企业知识库助手(RAG+本地模型)	VectorDB + LLM + Streamlit

举几个更容易理解的实际例子：

医院智能助手

可以回答科室位置、挂号规则，甚至帮用户预约号源，把医院的公开知识和挂号系统打通。

企业知识库系统

上传内部文档、员工手册、产品资料，员工用自然语言提问就能精准找到答案，不用再翻几百页的手册。

电商智能客服

对接商品库、订单系统，既能回答商品问题，也能查物流、处理售后，替代大量重复的客服工作。

1.6 LangChain 在大模型技术栈中的位置

整个大模型应用技术栈可以从上到下分为几层，LangChain处于开发框架层，承上启下：

• 上层：应用平台层：比如Dify、RAG Flow、MaxKB这类低代码/无代码的AI应用平台，可视化搭建知识库、Agent，底层很多都是基于LangChain做的封装。
• 中层：开发框架层：LangChain、Spring AI这类代码级开发框架，提供灵活的组件与编排能力，适合定制化开发。
• 下层：基础能力层：RAG、Agent、NL2SQL、多模态等具体技术方向，由LangChain的各个模块提供支撑。

基于LangChain，你可以开发出8类主流的AI项目：

1. 文档问答助手：加载内部文档，结合向量存储做语义搜索，回答员工/客户问题
2. 智能助理开发：基于Agent+工具，完成查天气、订机票、分析股票等复杂任务
3. 对话聊天机器人：带记忆的多轮对话，做个性化客服、聊天机器人
4. 数据分析与洞察：加载CSV/数据库数据，让大模型做趋势分析、生成报告
5. 多模态应用：结合文本、图片、语音，做图文问答、视频内容分析
6. 教育学习助手：结合题库、教材，做个性化答疑、公式讲解
7. 自动化工作流：自动生成文案、邮件，对接日历做任务调度
8. 研究与创新：分析文献、构建知识图谱，挖掘数据关联

1.7 LangChain 整体架构

LangChain早已不是单一的库，而是一个完整的生态全家桶，由四个核心部分组成：

1. LangChain：核心框架，提供基础组件与API，帮助开发者快速搭建Agent，兼容所有主流模型厂商。
2. LangGraph：面向Agent的低级别编排引擎，支持自定义Agent的每一步执行，内置记忆、人工介入能力，可以管理长时间运行的持久化任务，是做复杂Agent的核心利器。
3. LangSmith：AI应用的可观测性平台，提供链路追踪、效果评估、生产调试能力，帮助团队持续优化AI应用的效果。
4. Deep Agents：面向复杂任务的深度智能体架构，支持规划、子Agent调用、文件系统操作，对标Claude Code、Deep Research这类高级智能体应用。

1.7.1 Python 端核心包结构

Python版本的LangChain拆分成了多个包，按需安装，避免依赖冗余：

包名	作用
langchain-core	生态核心，定义所有基础接口与抽象规范
langchain	主入口包，包含构建LLM应用的绝大多数实现
langchain-text-splitters	专门的文本分割工具集，用于文档处理
langchain-mcp-adapters	MCP协议适配器，让LangChain可以对接MCP标准工具
langchain-tests	标准化测试套件，用于验证集成包的实现
langchain-classic	遗留版本的实现与组件，用于兼容老项目

1.8 大模型应用开发的四大经典范式

在正式写代码之前，我们先建立一个认知：大模型应用开发，本质上就是四种技术范式的组合与取舍。

范式1：纯 Prompt

最简单也最基础的模式：用户输入Prompt，大模型直接输出结果，一问一答，没有额外逻辑。适合简单的生成类任务，比如翻译、写文案。

范式2：Agent + Function Calling

如果说纯Prompt是“纸上谈兵”，那Agent就是能动手做事的“智能助理”。它充分利用大模型的推理决策能力，搭配记忆、工具、规划能力，自主拆解任务、一步步完成目标。

OpenAI的翁丽莲(Lilian Weng)在2023年首次提出了现代AI Agent的经典架构，可以用一个公式概括：

Agent = LLM + Memory + Tools + Planning + Action

当应用需要对接外部系统(比如查数据库、调API、发邮件)时，就需要Function Calling(工具调用)能力：大模型负责“思考”要不要调用工具、传入什么参数，执行完工具后再把结果返回给大模型，最终生成答案。

范式3：RAG(检索增强生成)

当大模型缺少领域知识、信息过时，或者需要用私有数据回答问题时，就用RAG：先把用户问题转成向量，去向量数据库里检索相关资料，再把资料和问题一起喂给大模型，让大模型基于参考资料生成答案。

打个通俗的比方：大模型考试遇到不会的题容易“瞎蒙”，RAG就是给它递了一本参考书，答题正确率能从60分提升到90分。

这是企业知识库、智能客服最常用的方案，成本低、见效快，数据也不会泄露给大模型厂商。

范式4：Fine-tuning(微调)

用业务数据继续训练大模型，让模型“记住”领域知识，适配特定风格。效果最好，但成本最高、周期最长，一般在前三种方案都达不到效果时才考虑。

技术选型路径

面对一个需求，该选哪种方案？行业通用的决策路径是这样的：

1. 先准备测试数据，验证纯Prompt能不能满足需求；
2. 如果需要补充领域/私有知识 → 上RAG；
3. 如果需要对接外部系统/工具 → 上Function Calling / Agent；
4. 如果以上都达不到预期，且有足够多的历史数据 → 考虑微调。

注意：绝大多数开发者都会跳过“准备测试数据”这一步，导致优化没有衡量标准，越调越乱。这是新手最容易踩的坑。

后续我们的学习，也会重点围绕RAG和Agent两大主流架构展开。

1.9 LangChain 四大核心组件

LangChain的功能模块很多，但核心可以归为四大件，也是我们后续章节的主线：

1. Model I/O：标准化大模型的输入输出，包括提示词模板、模型调用、输出解析。
2. Chains：用LCEL表达式把多个组件串成流程，实现链式调用、并行、分支、降级等编排能力。
3. Retrieval：对应RAG能力，覆盖文档加载、切分、向量化、存储、检索全流程。
4. Agents：让大模型具备自主规划、调用工具、多步执行的能力，处理开放式复杂任务。

---

第二章 Model I/O：搞定大模型的输入与输出

如果说LangChain是搭建AI应用的脚手架，那Model I/O就是脚手架里最底层的“水电接口”——所有和大模型的交互，最终都要落到“怎么把输入喂给模型”“怎么调用模型”“怎么把模型输出变成能用的数据”这三件事上。

Model I/O 是LangChain与语言模型交互的核心组件，完整覆盖了提示词模板(输入)→ 模型调用(处理)→ 输出解析(输出) 全链路。简单说，就是管好大模型的“进”和“出”。

它的核心逻辑可以拆成三步：

1. Format(格式化)：通过模板把原始数据包装成模型能理解的提示词，统一输入格式。
2. Predict(生成)：调用大模型接收输入，执行推理生成结果。
3. Parse(解析)：把模型返回的原始文本，转换成程序能用的结构化数据，比如JSON。

这一章我们就从“怎么调用模型”开始，一步步把输入输出的全流程讲透。

2.1 调用在线大模型

LangChain本身不生产大模型，它通过集成第三方模型服务来实现推理能力。工业界最常用的方式就是调用云端的大模型API，按需付费，不用自己搭建推理环境。

2.1.1 主流大模型服务平台

目前国内外有很多成熟的大模型API服务商，注册账号、充值、创建API Key之后，就能通过标准接口调用模型。常见的平台包括：

• CloseAI：https://platform.closeai-asia.com/
• OpenRouter：聚合上百种模型的中转平台，https://openrouter.ai/
• 阿里云百炼：阿里大模型服务平台，https://bailian.console.aliyun.com/
• 百度千帆：百度大模型开放平台，https://console.bce.baidu.com/qianfan/overview
• 硅基流动：主打高性价比开源模型推理，https://www.siliconflow.cn/

这些平台绝大多数都兼容OpenAI的API规范，这也是LangChain能做到“一套代码切换多模型”的基础。

2.1.2 原生 OpenAI SDK 调用

OpenAI定义的Chat Completion接口，事实上已经成了大模型API的行业标准。无论是通义千问、智谱清言还是DeepSeek，基本都遵循这套接口规范。

最基础的调用方式，就是直接用OpenAI官方SDK：

## pip install openai
from openai import OpenAI

client = OpenAI(
    base_url="https://openrouter.ai/api/v1",  ## 平台提供的接口地址
    api_key="sk-...",  ## 你的API密钥
)

completion = client.chat.completions.create(
    model="openai/gpt-oss-20b:free",  ## 模型名称
    messages=[{"role": "user", "content": "将'你好'翻译成意大利语"}],
)
print(completion.choices[0].message.content)

这种方式最原生，但所有逻辑都要自己封装，只适合简单的测试场景。

2.1.3 API Key 的正确管理方式

新手最容易犯的错误之一，就是把API Key直接写在代码里——一旦上传到公开代码仓库，密钥很快就会被爬虫盗刷，产生高额账单。

行业通用的密钥管理有三种方案，安全性从低到高：

1. 硬编码：直接写在代码里。仅适合临时本地测试，绝对不能提交到代码仓库。
2. .env 配置文件：用python-dotenv加载本地配置，配置文件加入.gitignore，是开发环境最常用的方案。
3. 系统环境变量：把密钥配置在操作系统环境变量中，适合生产环境部署。

最推荐的是.env方式，兼顾开发便捷性和安全性： 第一步，在项目根目录创建.env文件：

OPENAI_API_KEY="sk-你的密钥"
OPENAI_BASE_URL="https://openrouter.ai/api/v1"

第二步，在代码中加载并使用：

## pip install python-dotenv
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()  ## 自动加载.env文件中的环境变量

client = OpenAI()  ## 自动读取 OPENAI_API_KEY 和 OPENAI_BASE_URL
completion = client.chat.completions.create(
    model="openai/gpt-oss-20b:free",
    messages=[{"role": "user", "content": "将'你好'翻译成意大利语"}],
)
print(completion.choices[0].message.content)

只要环境变量名和OpenAI SDK约定的一致，甚至不需要手动赋值，SDK会自动读取，代码非常简洁。

2.1.4 LangChain 方式调用模型

在LangChain中，我们通常通过聊天模型接口(Chat Model)调用大模型。它的输入是消息列表，输出也是标准化的消息对象，比原生SDK更规范，也方便和其他组件串联。

LangChain提供了统一的初始化方法init_chat_model，可以快速接入不同厂商的模型：

import os
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage

## 初始化模型
llm = init_chat_model(
    model="openai/gpt-oss-20b:free",
    model_provider="openai",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
)

## 构造消息列表
messages = [
    SystemMessage(content="你是一个诗人"),
    HumanMessage(content="写一首关于春天的诗"),
]

## 调用模型
resp = llm.invoke(messages)
print(resp.content)  ## 模型生成的文本内容

调用后返回的是AIMessage对象，除了文本内容，还包含工具调用、元数据等信息，结构统一规范。

2.1.5 模型初始化核心参数

初始化模型时，有几个高频参数必须掌握：

参数	说明
model	模型名称，不同平台有不同的标识
base_url	API接口地址，由模型服务商提供
api_key	身份验证密钥
temperature	控制输出随机性，值越高越有创意，越低越确定，常用范围0-1
timeout	请求超时时间，单位为秒
max_tokens	限制输出的最大token数，控制回复长度和成本
max_retries	请求失败时的最大重试次数

这里补充一个基础概念：Token。 大模型处理文本的最小单位就是token，大致1个中文token对应1-1.8个汉字，1个英文token对应3-4个字母。模型厂商按token计费，输出也是逐个token生成的。如果想直观感受，可以使用OpenAI官方的Tokenizer工具做可视化统计。

2.1.6 四种消息类型

聊天模型的输入是消息列表，每条消息都有对应的角色(role)。LangChain定义了四种核心消息类型：

消息类型	作用
SystemMessage	系统指令，用于设定模型人设、规则、输出要求，引导模型行为
HumanMessage	用户输入，即用户发送的内容
AIMessage	模型的回复，包含生成的文本、工具调用信息等
ToolMessage	工具调用的结果，仅在工具调用场景下使用

除了用消息类构造，也可以直接用字典格式，和OpenAI原生格式完全兼容：

messages = [
    {"role": "system", "content": "你是个诗人"},
    {"role": "user", "content": "写首关于春天的诗"},
]
resp = llm.invoke(messages)

2.1.7 三类调用方式

LangChain的模型支持三类调用方式，分别对应不同的业务场景。

1. 非流式 vs 流式输出

• 非流式(默认行为)：等待模型生成完整答案后一次性返回。实现简单，适合后台任务、接口调用。

  resp = llm.invoke(messages)
  print(resp.content)
  

• 流式输出：模型每生成一个token就返回一个，像打字机一样逐字显示，用户体验更好，是对话类应用的标配。

  for chunk in llm.stream(messages):
      print(chunk.content, end="", flush=True)
  

2. 批量调用

一次性提交多个独立请求并行处理，适合批量生成、批量打标等IO密集型任务。

## 三组独立的对话
batch_messages = [
    [{"role": "user", "content": "写春天的诗"}],
    [{"role": "user", "content": "写夏天的诗"}],
    [{"role": "user", "content": "写秋天的诗"}],
]

results = llm.batch(batch_messages)

batch底层默认用线程池并行执行多个invoke，比循环串行调用快很多。

3. 同步 vs 异步

• 同步调用：invoke/stream/batch，请求发出后阻塞等待结果，代码简单直观。
• 异步调用：ainvoke/astream/abatch，带a前缀，配合asyncio使用，不阻塞主线程，适合高并发Web服务。

异步调用的性能优势非常明显：同时发起3个请求，同步调用耗时接近单个的3倍，异步调用几乎和单个请求耗时相当。

2.2 调用本地大模型

在线API方便，但有两个明显痛点：敏感数据要出域、长期使用成本不低。如果是隐私敏感场景，或者想节省推理成本，本地运行开源大模型是更好的选择。目前最流行的本地大模型运行工具，就是Ollama。

2.2.1 Ollama 是什么

Ollama是一个开源的本地大模型运行框架，专门针对Llama架构的开源模型做了深度优化，一键就能完成模型下载、启动和推理，把复杂的环境配置全部封装了起来。如今它已经和LangChain深度集成，是本地开发RAG、Agent的首选方案。

官方地址：https://ollama.com GitHub：https://github.com/ollama/ollama

2.2.2 安装 Ollama

Ollama跨平台支持Mac、Linux、Windows，安装非常简单：

• Windows/Mac：前往官网下载安装包，按向导下一步安装即可。

• Linux：一行命令自动完成全部安装：

  curl -fsSL https://ollama.com/install.sh | sh
  

  脚本会自动检查系统环境、下载二进制文件、配置系统服务并启动服务，全程无需手动干预。

2.2.3 下载与管理模型

安装完成后，在命令行执行ollama run 模型名，就能自动下载并启动模型。例如运行7B参数的DeepSeek推理模型：

ollama run deepseek-r1:7b

首次执行会自动下载模型文件，后续直接启动即可。全部支持的模型可以在官网模型库中检索。

如果默认存储路径空间不足，可以在设置中切换模型目录：

2.2.4 LangChain 调用本地 Ollama 模型

LangChain专门提供了langchain-ollama包对接Ollama，调用方式和在线模型几乎完全一致：

## pip install langchain-ollama
from langchain_ollama import ChatOllama

ollama_llm = ChatOllama(model="deepseek-r1:7b")
resp = ollama_llm.invoke("你好，请介绍一下你自己")
print(resp.content)

如果Ollama运行在其他机器上，或者修改了默认端口，只需指定base_url即可：

ollama_llm = ChatOllama(
    model="deepseek-r1:7b",
    base_url="http://你的IP:端口",
)

这正是LangChain统一接口的价值——从在线模型切换到本地模型，业务代码几乎不用改动。

2.3 Prompt Template 提示词模板

直接写死提示词的话，每次变更参数都要手动拼接字符串，非常不灵活。提示词模板(Prompt Template)就是用来解决这个问题的：把固定的提示框架做成模板，将变量留空，调用时动态填充，大幅提升提示词的复用性。

2.3.1 基础：str.format() 语法

LangChain的模板默认基于Python原生的str.format()语法，上手零成本。快速复习一下核心用法：

## 占位符替换
greeting = "Hello, {}!".format("Alice")

## 关键字参数
info = "Name: {name}, Age: {age}".format(name="Tom", age=25)

## 字典解包
person = {"name": "David", "age": 40}
info = "Name: {name}, Age: {age}".format(**person)

2.3.2 PromptTemplate：字符串模板

最基础的字符串提示模板，适合单轮、单角色的简单生成场景。

有两种常用的实例化方式：

from langchain_core.prompts import PromptTemplate

## 方式1：构造函数实例化
template = PromptTemplate(
    template="请评价{product}的优缺点，包括{aspect1}和{aspect2}。",
    input_variables=["product", "aspect1", "aspect2"],
)

## 方式2：from_template 快捷方法(更常用)
template = PromptTemplate.from_template("请给我一个关于{topic}的{type}解释。")

## 生成最终提示词
prompt = template.format(type="详细", topic="量子力学")
print(prompt)

如果部分变量有固定默认值，不用每次都传，可以使用部分变量预填充：

template = PromptTemplate.from_template("{foo} {bar}")
## 预先填充 foo 变量
partial_template = template.partial(foo="hello")
## 调用时只需要传剩下的 bar
prompt = partial_template.format(bar="world")
print(prompt)  ## hello world

2.3.3 ChatPromptTemplate：聊天模板

多角色、多轮对话的场景，需要用ChatPromptTemplate。它可以分别为system、human、ai角色设置模板，更贴合真实对话场景。

from langchain_core.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_messages([
    ("system", "你是一个AI开发工程师，你的名字是{name}。"),
    ("human", "你能帮我做什么?"),
    ("ai", "我能开发很多{thing}。"),
    ("human", "{user_input}"),
])

prompt = template.invoke({
    "name": "小谷AI",
    "thing": "AI应用",
    "user_input": "帮我写个RAG demo"
})

格式化之后会得到完整的消息列表，可以直接传给模型调用。

2.3.4 MessagesPlaceholder：消息占位符

做对话机器人时，历史对话是动态的、长度不固定，这时候就需要MessagesPlaceholder在模板中预留位置，运行时再把历史消息列表插入进去。

from langchain_core.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手。"),
    ("placeholder", "{conversation}"),  ## 预留历史对话的位置
])

## 运行时传入历史对话
prompt = template.format_messages(
    conversation=[
        ("human", "你好！"),
        ("ai", "想让我帮你做些什么？"),
        ("human", "能帮我做一个冰淇淋吗？"),
        ("ai", "不能"),
    ]
)

这是实现带记忆对话的核心组件，后续的Agent、对话链都会大量用到。

2.3.5 多模态提示模板

现在很多大模型都支持图文输入，模板同样支持传入图片链接：

from langchain_core.prompts import ChatPromptTemplate

template = ChatPromptTemplate.from_messages([
    {"role": "system", "content": "用中文简短描述图片内容"},
    {"role": "user", "content": [{"image_url": "{image_url}"}]},
])

prompt = template.format_messages(
    image_url="https://example.com/image.jpg"
)

2.3.6 外部文件加载Prompt

项目规模变大后，提示词写在代码里不利于维护。LangChain支持把提示词保存为JSON或YAML文件，运行时动态加载，方便产品、运营同学独立优化提示词，无需修改代码。

例如prompts/prompt.yaml：

_type: "prompt"
input_variables: ["name", "what"]
template: "请{name}讲一个{what}的故事"

加载并使用：

from langchain_core.prompts import load_prompt

template = load_prompt("prompts/prompt.yaml", encoding="utf-8")
print(template.format(name="年轻人", what="滑稽的"))

2.4 Output Parsers 输出解析器

大模型默认返回纯文本，但实际开发中我们往往需要结构化的数据，比如JSON、列表、数值。输出解析器(Output Parser)的作用，就是把模型的自由文本输出，转换成程序可以直接处理的结构化数据。

2.4.1 StrOutputParser：字符串解析

最简单也最常用的解析器，只提取消息对象中的content字段，把AIMessage转为纯字符串，搭配链式调用非常方便。

from langchain_core.output_parsers import StrOutputParser

## resp 是 llm.invoke 返回的 AIMessage 对象
str_resp = StrOutputParser().invoke(resp)
print(str_resp)  ## 纯文本内容

2.4.2 JsonOutputParser：JSON解析

如果需要模型返回结构化JSON，就用JsonOutputParser。它可以配合Pydantic定义数据结构，自动生成格式化指令注入提示词，还能对输出做类型校验。

使用分为四步：

1. 用Pydantic定义目标数据结构
2. 实例化JsonOutputParser
3. 将格式化指令放入系统提示词
4. 调用解析器将模型输出转为字典

from pydantic import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser

## 1. 定义数据结构
class Prime(BaseModel):
    prime: list[int] = Field(description="素数列表")
    count: list[int] = Field(description="小于对应素数的素数个数")

## 2. 创建解析器
json_parser = JsonOutputParser(pydantic_object=Prime)

## 3. 调用模型，把格式化指令放进提示词
resp = llm.invoke([
    {"role": "system", "content": json_parser.get_format_instructions()},
    {"role": "user", "content": "生成5个1000到100000之间的素数，并标出小于它的素数个数"}
])

## 4. 解析结果
result = json_parser.invoke(resp)
print(result)
## {'prime': [1009, 2003, 3001, 4001, 5003], 'count': [168, 303, 430, 584, 669]}

2.5 Structured Outputs 强制结构化输出

上面的JsonOutputParser是“靠提示引导模型输出JSON”，但模型不一定每次都严格遵守，可能输出多余文本导致解析失败。而with_structured_output基于模型原生的工具调用能力实现结构化输出，可靠性要高得多。

LangChain支持三种定义结构的方式：

2.5.1 TypedDict

Python内置类型，写法最简单，但没有字段校验功能，适合快速原型：

from typing import TypedDict, Annotated

class Animal(TypedDict):
    animal: Annotated[str, "动物名称"]
    emoji: Annotated[str, "动物表情"]

class AnimalList(TypedDict):
    animals: Annotated[list[Animal], "动物列表"]

## 绑定结构化输出
llm_structured = llm.with_structured_output(AnimalList)
result = llm_structured.invoke("生成三种动物和对应的emoji")
print(result)  ## 直接返回字典

2.5.2 Pydantic

功能最完整的方案，支持字段描述、类型校验、嵌套结构，是生产环境最推荐的方式：

from pydantic import BaseModel, Field

class Animal(BaseModel):
    animal: str = Field(description="动物名称")
    emoji: str = Field(description="动物表情")

class AnimalList(BaseModel):
    animals: list[Animal] = Field(description="动物列表")

llm_structured = llm.with_structured_output(AnimalList)
result = llm_structured.invoke("生成三种动物和对应的emoji")
print(result)  ## 返回Pydantic对象，可以直接用.访问属性

2.5.3 JSON Schema

如果需要最大程度的自定义，或者和其他系统对接，可以直接传入原始的JSON Schema：

schema = {
    "name": "animal_list",
    "schema": {
        "type": "array",
        "items": {
            "type": "object",
            "properties": {
                "animal": {"type": "string", "description": "动物名称"},
                "emoji": {"type": "string", "description": "动物表情"},
            },
            "required": ["animal", "emoji"],
        },
    },
}

llm_structured = llm.with_structured_output(
    schema,
    method="json_schema",
    include_raw=True  ## 同时返回原始响应和解析结果
)

---

第三章 Chains与LCEL：用表达式语言编排AI工作流

第二章我们掌握了提示词模板、模型调用、输出解析三大基础组件，但单次模型调用只能完成单一任务。真实的AI应用往往是多步骤的流水线：比如先检索知识库再生成答案、先翻译文本再总结摘要、先分析用户意图再调用对应能力。

把多个组件按业务逻辑串联成可执行的工作流，就是链(Chain)。而LangChain官方推荐的链编排方式，就是LCEL(LangChain Expression Language，LangChain表达式语言)——用类似Linux管道符的语法，像搭积木一样把组件拼接起来，天然支持流式、异步、批处理、重试、降级等企业级能力。

3.1 认识LCEL与Runnable

3.1.1 从手动拼接到声明式编排

不用LCEL的时候，我们要手动调用每个组件，把前一步的结果传给下一步：

## 手动拼接的写法
prompt_value = prompt_template.invoke({"topic": "大语言模型"})
ai_message = llm.invoke(prompt_value)
result = output_parser.invoke(ai_message)

步骤少的时候还能应付，一旦逻辑变复杂，代码会变得冗长混乱，流式、重试、监控等能力也要自己逐一实现。

LCEL的核心改进就是用管道符 | 声明组件的执行顺序，把多步逻辑浓缩成一行：

## LCEL 写法
chain = prompt_template | llm | output_parser
result = chain.invoke({"topic": "大语言模型"})

管道符的本质很简单：前一个组件的输出，自动作为后一个组件的输入。所有组件遵循统一的接口规范，所以可以自由拼接。

3.1.2 Runnable：所有组件的统一接口

LCEL能实现自由拼接，核心在于所有组件都实现了同一个抽象——Runnable(可运行对象)。只要是Runnable，就支持统一的调用方法，也就能用|串联。

LangChain中绝大多数核心组件都是Runnable，包括：

• 提示词模板(PromptTemplate / ChatPromptTemplate)
• 聊天模型(ChatModel)与大语言模型(LLM)
• 输出解析器(OutputParser)
• 检索器(Retriever)、工具(Tool)
• 后续章节会讲到的记忆、Agent等

所有Runnable都统一支持五类调用方式：

方法	作用	输入	输出
invoke()	同步单次调用	单个输入	单个结果
stream()	同步流式调用	单个输入	逐块返回结果
batch()	同步批量调用	多个输入	多个结果
ainvoke() / astream() / abatch()	对应异步版本	同同步	同同步

这意味着：单个组件支持的能力，拼成链之后天然全部支持，不需要额外开发。

3.2 构建第一条基础链

最经典也最常用的链，就是「提示词模板 → 大模型 → 输出解析器」三段式结构，我们用一个翻译场景来完整实现。

3.2.1 完整实现

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

## 1. 定义组件
prompt = ChatPromptTemplate.from_template(
    "请将以下文本翻译成{language}：\n{text}"
)
llm = ChatOpenAI(model="gpt-3.5-turbo")
parser = StrOutputParser()

## 2. 用管道符拼接成链
translate_chain = prompt | llm | parser

## 3. 调用链
result = translate_chain.invoke({
    "language": "法语",
    "text": "人工智能正在改变世界"
})
print(result)

执行流程是自动的：

1. 字典输入传给prompt，填充模板得到格式化的消息列表
2. 消息列表传给llm，调用模型得到AIMessage
3. AIMessage传给parser，提取出纯文本字符串返回

3.2.2 链的四种调用方式

和单个模型一样，整条链也同步支持四种调用模式，完全不用修改链的定义。

1. 流式输出

对话类场景必备，逐字返回翻译结果：

for chunk in translate_chain.stream({
    "language": "日语",
    "text": " LangChain 非常好用"
}):
    print(chunk, end="", flush=True)

2. 批量处理

一次性处理多条翻译任务，自动并行执行：

inputs = [
    {"language": "英语", "text": "你好世界"},
    {"language": "韩语", "text": "早上好"},
    {"language": "西班牙语", "text": "谢谢"},
]
results = translate_chain.batch(inputs)

3. 异步调用

配合Web框架(如FastAPI)使用，不阻塞主线程，提升并发能力：

import asyncio

async def main():
    result = await translate_chain.ainvoke({
        "language": "德语",
        "text": "异步调用"
    })
    print(result)

asyncio.run(main())

3.3 核心Runnable：构建复杂业务逻辑

只有顺序串联还不够，真实业务往往需要并行分支、条件判断、数据透传、自定义处理等能力。LangChain提供了一组核心Runnable组件，专门用来实现复杂工作流。

3.3.1 RunnablePassthrough：透传输入

默认情况下，前一步的输出会完全覆盖输入。如果后续步骤还需要用到原始输入参数，就用RunnablePassthrough把原始数据透传下去。

最常见的场景：RAG中既要把检索到的上下文传给模型，也要保留用户的原始问题。

from langchain_core.runnables import RunnablePassthrough

## 模拟一个检索函数
def fake_retriever(query):
    return f"关于{query}的知识库内容：..."

## 并行执行：检索 + 透传原始问题
chain = (
    {"context": fake_retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | parser
)

chain.invoke("什么是LangChain")

执行后，prompt模板会同时收到context和question两个变量。

简写技巧：字典本身会被LangChain自动识别为并行执行的RunnableParallel，所以上面的写法等价于显式声明RunnableParallel。

3.3.2 RunnableParallel：并行多分支

当同一个输入需要走多个独立分支处理，最后合并结果时，用RunnableParallel(也叫RunnableParallel)实现并行执行。

比如同时生成一篇文章的摘要、关键词和标题：

from langchain_core.runnables import RunnableParallel

## 三个独立的子链
summary_chain = ChatPromptTemplate.from_template("总结以下内容的摘要：{text}") | llm | parser
keyword_chain = ChatPromptTemplate.from_template("提取以下内容的5个关键词：{text}") | llm | parser
title_chain = ChatPromptTemplate.from_template("给以下内容起一个标题：{text}") | llm | parser

## 并行执行三个链
parallel_chain = RunnableParallel(
    summary=summary_chain,
    keywords=keyword_chain,
    title=title_chain
)

result = parallel_chain.invoke({"text": "LangChain是一个大语言模型应用开发框架..."})
print(result["title"])
print(result["summary"])
print(result["keywords"])

三个子链会并行执行，总耗时接近最慢的单个任务，而不是三者相加，大幅提升效率。

3.3.3 RunnableLambda：自定义逻辑

想把普通Python函数插入到链里做数据处理时，用RunnableLambda包装一下，就能变成标准Runnable参与拼接。

比如对模型输出做后处理，或者对输入做清洗：

from langchain_core.runnables import RunnableLambda

## 自定义处理函数
def to_upper(text):
    return text.upper()

## 包装成Runnable
upper_lambda = RunnableLambda(to_upper)

## 接入链：翻译后转成大写
chain = prompt | llm | parser | upper_lambda

还可以用lambda简写：

chain = prompt | llm | parser | RunnableLambda(lambda x: x.strip())

这是做输入输出数据转换最灵活的方式。

3.3.4 RunnableBranch：条件分支路由

根据输入的不同，走不同的处理链路，用RunnableBranch实现条件判断。它接收一组(条件函数, 对应链路)的配对，以及一个默认分支，按顺序匹配第一个满足条件的分支。

比如根据问题类型调用不同的专家模板：

from langchain_core.runnables import RunnableBranch

## 定义分支条件
def is_code_question(x):
    return "代码" in x["question"] or "编程" in x["question"]

## 不同分支的链
code_chain = ChatPromptTemplate.from_template("你是资深程序员，请回答：{question}") | llm | parser
medical_chain = ChatPromptTemplate.from_template("你是执业医生，请回答：{question}") | llm | parser
default_chain = ChatPromptTemplate.from_template("请回答：{question}") | llm | parser

## 分支路由
branch_chain = RunnableBranch(
    (is_code_question, code_chain),
    (lambda x: "医疗" in x["question"], medical_chain),
    default_chain  ## 都不匹配时走默认
)

result = branch_chain.invoke({"question": "Python怎么实现快速排序"})

这是实现简单意图路由的核心组件。

3.4 链的增强能力

LCEL除了编排逻辑，还提供了一系列链式方法，可以给链附加重试、降级、配置、监控等能力，不需要修改业务逻辑。

3.4.1 运行时配置：with_config

同一条链可以在不同场景下使用不同参数，比如切换模型、调整温度、设置超时，用with_config动态注入配置。

## 基础链
base_chain = prompt | llm | parser

## 衍生出创意模式的链
creative_chain = base_chain.with_config(
    configurable={"temperature": 0.9, "model": "gpt-4"}
)

## 衍生出严谨模式的链
precise_chain = base_chain.with_config(
    configurable={"temperature": 0.1, "max_tokens": 500}
)

3.4.2 失败重试：with_retry

调用在线API难免遇到网络波动、限流等临时错误，with_retry可以自动重试，提升链路稳定性。

## 最多重试3次
retry_chain = translate_chain.with_retry(max_attempts=3)

可以自定义哪些异常需要重试，哪些直接抛出，灵活适配业务。

3.4.3 降级容灾：with_fallbacks

当主链路彻底不可用时(比如在线API挂了)，可以自动切换到备用方案，比如降级到本地模型。

from langchain_ollama import ChatOllama

local_llm = ChatOllama(model="qwen2:7b")
fallback_chain = prompt | local_llm | parser

## 主链失败时自动走备用链
robust_chain = translate_chain.with_fallbacks([fallback_chain])

这是生产环境保障服务可用性的常用手段。

3.4.4 回调与监控

给链添加回调函数，可以记录每一步的输入输出、耗时、错误，用于日志、监控、计费统计。LangChain内置了很多回调处理器，也支持自定义。

from langchain_core.tracers import ConsoleCallbackHandler

## 控制台打印详细执行过程，适合调试
chain.invoke(
    {"language": "英语", "text": "回调测试"},
    config={"callbacks": [ConsoleCallbackHandler()]}
)

3.5 实战：多步骤内容创作链

我们用一个完整的案例把以上知识点串起来：实现一个小红书文案生成链，输入产品名，自动输出卖点、正文、标题三部分结构化结果。

3.5.1 流程拆解

1. 输入：产品名称
2. 并行生成：产品卖点 + 适用人群
3. 基于卖点和人群生成小红书正文
4. 基于正文生成3个备选标题
5. 最终返回结构化的JSON结果

3.5.2 完整代码

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
parser = StrOutputParser()

## 1. 生成卖点
selling_point_prompt = ChatPromptTemplate.from_template(
    "请分析{product}的3个核心卖点，每条不超过20字"
)
selling_point_chain = selling_point_prompt | llm | parser

## 2. 生成适用人群
audience_prompt = ChatPromptTemplate.from_template(
    "请列出{product}的3类核心目标用户群体"
)
audience_chain = audience_prompt | llm | parser

## 3. 生成正文
content_prompt = ChatPromptTemplate.from_template(
    """请根据以下信息写一篇小红书种草文案，300字左右，语气活泼有感染力：
    产品：{product}
    卖点：{selling_points}
    目标人群：{audience}
    """
)
content_chain = content_prompt | llm | parser

## 4. 定义输出结构
class NoteResult(BaseModel):
    product: str = Field(description="产品名称")
    selling_points: str = Field(description="核心卖点")
    audience: str = Field(description="目标人群")
    content: str = Field(description="小红书正文")
    titles: list[str] = Field(description="3个备选标题")

json_parser = JsonOutputParser(pydantic_object=NoteResult)

## 5. 生成标题并组装最终结果
final_prompt = ChatPromptTemplate.from_template(
    """基于以下内容生成3个吸引人的小红书标题，然后按要求返回完整JSON：
    产品：{product}
    卖点：{selling_points}
    目标人群：{audience}
    正文：{content}
    
    {format_instructions}
    """
)

## 完整链路
full_chain = (
    ## 第一步：并行生成卖点和人群，同时透传产品名
    RunnableParallel(
        product=RunnablePassthrough(),
        selling_points=selling_point_chain,
        audience=audience_chain
    )
    ## 第二步：生成正文，同时保留前面的所有字段
    | RunnableParallel(
        product=lambda x: x["product"],
        selling_points=lambda x: x["selling_points"],
        audience=lambda x: x["audience"],
        content=content_chain
    )
    ## 第三步：注入格式化指令，生成最终结构化结果
    | RunnablePassthrough.assign(
        format_instructions=lambda _: json_parser.get_format_instructions()
    )
    | final_prompt
    | llm
    | json_parser
)

## 调用
result = full_chain.invoke("便携咖啡机")
print(result)

---

第四章 Retrieval：用检索增强给大模型装上外挂知识库

大模型的语言生成能力很强，但存在两个天然短板：一是知识有截止日期，无法获取实时信息与最新数据；二是无法访问企业内部文档、个人笔记等私有数据，直接提问极易产生「幻觉」，编造虚假答案。

行业解决这一问题的主流方案是 RAG(Retrieval-Augmented Generation，检索增强生成)：先从知识库中检索出与问题相关的内容，再将内容连同问题一起交给大模型，让模型基于真实资料生成答案。LangChain 的 Retrieval 模块是实现 RAG 的核心技术底座，完整覆盖「文档加载 → 文档切分 → 向量化 → 向量存储 → 语义检索」全流程。

4.1 RAG 的完整工作流程

RAG 整体分为离线构建知识库和在线检索生成两大阶段，对应多个核心执行步骤，完整架构如下图所示：

离线阶段：把文档变成可检索的向量库

这一步是一次性前置准备，把原始文档处理成向量数据库索引：

1. 文档加载：读取 PDF、Word、Markdown、网页等不同格式的文档，转换成统一的文本对象。
2. 文档切分：将长文档切割成若干长度适中的文本块(Chunk)，兼顾语义完整性与检索精度。
3. 向量化：通过嵌入模型(Embedding Model)将每个文本块转换成高维语义向量。
4. 向量存储：将向量与原始文本、元数据存入向量数据库，建立索引，支持快速相似度检索。

在线阶段：根据用户问题检索并生成答案

用户提问时实时执行，是日常访问的主流程：

1. 问题向量化：将用户的提问用同一个嵌入模型转换成向量，保证向量空间一致。
2. 语义检索：在向量库中计算相似度，召回与问题最相关的 Top N 个文本块。
3. 提示词拼接：将检索到的相关内容作为上下文，和用户问题一起填入提示词模板。
4. 模型生成：大模型基于给定的上下文，生成准确、有据可依的回答。

简单来说，检索模块的核心就是帮大模型「先查资料再回答」，从根源上减少幻觉，同时让大模型能用上私有数据。

4.2 Document Loaders：文档加载器

不同格式的文档，读取方式千差万别。LangChain 提供了上百种文档加载器(Document Loader)，可以一键将各类格式的内容转换成统一的 Document 对象，屏蔽底层格式差异。

4.2.1 Document 对象

所有加载器输出的都是标准 Document 对象，结构非常简洁，只有两个核心字段：

• page_content：文档的文本内容，字符串类型。
• metadata：文档的元数据，字典类型，存放文件名、页码、作者、来源等附加信息。

## Document 对象结构示意
Document(
    page_content="这是文档的正文内容",
    metadata={"source": "产品手册.pdf", "page": 3, "author": "技术部"}
)

元数据的作用非常关键：后续检索时可以基于元数据做过滤，生成答案时也可以溯源到具体的文档页码，实现答案可追溯。

4.2.2 常用加载器

LangChain 的加载器大多集成在 langchain_community 包中，下面介绍几种最高频使用的加载器。

1. 纯文本 / Markdown 加载

最基础的加载器，读取 .txt、.md 等纯文本文件，适合结构化的笔记、文档。

from langchain_community.document_loaders import TextLoader

loader = TextLoader("README.md", encoding="utf-8")
docs = loader.load()  ## 返回 Document 列表

2. PDF 文档加载

PDF 是企业最常见的私有文档格式，LangChain 支持多种解析后端，各有优劣：

加载器	特点	适用场景
PyPDFLoader	纯 Python 实现，无额外依赖，解析速度一般	普通文本型 PDF
PyMuPDFLoader	解析速度快、精度高，支持提取页码	绝大多数常规 PDF
UnstructuredPDFLoader	支持复杂排版、扫描版 PDF	格式复杂的富文本/扫描件

最常用的是 PyMuPDFLoader：

## pip install pymupdf
from langchain_community.document_loaders import PyMuPDFLoader

loader = PyMuPDFLoader("产品手册.pdf")
docs = loader.load()
## 每页 PDF 生成一个 Document，metadata 自带页码信息

3. 网页内容加载

直接从 URL 加载网页正文，适合做实时资讯问答、网页摘要等场景。

## pip install beautifulsoup4
from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://example.com/article")
docs = loader.load()

4. 目录批量加载

一次性加载整个目录下的所有指定类型文件，适合批量构建知识库。

from langchain_community.document_loaders import DirectoryLoader

## 加载 docs 目录下所有 .md 文件
loader = DirectoryLoader(
    path="./docs",
    glob="**/*.md",
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"}
)
docs = loader.load()

4.3 Text Splitters：文档分割器

加载后的完整文档往往很长，直接向量化会丢失细节，也无法全部塞进模型的上下文窗口。因此需要将长文档切割成多个大小合适的文本块(Chunk)，这一步是影响 RAG 检索效果的关键环节。

4.3.1 分割的核心原则

切分粒度是一把双刃剑，需要根据场景权衡：

• 块太大：单个块包含的信息过多，检索精准度下降，容易引入无关内容。
• 块太小：单个块语义不完整，模型无法理解上下文，回答质量下降。
• 块重叠：相邻块之间保留部分重叠内容，避免语义被截断，保证上下文连贯性。

工业界常用经验值：

• 普通文档：chunk_size=500~1000 字符，chunk_overlap=50~200 字符
• 长文问答：chunk_size=1000~2000 字符，chunk_overlap=200~400 字符

4.3.2 常用分割器

1. RecursiveCharacterTextSplitter：递归字符分割

这是最通用、最推荐的默认分割器。它会按照一组分隔符的优先级递归尝试切割，优先在段落、句子、单词的边界处断开，尽可能保证语义完整。

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        ## 每个块的最大字符数
    chunk_overlap=100,     ## 相邻块的重叠字符数
    separators=["\n\n", "\n", "。", "！", "？", " ", ""],  ## 中文场景优化
    length_function=len
)

## 对文档列表进行切分
chunks = splitter.split_documents(docs)
print(f"切分后共 {len(chunks)} 个文本块")

2. MarkdownHeaderTextSplitter：按标题层级分割

如果是结构化的 Markdown 文档，按标题分割是更合理的方式：它会保留标题层级信息到元数据中，让每个文本块都自带上下文归属，检索效果远优于普通字符分割。

from langchain_text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "一级标题"),
    ("##", "二级标题"),
    ("###", "三级标题"),
]

splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False
)

chunks = splitter.split_text(markdown_content)
## 每个 chunk 的 metadata 会自动带上所属的各级标题

3. TokenTextSplitter：按 Token 数分割

如果希望严格控制每个块的 Token 数量，贴合模型的上下文窗口，可以使用按 Token 计数的分割器。

from langchain_text_splitters import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=200,
    chunk_overlap=20
)
chunks = splitter.split_documents(docs)

4.4 Embeddings：嵌入模型

文本是离散的符号，无法直接计算相似度。嵌入模型(Embedding Model)的作用，就是把一段文本转换成固定维度的高维向量，让语义相近的文本，在向量空间中的距离也更近。这是实现语义检索的数学基础。

4.4.1 常用嵌入模型

嵌入模型的质量直接决定检索的准确率，目前主流方案分为在线 API 和本地开源两类。

类型	代表模型	特点
在线 API	OpenAI text-embedding-3-small、text-embedding-ada-002	效果稳定，无需本地部署，按调用量付费
本地开源	bge-m3、m3e、gte 系列	中文效果优秀，免费开源，隐私性好

4.4.2 在线嵌入模型调用

以 OpenAI 兼容的嵌入接口为例，调用方式非常简洁：

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    base_url="https://openrouter.ai/api/v1"
)

## 单段文本转向量
vector = embeddings.embed_query("什么是 LangChain？")
print(f"向量维度：{len(vector)}")

## 批量文本转向量
vectors = embeddings.embed_documents(["文本1", "文本2", "文本3"])

4.4.3 本地嵌入模型调用

隐私敏感场景可以用 Ollama 运行开源嵌入模型，和本地大模型的调用逻辑完全一致：

## pip install langchain-ollama
from langchain_ollama import OllamaEmbeddings

## 使用 bge-m3 开源嵌入模型
embeddings = OllamaEmbeddings(model="bge-m3")

vector = embeddings.embed_query("人工智能的发展历程")

重要注意：检索和入库必须使用同一个嵌入模型，否则向量空间不一致，检索结果会完全失效。

4.5 Vector Stores：向量数据库

普通关系型数据库不擅长高维向量的相似度检索，向量数据库(Vector Store)就是专门为存储、索引、检索向量而生的数据库。它能在百万级数据量下，实现毫秒级的相似度查询。

4.5.1 主流向量数据库对比

LangChain 集成了数十种向量数据库，不同场景选择不同：

类型	代表产品	特点	适用场景
轻量本地	FAISS、Chroma	零依赖，本地文件存储，无需启动服务	开发测试、小型项目、个人知识库
独立服务	Milvus、Weaviate、Qdrant	分布式架构，支持海量数据，高性能	中大型企业级生产环境
数据库扩展	PGVector	基于 PostgreSQL，兼顾结构化与向量检索	已有 PG 技术栈的团队

4.5.2 FAISS 快速上手

FAISS 是 Meta 开源的向量检索库，性能优秀、开箱即用，是学习和小型项目的首选。我们用它演示完整的入库流程。

1. 构建向量库

## pip install faiss-cpu
from langchain_community.vectorstores import FAISS

## 一步完成：文本块 → 向量化 → 存入向量库
db = FAISS.from_documents(
    documents=chunks,
    embedding=embeddings
)

2. 持久化与加载

FAISS 支持将索引保存到本地磁盘，下次启动直接加载，不用重复向量化：

## 保存到本地
db.save_local("./faiss_index")

## 从本地加载
db = FAISS.load_local(
    folder_path="./faiss_index",
    embeddings=embeddings,
    allow_dangerous_deserialization=True
)

3. 基础相似度查询

向量库最基础的用法就是根据文本查询最相关的内容：

query = "LangChain 支持哪些大模型？"
related_docs = db.similarity_search(query, k=3)

for i, doc in enumerate(related_docs):
    print(f"--- 第 {i+1} 条相关内容 ---")
    print(doc.page_content)
    print(doc.metadata)

4.6 Retrievers：检索器

检索器(Retriever)是向量库的上层封装，提供统一的检索接口，支持多种检索策略，并且本身就是 Runnable，可以直接接入 LCEL 链路。

4.6.1 获取检索器

任何向量库都可以通过 as_retriever() 方法转换成标准检索器：

retriever = db.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

检索器的标准调用方法是 invoke()，输入查询文本，返回相关文档列表：

docs = retriever.invoke("什么是 RAG？")

因为它是标准 Runnable，所以可以直接用在 LCEL 管道中，和提示词、模型无缝拼接。

4.6.2 进阶检索策略

除了基础的相似度检索，LangChain 还支持多种进阶检索方式，进一步提升召回质量。

1. MMR 最大边际相关性检索

普通相似度检索容易返回内容高度重复的结果。MMR(Maximal Marginal Relevance)在相关性的基础上兼顾多样性，既能保证召回准确，又能避免信息冗余。

retriever = db.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,
        "fetch_k": 20,
        "lambda_mult": 0.5
    }
)

2. 相似度阈值检索

只返回相似度高于设定阈值的结果，过滤掉不相关的内容，从源头减少模型被无关信息干扰的概率。

retriever = db.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "score_threshold": 0.7,
        "k": 10
    }
)

3. 元数据过滤检索

检索时可以根据元数据限定范围，比如只检索某个文档、某个分类下的内容，大幅提升检索精准度。

retriever = db.as_retriever(
    search_kwargs={
        "k": 4,
        "filter": {"source": "产品手册.pdf"}
    }
)

4.7 实战：从零搭建 RAG 问答链

我们把本章和第三章的知识点结合起来，用 LCEL 搭建一条完整的 RAG 问答链，实现「基于本地文档回答问题」的能力。

完整实现代码

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_ollama import OllamaEmbeddings, ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

## ========== 1. 构建知识库 ==========
## 加载文档
loader = TextLoader("产品说明文档.txt", encoding="utf-8")
docs = loader.load()

## 切分文档
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
chunks = splitter.split_documents(docs)

## 构建向量库，获取检索器
embeddings = OllamaEmbeddings(model="bge-m3")
db = FAISS.from_documents(chunks, embeddings)
retriever = db.as_retriever(search_kwargs={"k": 3})

## 文档格式化函数
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

## ========== 2. 构建 RAG 问答链 ==========
## 设计 RAG 提示词模板
rag_prompt = ChatPromptTemplate.from_template("""
你是一个专业的文档助手，请仅根据下方提供的上下文回答用户的问题。
如果上下文中没有相关信息，请直接回答「抱歉，文档中没有找到相关内容」，不要编造答案。

【上下文】
{context}

【用户问题】
{question}
""")

## 初始化大模型
llm = ChatOllama(model="qwen2:7b")

## 拼接完整链路
rag_chain = (
    ## 并行执行：检索上下文 + 透传原始问题
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

## ========== 3. 调用测试 ==========
answer = rag_chain.invoke("这款产品的保修期是多久？")
print(answer)

链路执行流程

整条链的执行顺序是自动完成的：

1. 用户输入的问题字符串，同时传给 retriever 和 RunnablePassthrough。
2. retriever 从向量库检索出相关文档，格式化后赋值给 context。
3. 原始问题赋值给 question，两者一起填入提示词模板。
4. 大模型基于上下文生成回答，最终解析为纯字符串返回。

---

第五章 Agents：赋予大模型自主决策与行动能力

前面章节讲解的 Chain 是预设好的固定执行流水线——检索、生成、格式转换，每一步的执行顺序与逻辑都由开发者提前编码确定。这种模式在处理确定性任务时高效可靠，但面对开放动态的复杂问题时会明显受限：比如需要先获取实时数据再做计算、需要多步试错才能得到答案、需要根据用户意图动态调用不同业务接口。

这类需要动态决策、多步迭代、调用外部能力的任务，正是 Agent(智能体) 的核心场景。Agent 以大语言模型为推理大脑，以工具为执行手脚，能够自主规划执行路径、调用外部能力、根据反馈迭代调整，最终完成复杂任务。

5.1 Agent 介绍

5.1.1 从固定流程到自主决策

Chain 和 Agent 最本质的区别，在于执行路径的决策权归属：

• Chain：执行路径由开发者硬编码定义，模型只负责单步内容生成，没有决策权。
• Agent：执行路径由大模型自主推理决定，开发者只提供能力与规则，不指定具体执行步骤。

打个比方：Chain 像工厂的固定流水线，每道工序的顺序和内容完全不变；Agent 像一个完整的执行员工，拿到任务后自主规划执行方案，需要查资料就查资料，需要算数据就算数据，直到交付最终结果。

5.1.2 ReAct 执行范式

主流 Agent 普遍遵循 ReAct(Reasoning + Acting，推理+行动) 执行范式，和人类解决问题的逻辑完全一致，整个过程是一个可循环的执行链路：

1. Thought(思考)：分析当前问题与已有信息，判断下一步动作——是直接给出答案，还是调用某个工具获取更多信息。
2. Action(行动)：选定对应工具，生成调用参数，执行具体操作。
3. Observation(观察)：获取工具的执行结果，作为新的上下文反馈给大模型。
4. 循环/终止：基于新获取的信息继续推理迭代，直到判断任务已完成，输出最终答案。

5.1.3 三大核心组件

一个完整可运行的 Agent 系统由三个核心部分构成，各司其职：

• 大语言模型(LLM)：决策大脑。负责推理规划、工具选择、结果总结，决定每一步的执行动作。
• 工具集(Tools)：执行手脚。扩展大模型的能力边界，提供搜索、计算、业务API、文件操作等外部能力。
• 执行器(Agent Executor)：调度中枢。管控整个执行循环，处理参数解析、错误重试、终止条件、安全限制等工程逻辑。

5.2 Tools

工具是 Agent 连接真实世界与业务系统的桥梁，也定义了 Agent 的能力边界。大模型本身不具备的能力——实时信息查询、精确数值计算、内部系统操作，都可以通过工具来补充。

5.2.1 Tools 介绍

Tool 本质是标准化封装的可调用函数，有明确的名称、功能描述和参数规范。大模型通过阅读工具描述判断调用时机，按照参数规范生成调用指令。

一个合格的 Tool 包含三大核心要素：

• name：工具的唯一名称，简洁清晰，便于模型识别。
• description：工具的功能说明，是大模型判断是否调用的核心依据，描述越精准边界越清晰，调用准确率越高。
• args_schema：输入参数的格式定义，通常用类型声明或 Pydantic Schema 约束，保证模型传参的准确性。

LangChain 内置了上百种开箱即用的工具，覆盖绝大多数通用场景：

工具类型	代表工具	解决的核心问题
搜索引擎	Tavily Search、SerpAPI	获取实时资讯、公开资料，突破模型知识截止日期
数学计算	Calculator、LLMMathChain	解决大模型算术不准、复杂公式计算错误的问题
代码执行	PythonREPL	运行 Python 代码，处理数据、绘图、复杂逻辑运算
数据查询	SQLDatabaseToolkit	用自然语言操作关系型数据库
文件处理	FileReadTool、WriteTool	读写本地文件、批量处理文档

5.2.2 创建工具

企业级落地场景中，绝大多数工具需要对接内部业务系统，LangChain 提供了两种自定义工具的实现方式，适配不同复杂度的需求。

方式一：@tool 装饰器(最常用)

用装饰器快速把普通 Python 函数封装成标准 Tool，代码简洁、上手快，适合绝大多数业务场景：

from langchain_core.tools import tool

@tool
def get_order_info(order_id: str) -> str:
    """
    根据订单号查询订单详情，包含商品名称、原价、下单时间。
    参数 order_id 为数字格式的订单编号。
    """
    ## 此处替换为真实业务接口调用
    mock_db = {
        "1001": "商品：机械键盘，原价：299元，下单时间：2026-05-20",
        "1002": "商品：无线耳机，原价：599元，下单时间：2026-06-01"
    }
    return mock_db.get(order_id, "未找到对应订单")

注意：函数的文档字符串(docstring)就是工具的 description，必须写清楚功能、参数含义、适用场景，直接决定模型的调用准确率。

方式二：继承 BaseTool(复杂场景)

适合需要更精细控制、有复杂初始化逻辑的工具，比如需要持有数据库连接、初始化客户端的工具：

from langchain_core.tools import BaseTool
from pydantic import Field

class OrderQueryTool(BaseTool):
    name: str = "get_order_info"
    description: str = "根据订单号查询订单详情，包含商品名称和原价。"
    db_config: dict = Field(default_factory=dict)

    def _run(self, order_id: str) -> str:
        ## 具体业务执行逻辑
        return f"订单{order_id}详情：..."

5.2.3 绑定工具

早期 Agent 通过提示词引导模型输出工具调用指令，需要手动解析文本，不稳定且容易出错。当前主流方案是工具绑定(Tool Binding)：将工具定义直接传给大模型，利用模型原生的 Tool Calling 能力，直接输出结构化的调用指令，准确率和稳定性大幅提升。

LangChain 中通过模型的 bind_tools() 方法完成工具绑定：

from langchain_ollama import ChatOllama

llm = ChatOllama(model="qwen2:7b")

## 将工具列表绑定到模型
tools = [get_order_info]
llm_with_tools = llm.bind_tools(tools)

## 调用模型，自动返回结构化的工具调用指令
result = llm_with_tools.invoke("订单1002的商品是什么？")
print(result.tool_calls)

绑定后，模型会自动识别是否需要调用工具，并输出标准的 tool_calls 结构化数据，包含工具名称和调用参数，无需手动解析。

5.3 构建 Agent

基于 Tool Calling 的 Agent 是当前生产环境的主流方案，稳定性和准确率都优于传统 ReAct 提示词方案。下面我们用完整步骤，从零搭建一个可运行的业务 Agent。

完整构建步骤

第一步：定义工具集

from langchain_core.tools import tool

@tool
def get_order_info(order_id: str) -> str:
    """根据订单号查询订单详情，包含商品名称和原价。参数order_id为订单号。"""
    mock_db = {
        "1001": "商品：机械键盘，原价：299元",
        "1002": "商品：无线耳机，原价：599元",
        "1003": "商品：显示器，原价：1299元"
    }
    return mock_db.get(order_id, "未找到对应订单")

@tool
def calculate_discount(price: float, discount: float) -> str:
    """计算折扣后价格。参数price为原价，discount为折扣率(如0.8表示8折)。"""
    return f"折扣后价格：{price * discount:.2f}元"

tools = [get_order_info, calculate_discount]

第二步：定义提示词与初始化模型

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_ollama import ChatOllama

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是智能客服助手，使用提供的工具回答用户问题。如果工具无法解决，请如实告知用户。"),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

llm = ChatOllama(model="qwen2:7b", temperature=0)

其中 agent_scratchpad 是固定占位符，用于存放每一轮的思考过程、工具调用指令、执行结果，是 Agent 执行循环的核心上下文。

第三步：创建 Agent 与执行器

from langchain.agents import AgentExecutor, create_tool_calling_agent

## 创建 Tool Calling Agent
agent = create_tool_calling_agent(llm, tools, prompt)

## 包装为执行器，管控整个执行循环
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,               ## 打印完整执行链路，调试必备
    max_iterations=5,           ## 最大迭代次数，防止无限循环
    handle_parsing_errors=True, ## 自动处理解析错误，提升稳定性
)

第四步：运行测试

result = agent_executor.invoke({
    "input": "订单1002打75折后是多少钱？"
})

print("最终答案：", result["output"])

开启 verbose 后可以看到完整的自主决策链路：模型先调用 get_order_info 获取订单原价，再调用 calculate_discount 计算折扣价格，最后整合结果输出答案，全程无需人工干预。

执行器核心调优参数

AgentExecutor 有几个高频配置参数，是生产环境调优的重点：

• max_iterations：限制最大执行步数，避免 Agent 陷入死循环，通常设为 3~10。
• max_execution_time：单任务最大执行时间，超时强制终止，防止异常任务占用资源。
• handle_parsing_errors：解析失败时自动让模型重试，而非直接报错中断。
• return_intermediate_steps：返回每一步的中间结果，用于调试排查和日志审计。

5.4 LangSmith

Agent 的执行过程是典型的黑盒：出了问题很难定位是提示词不合理、模型能力不足，还是工具调用出错。LangSmith 是 LangChain 官方推出的 LLM 应用全生命周期开发平台，专门解决大模型应用的调试、测试、监控、评估难题。

核心能力

LangSmith 覆盖了 LLM 应用从开发调试到生产运维的全流程：

1. 全链路追踪：完整记录每一次调用的所有细节——提示词、模型输出、工具调用参数与结果、每步耗时、Token 消耗，可视化呈现 Agent 的完整 ReAct 循环，调试问题一目了然。
2. 提示词管理：提示词在线编辑、版本管理、灰度发布，产品和运营可以直接优化提示词，无需修改代码。
3. 数据集与评估：批量构造测试集，自动运行评估指标，量化对比不同提示词、不同模型、不同版本的效果，避免凭感觉优化。
4. 生产监控：错误率、平均耗时、Token 成本、工具调用成功率等核心指标大盘，实时监控线上服务健康状态。

快速接入

LangSmith 接入几乎零代码侵入，只需要配置几个环境变量即可自动上报数据：

## .env 文件配置
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY="你的LangSmith API Key"
LANGCHAIN_PROJECT="你的项目名称"

配置完成后，所有 Chain 和 Agent 的执行数据都会自动上报到 LangSmith 平台，在控制台可以直接查看完整的调用链路详情。

5.5 记忆

单轮 Agent 只能处理独立问题，而真实业务场景中的对话往往是多轮的——用户会用指代、省略等自然表达，需要 Agent 记住历史上下文才能正确理解意图。

实现原理

Agent 的记忆能力基于 LangChain 统一的 RunnableWithMessageHistory 体系实现：

1. 在提示词中预留 chat_history 占位符，用于注入历史对话消息。
2. 通过会话 ID(session_id)区分不同用户/对话，自动加载对应会话的历史记录。
3. 每轮对话结束后，自动将本轮的用户输入和 AI 回复保存到历史存储中。

带记忆的 Agent 实现

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
import os

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    history_dir = "./agent_history"
    os.makedirs(history_dir, exist_ok=True)
    return FileChatMessageHistory(f"{history_dir}/{session_id}.json")

## 包装带记忆的 Agent 执行器
agent_with_history = RunnableWithMessageHistory(
    agent_executor,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

## 多轮对话测试
## 第一轮：明确指定订单
resp1 = agent_with_history.invoke(
    {"input": "帮我查一下订单1003的信息"},
    config={"configurable": {"session_id": "user_001"}}
)

## 第二轮：指代提问，无需重复订单号
resp2 = agent_with_history.invoke(
    {"input": "打8折后是多少钱？"},
    config={"configurable": {"session_id": "user_001"}}
)

第二轮提问中的指代会被模型通过历史对话正确识别，实现自然连贯的多轮交互。

生产环境注意：长对话会持续占用模型上下文窗口，建议配合摘要记忆或滑动窗口记忆，平衡记忆效果和 Token 成本。

5.6 MCP

MCP(Model Context Protocol，模型上下文协议)是由 Anthropic 推出的开放标准协议，目标是为大模型提供一套标准化的外部能力接入规范，让大模型应用可以统一、安全地连接各类数据源和工具，避免重复对接开发。

5.6.1 MCP 介绍

在 MCP 出现之前，每个大模型应用对接外部系统都要单独开发适配器：对接订单系统写一套、对接知识库写一套、对接文件系统写一套，换个模型或框架还要重新适配，重复工作量极大。

MCP 的定位就是大模型界的「USB 通用协议」：

• 只要能力提供方(外部业务系统)实现了 MCP 服务器标准，就能被所有支持 MCP 的应用侧(大模型应用)识别和使用。
• 一次开发，跨应用、跨模型、跨框架复用，大幅降低系统集成成本。

5.6.2 MCP 架构

MCP 采用经典的客户端-服务器分层架构，两端完全解耦：

1. MCP 客户端：集成在大模型应用侧(如 LangChain、Claude Desktop)，负责发现服务能力、发起调用请求、处理返回结果。
2. MCP 服务器：部署在能力提供侧，对外暴露具体的工具、资源、提示词，每个业务系统对应一个 MCP 服务。
3. 传输层：支持 stdio(本地进程)、HTTP、WebSocket 等多种传输方式，本地工具和远程服务都能统一接入。

这种架构的核心优势是语言无关、框架无关：用 Go 编写的 MCP 服务，Python 技术栈的 LangChain 可以直接使用；切换到其他支持 MCP 的框架，也能无缝复用现有能力。

5.6.3 MCP 层级

MCP 协议按能力分为三层，逐层扩展，按需实现：

1. 基础协议层：核心底座。定义消息格式、握手流程、错误码、序列化规范，所有 MCP 实现必须兼容。
2. 核心能力层：三大标准能力，覆盖绝大多数业务场景
   • 工具(Tools)：可执行的操作，和 LangChain 的 Tool 概念一致，比如查询订单、发送通知。
   • 资源(Resources)：可读取的上下文数据，比如文档内容、数据库记录、系统状态，模型可以按需读取补充上下文。
   • 提示词(Prompts)：可复用的标准化提示模板，比如代码审查、数据分析提示，跨应用统一复用。
3. 扩展能力层：采样控制、日志上报、权限校验、进度通知等进阶能力，可根据业务需求选择性实现。

5.6.4 MCP 工作流程

一次完整的 MCP 调用分为五个标准步骤：

1. 连接建立：客户端通过指定传输方式连接 MCP 服务器，完成协议版本协商和能力握手。
2. 能力发现：客户端查询服务器支持的全部工具、资源、提示词列表，同步给大模型。
3. 意图决策：大模型分析用户问题，自主决定调用哪个工具、读取哪个资源。
4. 能力调用：客户端向 MCP 服务器发送标准化调用请求，服务器执行业务逻辑后返回结果。
5. 结果生成：大模型基于返回的结果，整合生成最终回答反馈给用户。

5.6.5 MCP SDK

官方提供了 Python、TypeScript 等多语言 SDK，封装了全部协议底层细节，开发者只需要关注业务逻辑。以 Python SDK 为例，几行代码就能实现一个 MCP 服务器：

## pip install mcp
from mcp.server.fastmcp import FastMCP

## 创建 MCP 服务器实例
mcp = FastMCP("订单服务")

## 定义一个工具
@mcp.tool()
def get_order_info(order_id: str) -> str:
    """根据订单号查询订单详情"""
    mock_db = {"1001": "机械键盘 299元", "1002": "无线耳机 599元"}
    return mock_db.get(order_id, "未找到订单")

## 启动服务器(stdio 模式)
if __name__ == "__main__":
    mcp.run()

开发者只需要用装饰器标记工具，SDK 会自动处理协议握手、参数解析、消息序列化等全部底层工作。

5.6.6 LangChain 使用 MCP

LangChain 原生支持 MCP 协议，可以一键将 MCP 服务器的能力转换为 LangChain 标准 Tool，直接接入 Agent 使用，无需重复封装。

## pip install langchain-mcp
from langchain_mcp import MCPToolkit

## 连接本地 MCP 服务器(stdio 方式)
toolkit = MCPToolkit.from_stdio(
    command="python",
    args=["order_mcp_server.py"]
)

## 获取全部工具
tools = toolkit.get_tools()

## 直接传入 Agent 使用，和普通工具完全一致
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

对于企业来说，内部业务系统只需要维护一套 MCP 服务，所有基于 LangChain 的 AI 应用都能直接复用，彻底避免重复造轮子。

5.7 监督者模式多 Agent 架构

单 Agent 能力虽然灵活，但在复杂企业场景中会遇到瓶颈：一个 Agent 既要懂代码、又要懂财务、还要懂客服，工具越堆越多，模型选择成本上升，准确率下降。

工业界的成熟解决方案是多 Agent 架构，其中监督者模式(Supervisor Pattern) 是应用最广泛的一种。

架构组成

监督者模式也叫主管-工人模式，核心分为两类角色：

1. 监督者 Agent(Supervisor)：全局调度大脑。不执行具体业务，只负责接收用户问题、拆解任务、分发给合适的子 Agent、汇总校验结果、处理异常。
2. 专业子 Agent(Worker Agents)：领域执行单元。每个子 Agent 只专注一个领域，配备专属的提示词和工具，深耕单一任务。常见的子 Agent 如：
   • 文档检索 Agent：负责知识库问答，配备 RAG 检索工具
   • 代码 Agent：负责代码编写、调试，配备代码执行、文件读写工具
   • 数据分析 Agent：负责数据查询、统计、可视化，配备 SQL、Python 工具
   • 客服 Agent：负责订单、售后问题，配备业务系统工具

工作流程

1. 用户提交问题，监督者接收并分析问题所属领域。
2. 监督者拆解任务，将子任务分发给对应领域的子 Agent。
3. 子 Agent 调用自身工具完成子任务，将结果返回给监督者。
4. 监督者校验结果，如有缺失或错误，发回子 Agent 修正；如需要其他领域配合，再分发给其他子 Agent。
5. 所有子任务完成后，监督者整合全部结果，生成最终回答返回给用户。

优势与适用场景

核心优势

• 专业度更高：每个子 Agent 只聚焦一个领域，提示词和工具更精准，效果远胜全能型单 Agent。
• 扩展性更强：新增业务领域只需要新增一个子 Agent，不影响现有系统，团队可以并行开发。
• 稳定性更好：单个子 Agent 出错不影响全局，监督者可以重试、降级或切换备用方案。
• 执行效率更高：无依赖的子任务可以并行执行，大幅缩短复杂任务的总耗时。

适用场景

• 大型企业智能助手：覆盖多部门、多业务线的统一智能入口
• 自动化办公流水线：跨系统、多步骤的复杂业务流程自动化
• 复杂内容生产：需要调研、撰写、校对、排版多环节配合的内容任务
• 研发效能助手：集代码生成、文档查询、故障排查于一体的研发助手

---

本章我们从基础概念到工程落地，完整讲解了 Agent 的核心体系：从工具的定义与绑定，到单 Agent 的构建，再到记忆能力、调试监控、MCP 标准协议，最后延伸到企业级多 Agent 架构。

从 Model I/O 到 Chains，从 Retrieval 到 Agents，LangChain 的四大核心模块共同构成了大模型应用开发的完整技术栈。掌握这些能力，就具备了从零搭建企业级 AI 应用的全部基础。