LangChain

1. 什么是 LangChain

LangChain 是一个用于开发由大型语言模型（LLM）驱动的应用程序的开源框架。它的定位是"LLM 应用的中间件 / 胶水层"——抽象了 LLM 调用、链式组合、工具集成、记忆管理、检索增强等通用能力，让开发者不必从零拼装，把 LLM、工具、数据和业务逻辑像搭积木一样组合成应用。

我们希望大模型应用不仅仅是聊天，更能从已有的数据库或文件中提取信息并执行具体操作。LangChain 正是为此而生：它允许开发者将通义千问、DeepSeek、OpenAI 等大语言模型与外部系统和数据源结合，完成更复杂的操作。

也可以借助 Dify、RAGFlow 之类的低代码平台，但自由度受限，对复杂业务适配欠缺；也可以完全自研裸调 SDK，但费时费力。LangChain 提供了现成组件，同时保留了足够的自定义自由度。

创始人：Harrison Chase（2022 年开源）
语言：Python / JavaScript（TypeScript）
官网：https://python.langchain.com/docs/introduction/
GitHub：https://github.com/langchain-ai/langchain
当前版本：LangChain 1.0+（2025），底层基于 langchain-core 和 langgraph 重构

地位类比：

LangChain = Java 生态中 Spring Boot（脚手架，快速构建应用）+ Apache Camel（集成总线，支持 200+ 工具和服务）的结合体
LangChain = C++ 生态中的 ROS + gRPC

一句话：它是连接 LLM 能力与现实需求的"数字桥梁"，把大模型的原始能力转化为可落地的行业解决方案。

2. LangChain V1.0 重构

LangChain v1.0 是一个专注于构建 Agent 的、可直接用于生产环境的基础框架，标志着 AI 智能体开发正式进入工程化阶段。最核心的变化是基于 LangGraph 重构了底层架构，并引入了全新的 Agent 构建 API。

2.1 全面拥抱 LangGraph

LangChain v1.0 不再是独立的链式调用工具，而是构建在 LangGraph 之上的"Agent 快速通道"。

统一运行时：所有 v1.0 的 Agent 都在 LangGraph 运行时上执行，直接获得持久化、流式输出、以及"时光倒流"（Time Travel）调试能力。
定位差异：LangGraph 负责底层的图编排和精细控制；LangChain v1.0 提供开箱即用的高层抽象，几行代码就能跑起一个生产级 Agent。

2.2 全新 API：`create_agent`

这是 v1.0 中最重要的 API，取代了过去的 AgentExecutor 和各种 Chain。

标准化入口：比底层 LangGraph 更简单，比旧版 Agent 更灵活。
替代旧版：直接替代 langgraph.prebuilt.create_react_agent 以及旧版 ReAct Agent 构建方式。

2.3 引入中间件机制

可以像在 Web 开发中一样，在 Agent 的生命周期中插入自定义逻辑（详见第 10 节）：

拦截与控制：在模型调用前后、工具调用前后插入代码。
预置中间件：PII Redaction（脱敏）、Summarization（自动摘要）、Human-in-the-loop（人工审批）等。

2.4 标准化内容块

解决不同大模型（OpenAI、Anthropic、Google 等）之间消息格式不统一的痛点。

引入 content_blocks 属性，跨提供商统一访问文本、推理过程、引用和工具调用。
将内容与原始字符串解耦，使多模态和复杂输出的处理更加类型安全。

2.5 结构化输出改进

主循环集成：结构化输出直接集成在 Agent 主循环中生成，不再需要额外的 LLM 调用。
降本增效：减少 Token 消耗和延迟，模型可选择直接返回结果或调用工具。

2.6 命名空间清理与迁移

精简核心：langchain 包现在只包含构建 Agent 所需的核心组件（Agent、Model、Tools）。
遗留代码迁移：旧版 Chains、AgentExecutor 和遗留工具都移到了新包 langchain-classic 中。
向后兼容：安装 langchain-classic 即可运行旧项目，同时逐步迁移到新架构。

2.7 其他变动

Python 版本：放弃 Python 3.9，建议 3.10+。
LangGraph v1.0：同步发布正式版，承诺核心 API 稳定性。

如何升级：新项目直接用 create_agent 和 v1.0 API；旧项目先装 langchain-classic 保持运行，再逐步用 Middleware 和 create_agent 替换老的 Chain 和 Executor。

3. LangChain 体系

LangChain 由以下开源库组成：

库	作用
langchain	主要入口，构建 LLM 应用所需的核心实现
langchain-core	生态系统的核心接口和抽象
langchain-community	第三方集成（向量库、文档加载器等）
langchain-classic	旧版 langchain 实现与组件
合作伙伴库	如 `langchain-openai`、`langchain-anthropic`，仅依赖 `langchain-core`
LangGraph	将步骤建模为图的边和节点，构建有状态多参与者应用
DeepAgents	能规划、使用子代理、利用文件系统处理复杂任务的代理
LangServe	将 LangChain 链部署为 REST API
LangSmith	调试、测试、评估和监控 LLM 应用的开发者平台

LangChain 简化了 LLM 应用生命周期的各阶段：

开发：用开源构建块和组件搭建应用，借助第三方集成和模板快速启动。
生产化：用 LangSmith 检查、监控和评估链，持续优化。
部署：用 LangServe 将任何链转化为 API。

4. 核心组件（七大部分）

4.1 代理 (Agents)

v1.0 中 Agent 是应用的核心构建块。

统一接口 create_agent，取代繁杂的 AgentExecutor。
核心逻辑：Agent 本质是一个"循环"——接收输入，决定是否调用工具，处理工具结果，重复直到完成任务。
底层驱动：默认基于 LangGraph 运行，原生支持状态管理、分支逻辑和持久化。

4.2 模型 (Models)

模型是 Agent 的"大脑"。

标准内容块：无论底层用哪个模型，输出（文本、工具调用、推理过程）都统一为标准格式，方便无缝切换。
Chat Models：v1.0 推荐全面使用 Chat Model 接口（而非纯文本 LLM），以支持多模态输入和结构化交互。

4.3 消息 (Messages)

消息是模型交互的基础数据单元，也是 Agent 的"短期记忆"（由消息列表组成）。

HumanMessage：用户输入
AIMessage：模型回复（含思维过程 thought_process 和工具调用 tool_calls）
SystemMessage：设定 Agent 的行为准则和角色
ToolMessage：工具执行后的返回结果

4.4 工具 (Tools)

工具是 Agent 连接外部世界（API、数据库、搜索引擎）的桥梁。

定义方式：用 @tool 装饰器把 Python 函数转换为工具。
错误处理：v1.0 优化了工具调用错误处理，参数填错时系统会自动捕获并反馈给 Agent，让其自我修正重试。

4.5 短期记忆 (Short-term Memory)

旧版 vs 新版：不再使用 ConversationBufferMemory。
线程级持久化：通过 Checkpointer 机制实现，自动保存每一步状态；传入 thread_id 即可加载之前的上下文继续任务。

4.6 流 (Streaming)

流式 Token：最基础的"打字机效果"。
流式进度：实时展示 Agent 执行状态。
流式自定义更新：手动发送业务信号（如"已下载 10/100 条数据"）。

4.7 结构化输出 (Structured Output)

机制：create_agent 通过 response_format 参数直接配置（如传入 Pydantic 类）。
结果：结构化数据会被捕获、验证，并以代理状态的 structured_response 键返回。

5. Model I/O

可以把对模型的使用过程拆解成三块：输入提示 (Format) → 调用模型 (Predict) → 输出解析 (Parse)，这个整体在 LangChain 中统称为 Model I/O。

组件	作用
LLM / ChatModel	封装模型调用（OpenAI、Anthropic、本地模型等）
Prompt Template	结构化 prompt，支持变量插值和 few-shot
Output Parser	把模型输出解析成 JSON、Pydantic 对象等结构化数据

5.1 模型 (Model)

LangChain 支持三大类模型：

大语言模型（LLM）：也叫文本补全模型，接收字符串、返回补全字符串，只能一问一答。OpenAI 的文本补全 API 已停更，国内厂商普遍不支持，不予考虑。
聊天模型（Chat Model）：主流选择。输入是聊天消息列表，返回聊天消息，支持多轮对话。包装器为 ChatOpenAI；兼容 OpenAI 的模型都可用 ChatOpenAI，不兼容的用 init_chat_model 创建。
文本嵌入模型（Embedding Model）：将文本转为词向量（详见第 8 节）。

# 通过 LangChain 调用大模型，实现交互
import os
from langchain_openai import ChatOpenAI

MODEL_API_KEY = os.getenv("DASHSCOPE_API_KEY")

client = ChatOpenAI(
    api_key=MODEL_API_KEY,
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-max-latest"
)

msg = [
    ('system', '请将以下的内容翻译成英文'),
    ('human', '你好，你今天过得好吗？'),
]

result = client.invoke(msg)
print(result)

直接调用有两个问题：① 提示词固定，无法灵活变动；② 原始答复复杂，含大量额外信息。分别用提示词模板和输出解析器解决。

5.2 提示模板 (Prompt Template)

PromptTemplate 通过接收原始用户输入，返回一个准备好传递给模型的提示词。它是模板化字符串，可插入变量生成不同提示。

特点：清晰易懂、增强可重用性、简化维护、智能处理变量、参数化生成。

类型：

PromptTemplate：常用的 String 提示模板
ChatPromptTemplate：常用的 Chat 提示模板，组合各种角色的消息模板
FewShotPromptTemplate：通过示例教模型如何回答
提示模板部分格式化：先给某些参数赋值，其余后期赋值
自定义模板

from langchain_core.prompts import PromptTemplate

# 方法一
template = "您是一位专业的程序员。\n对于信息 {text} 进行简短描述"
prompt = PromptTemplate.from_template(template)
print(prompt.format(text="langchain"))

# 方法二
prompt2 = PromptTemplate(
    input_variables=["text"],
    template="您是一位专业的程序员。\n对于信息 {text} 进行简短描述"
)
print(prompt2.format(text="langchain 框架"))

ChatPromptTemplate 聊天提示模板

ChatPromptTemplateChatPromptTemplateChatPromptTemplate 创建聊天消息列表，模板带有对应角色。常用的有 AIMessagePromptTemplate、SystemMessagePromptTemplate、HumanMessagePromptTemplate。 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate from langchain_core.output_parsers import StrOutputParser from models import DEEPSEEK_URL, DEEPSEEK_V4_PRO import os api_key = os.getenv("MY_DEEPSEEK_API_KEY") base_url = DEEPSEEK_URL model = DEEPSEEK_V4_PRO llm = ChatOpenAI(api_key=api_key, base_u

FewShotPromptTemplate 少样本提示词模板

Few-shot 提示技术 — 通过给模型提供几个示例（examples），让模型学习输出格式和风格，再回答真正的问题。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import (
    FewShotPromptTemplate,
    PromptTemplate,
)

from models import DEEPSEEK_URL, DEEPSEEK_V4_PRO
import os

api_key = os.getenv("MY_DEEPSEEK_API_KEY")
base_url = DEEPSEEK_URL
model = DEEPSEEK_V4_PRO

llm = ChatOpenAI(api_key=api_key, base_url=base_url, model=model)

examples = [
    {"sinput": "2+2", "soutput": "4", "sdescription": "加法运算"},
    {"sinput": "5-2", "soutput": "3", "sdescription": "减法运算"},
]

template = "算式：{sinput}, 值：{soutput}, 类型：{sdescription}"

# example_prompt = PromptTemplate(template=template)
example_prompt = PromptTemplate.from_template(template)

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="你是一个数学专家, 能够准确说出算式的类型，",
    suffix="现在给你算式: {input} ， 值: {output} ，告诉我类型：",
    input_variables=["input", "output"],
)


result = llm.invoke(prompt.format(input="2*5", output="10"))
print(result.content)  # 使用: 乘法运算

5.3 输出解析器 (Output Parser)

输出解析器 (Output Parser)输出解析器 (Output Parser)概述 LLM 的原始输出永远是一段字符串(AIMessage)。输出解析器(Output Parser)负责把这段字符串解析为程序可直接使用的结构化数据(字符串、列表、字典、日期、对象等)。它通常承担两个职责: 1. 格式约束:通过 get_format_instructions() 生成一段格式说明,拼进 Prompt 告诉模型"按什么格式输出"。 2. 解析输出:通过 parse() / invoke() 把模型返回的文本转成目标数据类型。在 LCEL 链式写法中,解析器放在最后,chain.invoke() 的返回值就直接是解析后的结果,无需手动调用 parse(): chain = prompt | llm | parser 常见类型:StrOutputParser、CommaSeparatedListOutputParser、JsonOutputParser、DatetimeOutputParser、XMLOutputParser、PydanticOutputParser 等。两类解析器的区别 - 只解析、不约束格式:如 StrOutputParser、J

6. 链 (Chain) 与 LCEL

链把模型输入输出整合在一个流程中操作：利用提示模板格式化输入，传给模型，再返回输出。LangChain 的名字正源自其核心设计——用链（Chain）将各组件链接起来构建复杂应用。

v1.0 已把各种内置 Chain 移到 langchain-classic，但 LCEL 作为一种核心思想仍有必要学习。在 v0.1 后，链分为遗留链和 LCEL 链，我们以 LCEL 链为主。

6.1 LCEL（LangChain Expression Language）

用声明式方法链接组件。所有可链起来的组件（LLM、输出解析器、检索器、提示词模板等）都支持三个方法：

stream：流式返回响应的块
invoke：接受输入返回输出
batch：接受批量输入返回输出列表

最简单常见的写法是用管道操作符 |（或 RunnableSequence）：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate
from models import get_lc_model_client

client = get_lc_model_client()

chat_template = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template("请将以下的内容翻译成 {language}"),
        ('human', '{text}')
    ]
)
parser = StrOutputParser()

# prompt | llm | parser 等价于一个可调用的 Runnable
chain = chat_template | client | parser
print(chain.invoke({'language': '意大利文', 'text': '朋友啊再见！'}))

6.2 LCEL 高级特性与组件

RunnableLambda —— 把自定义函数加入链

from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, chain
from models import get_lc_model_client

client = get_lc_model_client()
chat_template = ChatPromptTemplate.from_template("{a} + {b} 是多少？")

def length_function(text):
    return len(text)

@chain
def multiple_length_function(_dict):
    return len(_dict["text1"]) * len(_dict["text2"])

chain1 = chat_template | client

full_chain = (
    {
        "a": itemgetter("foo") | RunnableLambda(length_function),
        "b": {"text1": itemgetter("foo"), "text2": itemgetter("bar")} | multiple_length_function,
    }
    | chain1
)
print(full_chain.invoke({"foo": "abc", "bar": "abcd"}))

RunnableParallel —— 并行执行多个任务

from langchain_core.runnables import RunnableLambda, RunnableParallel

runnable_1 = RunnableLambda(lambda x: x + 1)
runnable_2 = RunnableLambda(lambda x: x * 2)
runnable_3 = RunnableLambda(lambda x: x * 3)

# 顺序执行：1+1=2 -> 2*2=4 -> 4*3=12
print((runnable_1 | runnable_2 | runnable_3).invoke(1))  # 12

# 并行执行
chain = runnable_1 | RunnableParallel(mul_two=runnable_2, mul_three=runnable_3)
print(chain.invoke(1))  # {'mul_two': 4, 'mul_three': 6}

RunnablePassthrough —— 传递 / 增强数据

常用在链的第一个位置接收用户输入；也可通过 assign 对数据增强后再往后传。

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

# 原样传递
runnable = RunnableParallel(
    passed=RunnablePassthrough(),
    modified=lambda x: x["num"] + 1,
)
print(runnable.invoke({"num": 1}))  # {'passed': {'num': 1}, 'modified': 2}

# 增强后传递
runnable = RunnableParallel(
    passed=RunnablePassthrough().assign(query=lambda x: x["num"] + 2),
    modified=lambda x: x["num"] + 1,
)
print(runnable.invoke({"num": 1}))  # {'passed': {'num': 1, 'query': 3}, 'modified': 2}

6.3 跟踪和调试

方法一：LangSmith——跟踪和评估 LLM 应用的平台，个人开发一定额度内免费。
方法二：开启调试——langchain.debug = True，打印链执行的详细过程。

import langchain
langchain.debug = True
# ... 构建并调用 chain，控制台会打印每步细节

6.4 查看链及其组件

print(chain.input_schema.model_json_schema())
print(chain.output_schema.model_json_schema())
print(chain.get_prompts())

# 打印链的结构图，需 pip install grandalf
chain.get_graph().print_ascii()

6.5 发布应用程序（LangServe）

# 服务端：把链部署为 WEB 服务
from fastapi import FastAPI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate
from langserve import add_routes
from models import get_lc_model_client

client = get_lc_model_client()
prompt_template = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template("请将以下的内容翻译成 {language}"),
        ('human', '{text}')
    ]
)
chain = prompt_template | client | StrOutputParser()

app = FastAPI(title="基于 LangChain 的服务", version="V1.5", description="翻译服务")
add_routes(app, chain, path="/tslServer")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="localhost", port=8000)

# 客户端调用
from langserve import RemoteRunnable

client = RemoteRunnable("http://localhost:8000/tslServer")
print(client.invoke({'language': '意大利文', 'text': '为了部落！'}))

7. 增强提示词模板与输出解析器

7.1 少量样本示例的提示模板（FewShot）

通过示例教模型如何回答，使用 FewShotPromptTemplate 或 FewShotChatMessagePromptTemplate。

from langchain.prompts import PromptTemplate
from langchain.prompts.few_shot import FewShotPromptTemplate
from models import get_lc_model_client

client = get_lc_model_client()

examples = [
    {"input": "2+2", "output": "4", "description": "加法运算"},
    {"input": "5-2", "output": "3", "description": "减法运算"},
]

prompt_sample = PromptTemplate.from_template("算式：{input} 值：{output} 类型：{description}")

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=prompt_sample,
    prefix="你是一个数学专家, 能够准确说出算式的类型，",
    suffix="现在给你算式: {input}，值: {output}，告诉我类型：",
    input_variables=["input", "output"]
)

print(prompt.format(input="2*5", output="10"))
print(client.invoke(prompt.format(input="2*5", output="10")).content)

7.2 提示模板部分格式化

适用于需要先给某些参数赋值，其余后期赋值。

import datetime
from langchain.prompts import PromptTemplate
from models import get_lc_model_client

client = get_lc_model_client()

def get_datetime():
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

prompt = PromptTemplate(
    template="讲一个关于 {date} 的 {story_type}",
    input_variables=["date", "story_type"]
)

half_prompt = prompt.partial(date=get_datetime())  # 先填 date
print(client.invoke(half_prompt.format(story_type="笑话")).content)
print(client.invoke(half_prompt.format(story_type="悲伤故事")).content)

7.3 更多输出解析器示例

CSV 解析器 CommaSeparatedListOutputParser：逗号分隔，以列表返回
日期时间解析器 DatetimeOutputParser：解析为日期时间
JSON 解析器 JsonOutputParser：确保符合特定 JSON 格式
XML 解析器 XMLOutputParser：以 XML 格式返回

# JSON 解析器
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from models import get_lc_model_client

client = get_lc_model_client()
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的程序员"),
    ("user", "{input}")
])
chain = prompt | client | JsonOutputParser()
print(chain.invoke({"input": "langchain 是什么? 问题用 question 回答用 ans 返回一个 JSON 格式"}))

# 日期时间解析器
from langchain.output_parsers import DatetimeOutputParser
from langchain.prompts import PromptTemplate
from models import get_lc_model_client

client = get_lc_model_client()
output_parser = DatetimeOutputParser()
prompt = PromptTemplate.from_template(
    "回答用户的问题：{question}\n{format_instructions}",
    partial_variables={"format_instructions": output_parser.get_format_instructions()},
)
chain = prompt | client | output_parser
print(chain.invoke({"question": "新中国是什么时候成立的？"}))  # 1949-10-01

8. Memory 与 RAG

8.1 对话历史管理（Memory）

让 LLM 在多轮对话中记住上文。经典策略：

策略	做法
Buffer	全量保存历史，Token 多了就截断
Summary	用 LLM 对历史做摘要，只保留摘要
Window	只保留最近 K 轮对话
向量记忆	历史存入向量库，检索式召回相关片段

在 LangGraph 中，Memory 变成 Checkpointer，更灵活可控。

大模型是无状态的，记不住每次对话内容。新版本 LangChain 提供消息历史组件 ChatMessageHistory 和自动会话历史管理组件 RunnableWithMessageHistory。

# 自动会话历史管理 RunnableWithMessageHistory
from langchain.messages import HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from models import get_lc_model_client

client = get_lc_model_client()

prompt_template = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template("你是一个聊天助手，用 {language} 回答所有的问题"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)
chain = prompt_template | client | StrOutputParser()

store = {}  # 保存所有用户的聊天历史

def get_session(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

chatbot_with_his = RunnableWithMessageHistory(
    chain, get_session,
    input_messages_key="input",
    history_messages_key="history"
)

config_chn = {'configurable': {'session_id': 'yunfang_chinese'}}
print(chatbot_with_his.invoke(
    {"input": [HumanMessage(content="你好，我是云帆。")], "language": "中文"}, config=config_chn))
print(chatbot_with_his.invoke(
    {"input": [HumanMessage(content="请问我的名字是什么？")], "language": "中文"}, config=config_chn))

8.2 RAG（检索增强生成）

让 LLM 能"读"外部文档再回答。在许多 LLM 应用中，用户特定数据不在大模型里，而在外部系统或文档中。LangChain 的数据连接组件包括：文档加载器、文档切分、文本嵌入、向量存储、检索器。

核心流程：

Document Loader：加载 PDF / 网页 / 数据库等
Text Splitter：把长文本切成适当大小的 chunks
Embedding：把 chunks 向量化
Vector Store：存入向量数据库（Chroma、FAISS、Pinecone 等）
Retriever：根据用户 query 检索最相关的 k 个 chunks
注入 Prompt：检索结果拼进 prompt → LLM 基于上下文回答

文档与文档加载

LangChain 支持 CSV、目录、HTML、JSON、Markdown、PDF 等格式。加载器的 load 方法把数据源转换成一个或多个 Document 对象（含文本及元数据）。数据以 Document 对象和向量形式在各包装器中流通。

常用文本分割器：

CharacterTextSplitter：基于字符（默认 \n\n）切割，chunk_size 设大小、chunk_overlap 设重叠。
RecursiveCharacterTextSplitter：支持自然语言及多种编程语言（JavaScript、cpp、go、java、php、python 等）代码切割。
MarkdownHeaderTextSplitter：根据指定标题切割 Markdown 文档。

from langchain_core.documents import Document

documents = [
    Document(page_content="猫是柔软可爱的动物，但相对独立", metadata={"source": "常见动物宠物文档"}),
    Document(page_content="狗是人类很早开始的动物伴侣，具有团队能力", metadata={"source": "常见动物宠物文档"}),
    Document(page_content="金鱼是我们常常喂养的观赏动物之一，活泼灵动", metadata={"source": "鱼类宠物文档"}),
    Document(page_content="鹦鹉是猛禽，但能够模仿人类的语言", metadata={"source": "飞禽宠物文档"}),
    Document(page_content="兔子是小朋友比较喜欢的宠物，但是比较难喂养", metadata={"source": "常见动物宠物文档"}),
]

文本嵌入模型与向量数据库

Embeddings 类为多种嵌入模型（OpenAI、Cohere、HuggingFace、DashScope 等）提供统一接口，将文档和查询都转为向量，通过计算向量空间距离寻找最相似文本。它与 LLM 包装器、聊天模型包装器并称三大模型包装器。

import os
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma
from langchain_core.runnables import RunnableLambda
from models import ALI_TONGYI_EMBEDDING_MODEL, ALI_TONGYI_API_KEY_OS_VAR_NAME

llm_embeddings = DashScopeEmbeddings(
    model=ALI_TONGYI_EMBEDDING_MODEL,
    dashscope_api_key=os.getenv(ALI_TONGYI_API_KEY_OS_VAR_NAME)
)

# 实例化向量空间
vector_store = Chroma.from_documents(documents=documents, embedding=llm_embeddings)

print(vector_store.similarity_search("狸花猫"))
# 按分数排序，分数越小越相似
print(vector_store.similarity_search_with_score("狸花猫"))

# 检索器：bind(k=1) 返回相似度最高的第一个
docs_find = RunnableLambda(vector_store.similarity_search).bind(k=1)
print(docs_find.batch(["狸花猫", "海豚"]))

检索器 (Retriever)

直接操作 Chroma、FAISS、Pinecone 等向量库需要了解各自查询语法、连接管理等。LangChain 封装了 VectorStoreRetriever 提供统一接口——在向量库实例上调用 as_retriever() 即得检索器，可轻松切换底层向量库而无需修改查询代码。

完整 RAG Chain（本地文档）

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from models import get_ali_model_client

client = get_ali_model_client()
docs_find = RunnableLambda(vector_store.similarity_search).bind(k=1)

message = """
仅使用提供的上下文回答下面的问题：
{question}
上下文：
{context}
"""
prompt_template = ChatPromptTemplate.from_messages([('human', message)])

chain = {"question": RunnablePassthrough(), "context": docs_find} | prompt_template | client
print(chain.invoke("请介绍一下猫").content)

基于在线网页的 RAG

import os, bs4
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.retrieval import create_retrieval_chain
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import RecursiveCharacterTextSplitter
from models import get_lc_model_client, ALI_TONGYI_EMBEDDING_MODEL, ALI_TONGYI_API_KEY_SYSVAR_NAME

client = get_lc_model_client()
llm_embeddings = DashScopeEmbeddings(
    model=ALI_TONGYI_EMBEDDING_MODEL,
    dashscope_api_key=os.getenv(ALI_TONGYI_API_KEY_SYSVAR_NAME)
)

# 1. 加载网页
loader = WebBaseLoader(
    web_path=["https://www.news.cn/fortune/20250212/895ac6738b7b477db8d7f36c315aae22/c.html"],
    bs_kwargs=dict(parse_only=bs4.SoupStrainer(class_=("main-left left", "title")))
)
docs = loader.load()

# 2. 切割 -> 3. 嵌入入库 -> 4. 检索器
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
documents = splitter.split_documents(docs)
vector_store = Chroma.from_documents(documents=documents, embedding=llm_embeddings)
retriever = vector_store.as_retriever()

system_prompt = "您是问答任务的助理。使用以下检索到的上下文来回答问题。如果你不知道答案，就说你不知道。"
prompt_template = ChatPromptTemplate.from_messages(
    [("system", system_prompt), ("assistant", "{context}"), ("human", "{input}")]
)

# 5. 组合检索链
chain1 = create_stuff_documents_chain(client, prompt_template)
chain2 = create_retrieval_chain(retriever, chain1)
print(chain2.invoke({"input": "张成刚是谁？"})["answer"])

9. Agents 与工具调用

Agent = LLM + 工具调用能力。LLM 不再只是"回答"，而是自主选择调用什么工具、按什么顺序、何时停——规划 → 行动 → 观察 → 再行动的循环。现在火爆的 MCP、Agent2Agent 本质都离不开大模型对工具的调用。

9.1 工具 Tools

工具是代理、链或 LLM 与世界互动的接口，包含：名称、描述、输入的 JSON 模式、要调用的函数、是否将结果直接返回用户。Toolkits 是面向特定目标的多个工具的集合。

9.2 工具调用的几种方式

# 方式一：bind_tools 把工具绑定到模型
import pymysql; pymysql.install_as_MySQLdb()
import os
from langchain.messages import HumanMessage
from langchain_community.utilities import SQLDatabase
from langchain.tools import tool
from models import get_lc_model_client

client = get_lc_model_client()
db = SQLDatabase.from_uri('mysql+mysqldb://root:root123456@127.0.0.1:3306/world?charset=utf8mb4')

@tool
def get_table_names():
    """获取数据库中的所有表名"""
    return db.get_table_names()

client_with_tools = client.bind_tools([get_table_names])
resp = client_with_tools.invoke([HumanMessage(content="请从国家表中查询出 China 的所有数据")])
print(resp.tool_calls)  # 模型决定调用哪个工具

# 方式二：使用内置工具（如 Tavily 搜索）
from langchain_community.tools import TavilySearchResults

search = TavilySearchResults(max_results=2)
client_with_tools = client.bind_tools([search])
result = client_with_tools.invoke([HumanMessage(content="长沙的天气如何？")])
print(result.tool_calls)

9.3 v1.0 推荐：`create_agent`

create_agent 自动处理"调用工具 → 观察结果 → 再决策"的完整循环，并基于 LangGraph 支持记忆、状态、持久化。

from datetime import datetime
import subprocess, webbrowser, os
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

@tool
def get_current_time(input: str = "") -> str:
    """获取当前时间"""
    return f"当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}。"

@tool
def open_calc(input: str = "") -> str:
    """打开计算器"""
    try:
        subprocess.Popen(['calc.exe'])
        return "计算器已打开"
    except Exception as e:
        return f"打开计算器失败: {str(e)}"

@tool
def open_browser(url: str) -> str:
    """打开浏览器访问指定网址"""
    webbrowser.open(url)
    return f"已打开浏览器访问 {url}"

tools = [get_current_time, open_calc, open_browser]

llm = ChatOpenAI(
    openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
    openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
    model_name="qwen-max"
)

agent = create_agent(
    model=llm,
    tools=tools,
    system_prompt="你是人工智能助手，需要帮助用户解决各种问题。",
    checkpointer=InMemorySaver()  # 短期记忆
)

response = agent.invoke(
    {"messages": [{"role": "user", "content": "现在几点了？"}]},
    config={"configurable": {"thread_id": "user_1"}}
)
print(response["messages"][-1].content)

9.4 实战：用 Agent 查询数据库

SQLDatabaseToolkit 提供一组数据库工具，配合精心设计的 system_prompt（强制先查表名和 Schema、禁止猜测、禁止写操作），让 Agent 安全地自然语言查库。

import pymysql; pymysql.install_as_MySQLdb()
from langchain.agents import create_agent
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
from langchain.messages import AIMessage
from models import get_ali_model_client, ALI_TONGYI_MAX_MODEL

client = get_ali_model_client(model=ALI_TONGYI_MAX_MODEL)
db = SQLDatabase.from_uri('mysql+mysqldb://root:root123456@127.0.0.1:3306/world?charset=utf8mb4')

toolkit = SQLDatabaseToolkit(db=db, llm=client)

system_prompt = """你是一个精通 MySQL 数据库的 AI 助手。
【极其重要的规则】
1. 绝对不要猜测表名。必须先用工具查看数据库里有哪些表。
2. 绝对不要猜测字段名。确定表名后必须查看表的 Schema。
3. 只有确认了表名和字段名之后，才能编写 SQL。
4. 如果 SQL 执行报错，仔细查看错误信息，修正后重试。
注意：永远不要执行 DROP/DELETE/INSERT/UPDATE 等修改操作；只输出最终自然语言回答。
"""

agent = create_agent(model=client, tools=toolkit.get_tools(), system_prompt=system_prompt)

result = agent.invoke({"messages": [("user", "请从国家表中查询出 China 的相关数据")]})

# 提取实际执行的 SQL
for msg in result['messages']:
    if isinstance(msg, AIMessage) and msg.tool_calls:
        for tc in msg.tool_calls:
            if tc['name'] == 'sql_db_query':
                print("执行的 SQL:", tc['args'].get('query'))

print(result["messages"][-1].content)

10. 中间件 (Middleware)

中间件是一种流程控制机制，用于在智能体执行过程中拦截、修改或增强请求与响应的处理逻辑，无需修改核心 Agent 或工具代码。它在每个步骤之前/之后暴露钩子（hooks），是 v1.0 最大的亮点之一。

10.1 内置中间件

参考：https://docs.langchain.com/oss/python/langchain/middleware/built-in

PIIMiddleware：发送至模型前屏蔽敏感信息
SummarizationMiddleware：对话历史过长时自动压缩
HumanInTheLoopMiddleware：工具调用执行前暂停，等待人工审批 / 编辑 / 拒绝
ModelCallLimitMiddleware：限制模型调用次数，防止无限循环或过高成本
ToolCallLimitMiddleware：限制工具调用次数（全局或特定工具）

# PIIMiddleware 演示：脱敏
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain.tools import tool
from models import get_lc_model_client

@tool
def read_user_data(user_id: str) -> str:
    """读取用户数据的工具"""
    return f"用户 {user_id} 的数据：姓名张三，邮箱 zhangsan@example.com，手机号 13800138000"

agent = create_agent(
    model=get_lc_model_client(),
    tools=[read_user_data],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True, apply_to_output=True),
        PIIMiddleware(
            "phone",
            detector=r"(?:13[0-9]|14[01456879]|15[0-35-9]|16[2567]|17[0-8]|18[0-9]|19[0-35-9])\d{8}",
            strategy="redact", apply_to_input=True, apply_to_output=True
        ),
    ],
    system_prompt="你是一个安全的数据处理助手，严格保护用户隐私"
)

result = agent.invoke({"messages": [{"role": "user", "content": "请读取用户 001 的数据"}]})
print(result["messages"][-1].content)

10.2 自定义中间件

通过在执行流程的特定点运行钩子构建。基于装饰器适合单钩子的快速实现；基于类（继承 AgentMiddleware）适合多钩子的复杂场景。

import re
from typing import Any, Dict
from langchain.agents.middleware import AgentMiddleware

class SensitiveDataMiddleware(AgentMiddleware):
    """自定义脱敏中间件"""

    def __init__(self, patterns: list = None):
        super().__init__()
        self.patterns = patterns or [
            (r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+', '[EMAIL]'),
            (r'(\+86)?1[3-9]\d{9}', '[PHONE]')
        ]

    def _desensitize(self, text: str) -> str:
        for pattern, replacement in self.patterns:
            text = re.sub(pattern, replacement, text)
        return text

    def before_model(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """模型调用前脱敏消息内容"""
        for message in state.get('messages', []):
            if hasattr(message, 'content') and isinstance(message.content, str):
                message.content = self._desensitize(message.content)
        return state

    def after_model(self, state: Dict[str, Any]) -> Dict[str, Any]:
        return state

11. 生态演进与总结

11.1 生态演进

LangChain v0.1 (2023)     LangChain v0.2-0.3 (2024)      LangChain v1.0 (2025+)
┌─────────────┐          ┌─────────────────────┐       ┌──────────────────┐
│ Chain 为主   │ ──────▶  │ LCEL / Runnable 为主 │ ────▶ │ LangGraph 为主    │
│ AgentExecutor│          │ 流式/批处理/并发     │       │ 有状态图编排      │
│ 记忆模块     │          │ 社区包拆分           │       │ create_agent +    │
│             │          │                      │       │ Middleware        │
└─────────────┘          └─────────────────────┘       └──────────────────┘

LangGraph：最值得关注的新核心。用有向图（StateGraph）定义 Agent 状态机，支持条件分支、循环、人机交互、持久化。
LangSmith：调试和监控 LLM 应用的 SaaS 平台（trace、evaluation、prompt hub）。
LangServe：一键把 LangChain Runnable 部署成 REST API。

11.2 典型场景速查

场景	关键组件	复杂度
聊天机器人	ChatModel + Memory	⭐
文档问答 (RAG)	Loader → Split → Embed → VectorStore → Retriever	⭐⭐
SQL 自然语言查询	create_agent + SQLDatabaseToolkit	⭐⭐
多步推理 Agent	create_agent + 多个 Tool	⭐⭐⭐
复杂工作流	LangGraph StateGraph + 条件分支 + Checkpointer	⭐⭐⭐⭐

11.3 核心要点

要点	一句话
本质	LLM 应用的中间件 / 胶水层，不是模型也不是应用
核心抽象	Chain（流水线）→ Agent（决策循环）→ Graph（状态机）
v1.0 标志	`create_agent` 取代 AgentExecutor，全面基于 LangGraph，引入中间件
学习路径	LCEL 管道 → 输出解析 → Memory → RAG → Tool → create_agent → Middleware
避坑	别追旧版 AgentExecutor；用 `create_agent`；用 `langchain-<provider>` 而非全局安装；旧代码装 `langchain-classic` 过渡
竞品	LlamaIndex（偏数据 / 索引）、Dify / Coze（低代码）、自研裸调 SDK