LangChain 教程

LangChain 教程

LangChain 是一个用于构建 LLM（大语言模型）应用的 Python/JS 框架，提供模型调用、提示词管理、工具调用、记忆、Agent 等开箱即用的组件。

它的核心价值在于：把「调用大模型」这件简单但琐碎的事，抽象成一套可组合的模块化 API，让你专注业务逻辑而不是底层细节。

核心组件

模块	作用
Chat Models	统一调用各种 LLM（DeepSeek、OpenAI、通义等）
Prompt	提示词模板与消息组装
Tools	让模型调用外部函数（搜索、计算、API）
Agent	让模型自主决策：分析任务→调用工具→完成
Memory	对话记忆与跨会话存储
RAG	检索增强生成：接入私有知识库

快速示例

from langchain.chat_models import init_chat_model

# 一行初始化模型(自动识别提供商)
model = init_chat_model("deepseek-chat", model_provider="deepseek")

resp = model.invoke("用一句话介绍 LangChain")
print(resp.content)

LangChain 简介

LangChain 于 2022 年由 Harrison Chase 创建，是目前最流行的 LLM 应用开发框架之一。它解决的核心问题是：让大模型与你的数据、工具、系统协同工作。

为什么需要 LangChain

• 多模型统一：一套 API 切换 DeepSeek / OpenAI / 通义 / 本地模型
• 提示词工程：模板化、结构化管理提示词
• 工具调用：模型可以调用函数、搜索、查数据库
• 记忆管理：对话上下文自动维护
• 应用模式：RAG、Agent、多 Agent 协作开箱即用

核心概念

概念	一句话理解
Chat Model	对话型大模型封装
Message	一条对话消息（角色+内容）
Prompt	发给模型的指令模板
Tool	模型可调用的外部函数
Agent	会「思考+行动」的模型循环

典型应用场景

智能客服    → RAG + 工具
个人助手    → Agent + 记忆
知识库问答  → RAG + 向量检索
自动化流程  → 多 Agent 协作

LangChain 环境搭建

搭建 LangChain 开发环境：安装依赖 + 配置模型 API，五分钟跑通第一个对话。

1. 安装

# Python 3.9+
pip install langchain langchain-core langchain-openai

# 使用 DeepSeek / 通义等兼容 OpenAI 接口的模型
pip install langchain-openai

# LangGraph(Agent 工作流)
pip install langgraph

2. 配置 API Key

# DeepSeek
export DEEPSEEK_API_KEY="sk-xxxx"

# OpenAI
export OPENAI_API_KEY="sk-xxxx"

# 阿里百炼(通义)
export DASHSCOPE_API_KEY="sk-xxxx"

3. 第一个对话

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "deepseek-chat",
    model_provider="deepseek",
    temperature=0.7,
)

resp = model.invoke("你好，请介绍一下你自己")
print(resp.content)

4. 验证安装

python -c "import langchain; print(langchain.__version__)"
# 输出 0.3.x 或 1.x 即成功

LangChain DeepSeek

DeepSeek 是国内热门的开源大模型提供商，API 兼容 OpenAI 格式，LangChain 可直接接入。

接入 DeepSeek

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "deepseek-chat",          # 模型名
    model_provider="deepseek",
    api_key="sk-xxx",         # 或用环境变量 DEEPSEEK_API_KEY
    temperature=0.6,
)

resp = model.invoke("解释一下什么是 RAG")
print(resp.content)

DeepSeek 模型列表

模型	特点
deepseek-chat	通用对话（V3 系列）
deepseek-reasoner	推理增强（R1 系列，展示思考过程）

流式输出

for chunk in model.stream("写一首关于春天的诗"):
    print(chunk.content, end="", flush=True)

设置 base_url(自建或其他兼容服务)

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="deepseek-chat",
    api_key="sk-xxx",
    base_url="https://api.deepseek.com/v1",   # 兼容 OpenAI 接口
)

LangChain 集成阿里百炼

阿里云百炼（DashScope） 提供通义千问系列模型，LangChain 通过 langchain-openai 或专用包接入。

方式一：兼容 OpenAI 接口

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="qwen-plus",
    api_key="sk-xxx",                 # DASHSCOPE_API_KEY
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

resp = model.invoke("介绍杭州")
print(resp.content)

方式二：init_chat_model

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "qwen-plus",
    model_provider="alibabacloud",    # 或 "dashscope"
    temperature=0.5,
)
print(model.invoke("你好").content)

常用通义模型

模型	适用
qwen-plus	综合能力均衡，日常对话
qwen-max	最强推理，复杂任务
qwen-turbo	快速低延迟
qwen-long	超长上下文

流式与多轮

from langchain_core.messages import HumanMessage, SystemMessage

messages = [
    SystemMessage("你是一个中文助手"),
    HumanMessage("今天天气如何？"),
]
resp = model.invoke(messages)
print(resp.content)

LangChain Agent

Agent（智能体） 是 LangChain 的核心：让大模型自主思考、调用工具、反复迭代直到完成任务。

Agent 是什么

普通对话:  用户 → 模型 → 回答
Agent 循环: 用户 → 模型(分析) → 调用工具 → 模型(再分析) → 完成

最简单的人机交互 Agent

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

model = init_chat_model("deepseek-chat", model_provider="deepseek")
agent = create_agent(model, [add])

result = agent.invoke({"messages": [("user", "123 + 456 等于多少？")]})
print(result["messages"][-1].content)   # 579

Agent 的关键概念

概念	说明
Tools	模型可调用的函数（带描述+参数模式）
ReAct	思考(Reason)+行动(Act)循环模式
State	Agent 的运行时状态（消息、中间结果）
Middleware	拦截/修改模型调用与工具调用

LangChain 模型调用

LangChain 统一了各种模型的调用方式：同步、异步、流式、批量，以及带消息的多轮对话。

初始化模型

from langchain.chat_models import init_chat_model

# 按名称+提供商初始化
model = init_chat_model("deepseek-chat", model_provider="deepseek")

# 或直接用具体类
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

四种调用方式

# 1. 同步调用
resp = model.invoke("你好")
print(resp.content)

# 2. 异步调用
import asyncio
resp = asyncio.run(model.ainvoke("你好"))

# 3. 流式输出
for chunk in model.stream("写一篇短文"):
    print(chunk.content, end="")

# 4. 批量调用
resps = model.batch(["问题1", "问题2", "问题3"])
for r in resps:
    print(r.content)

多轮对话(消息列表)

from langchain_core.messages import (
    SystemMessage, HumanMessage, AIMessage
)

messages = [
    SystemMessage("你是资深的 Python 工程师"),
    HumanMessage("什么是装饰器？"),
    AIMessage("装饰器是…（模型之前的回答）"),
    HumanMessage("能给个例子吗？"),
]
resp = model.invoke(messages)
print(resp.content)

LangChain 常用参数

模型调用参数控制生成行为：温度、最大长度、随机种子、停止词等，理解它们才能调出稳定高质量的输出。

常用参数速查

model = init_chat_model("deepseek-chat", model_provider="deepseek",
    temperature=0.7,        # 随机性 0~2,越低越确定
    max_tokens=2048,        # 最大生成长度
    timeout=60,             # 请求超时(秒)
    max_retries=3,          # 失败重试次数
)

temperature 的作用

值	行为	场景
0	确定性输出	提取、分类、代码生成
0.3-0.7	平衡	日常对话、摘要
0.8-1.5	创造性	写作、头脑风暴

按调用覆盖参数

# 初始化时设默认,调用时临时覆盖
resp = model.invoke(
    "写一个科幻故事",
    temperature=1.2,        # 本次调用更高随机
    max_tokens=500,
)
print(resp.content)

绑定参数(bind)

# 固定参数绑定到模型,后续调用自动带上
structured_model = model.bind(
    temperature=0,
    response_format={"type": "json_object"},
)

# 也可以绑定工具(供模型选择调用)
model_with_tool = model.bind_tools([my_tool])

LangChain Chat Model 高级用法

进阶掌握 Chat Model：响应对象结构、函数/工具绑定、JSON 模式、模型切换。

响应对象结构

resp = model.invoke("你好")

resp.content          # 文本内容
resp.role             # 'assistant'
resp.id               # 消息 ID
resp.response_metadata  # 元数据(耗时/token等)
# 使用量统计
usage = resp.usage_metadata
print(usage["input_tokens"], usage["output_tokens"])

绑定工具(bind_tools)

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气"""
    return f"{city} 今天晴,25°C"

model_with_tools = model.bind_tools([get_weather])

resp = model.invoke("北京天气怎么样？")
# 模型可能返回工具调用意图:
# resp.tool_calls -> [{'name': 'get_weather', 'args': {'city': '北京'}, ...}]
if resp.tool_calls:
    for call in resp.tool_calls:
        print(call["name"], call["args"])

JSON 结构化输出

model_json = model.bind(response_format={"type": "json_object"})

resp = model_json.invoke(
    '把"北京 25 度,晴"提取为 JSON,字段: city, temp, weather'
)
print(resp.content)
# {"city": "北京", "temp": 25, "weather": "晴"}

动态切换模型

# 普通问答用便宜模型,复杂任务换强模型
def get_model(task: str):
    if "复杂" in task or "推理" in task:
        return init_chat_model("deepseek-reasoner", model_provider="deepseek")
    return init_chat_model("deepseek-chat", model_provider="deepseek")

LangChain 消息类型

LangChain 用消息（Message）表示对话内容，不同角色对应不同消息类型。

消息类型

from langchain_core.messages import (
    SystemMessage,      # 系统设定(角色、规则)
    HumanMessage,       # 用户输入
    AIMessage,          # 模型回复
    ToolMessage,        # 工具执行结果
)

system = SystemMessage("你是一个友好的助手")
user = HumanMessage("你好")
ai = AIMessage("你好！有什么可以帮你？")
tool_result = ToolMessage("查询结果: 晴,25°C", tool_call_id="call_123")

构建多轮对话

messages = [
    SystemMessage("你是代码助手"),
    HumanMessage("帮我写一个阶乘函数"),
    AIMessage("好的,以下是 Python 实现:

def fact(n):
    return 1 if n <= 1 else n * fact(n-1)"),
    HumanMessage("能用递归解释一下吗？"),
]
resp = model.invoke(messages)

从字符串快速创建

# 简洁写法:元组形式 (role, content)
messages = [
    ("system", "你是翻译助手"),
    ("human", "把 hello 翻译成中文"),
]
resp = model.invoke(messages)
print(resp.content)

消息工具函数

from langchain_core.messages import AnyMessage, AIMessageChunk

# 类型判断
is_ai = isinstance(msg, AIMessage)

# 流式合并(分块累积成完整消息)
full = AIMessageChunk(content="")
for chunk in model.stream("写诗"):
    full += chunk        # 支持 += 合并
print(full.content)

LangChain @tool

@tool 装饰器 把普通 Python 函数变成「模型可调用的工具」：自动提取函数签名、docstring 作为工具描述。

最简工具

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """两个数字相乘。"""
    return a * b

# 查看工具信息
print(multiply.name)         # multiply
print(multiply.description)  # 两个数字相乘。
print(multiply.args_schema)  # 参数 JSON Schema

带类型与默认值的工具

from typing import Optional

@tool
def search(keyword: str, limit: int = 5) -> str:
    """搜索知识库,返回前 limit 条结果。"""
    # 实际调用搜索引擎/数据库
    results = [f"结果{i}: {keyword} 相关内容" for i in range(limit)]
    return "
".join(results)

工具名与描述自定义

@tool("query_weather")
def get_weather(city: str) -> str:
    """获取指定城市当前天气,支持中文城市名。"""
    return f"{city}: 晴 25°C"

# 或
@tool(parse_docstring=True)   # 解析 docstring 中的 Args/Returns
def add(a: int, b: int) -> int:
    """两数相加。

    Args:
        a: 第一个加数
        b: 第二个加数

    Returns:
        两数之和
    """
    return a + b

工具与模型配合

from langchain.agents import create_agent

tools = [multiply, search]
model = init_chat_model("deepseek-chat", model_provider="deepseek")
agent = create_agent(model, tools)

resp = agent.invoke({"messages": [("user", "帮我搜索 LangChain 教程,并计算 8*7")]})
print(resp["messages"][-1].content)

LangChain 工具高级特性

工具的进阶用法：重试、错误处理、结构化参数、异步工具、StructuredTool。

1. 结构化参数(Pydantic)

from pydantic import BaseModel, Field
from langchain_core.tools import StructuredTool

class BookQuery(BaseModel):
    keyword: str = Field(description="搜索关键词")
    category: str = Field(default="all", description="分类")
    page: int = Field(default=1, ge=1, description="页码")

def search_books(query: BookQuery) -> str:
    """按条件搜索图书"""
    return f"搜索[{query.category}] {query.keyword} 第{query.page}页"

tool = StructuredTool.from_function(
    func=search_books,
    name="search_books",
    description="在书库中搜索图书",
    args_schema=BookQuery,
)

2. 工具错误处理

from langchain_core.tools import ToolException

@tool(handle_tool_error=True)   # 错误返回给模型,而不是崩溃
def safe_divide(a: float, b: float) -> float:
    """安全的除法"""
    if b == 0:
        raise ToolException("除数不能为 0")
    return a / b

# handle_tool_error=True 时,异常会转成工具错误消息
# 模型可以据此修正参数重试

3. 异步工具

import asyncio

@tool
async def fetch_url(url: str) -> str:
    """异步抓取网页内容"""
    # 模拟异步请求
    await asyncio.sleep(0.1)
    return f"...{url} 的内容..."

# Agent 支持异步工具,用 ainvoke 调用
result = asyncio.run(agent.ainvoke({"messages": [("user", "抓取 example.com")]}))

4. 工具装饰器传参

@tool(
    name="get_time",
    description="获取当前时间,可指定时区",
    response_format="content_and_artifact",   # 返回内容+额外数据
)
def get_time(tz: str = "Asia/Shanghai"):
    return f"当前时间 {tz}", {"tz": tz}   # (content, artifact)

LangChain 工具访问

工具访问（Tool access）指工具在 Agent 中的注册、检索与权限控制，以及多 Agent 间共享工具的方式。

1. 工具注册与检索

from langchain_core.tools import tool

@tool
def search_docs(q: str) -> str: """搜索文档""" ...

@tool
def calculate(expr: str) -> str: """数学计算""" ...

# 收集到列表,传给 Agent
tools = [search_docs, calculate]
agent = create_agent(model, tools)

# 按需检索(工具多时)
from langchain_core.tools import tool as _t
all_tools = [search_docs, calculate, get_weather]
selected = [t for t in all_tools if "search" in t.name or "calculate" in t.name]

2. 工具调用解析

# 手动解析模型返回的工具调用
resp = model.bind_tools(tools).invoke("算一下 15*4 并搜索 LangChain")
for call in resp.tool_calls:
    name = call["name"]
    args = call["args"]
    print(f"调用 {name} 参数 {args}")

    # 找到工具并执行
    tool_map = {t.name: t for t in tools}
    if name in tool_map:
        result = tool_map[name].invoke(args)
        print("结果:", result)

3. 工具权限(简单白名单)

ALLOWED = {"search_docs", "get_weather"}   # 只允许读类工具

def filter_tools(all_tools, allowed):
    return [t for t in all_tools if t.name in allowed]

safe_agent = create_agent(model, filter_tools(tools, ALLOWED))

4. 动态工具(按用户提供)

# 运行时根据用户需求绑定工具
def make_agent(user_tools):
    """用户勾选了哪些工具就注入哪些"""
    return create_agent(model, user_tools)

LangChain create_agent() 函数

create_agent() 是 LangChain 1.x 中最简单的 Agent 创建方式：一行生成「模型 + 工具」的对话式智能体。

基本用法

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_time(tz: str = "Asia/Shanghai") -> str:
    """获取指定时区的当前时间"""
    from datetime import datetime
    return datetime.now().astimezone().strftime("%Y-%m-%d %H:%M")

model = init_chat_model("deepseek-chat", model_provider="deepseek")
agent = create_agent(model, [get_time])

result = agent.invoke({"messages": [("user", "现在几点了?")]})
print(result["messages"][-1].content)

支持的系统提示

agent = create_agent(
    model,
    [get_time],
    system_prompt="你是一个时间助手,只回答时间相关问题。",
)

# 或传入列表(可包含图片等)
agent = create_agent(model, [get_time], system_prompt=[("system", "..." )])

多轮对话

# 每次 invoke 传入完整消息历史即可保持上下文
result1 = agent.invoke({"messages": [("user", "现在几点?")]})
result2 = agent.invoke({
    "messages": result1["messages"] + [("user", "那东京呢?")]
})
print(result2["messages"][-1].content)

返回结构

result = agent.invoke({"messages": [("user", "几点了?")]})

result["messages"]   # 完整消息列表(含工具调用记录)
result["messages"][-1]  # 最终回答
# 若工具被调用,中间会有 ToolMessage

# 只取最终答案
print(result["messages"][-1].content)

LangChain Agent 工作流程

理解 Agent 的内部工作流程：模型如何思考、调用工具、看到结果、继续行动，直到产出最终答案。

Agent 循环(ReAct)

┌────────────────────────────────────┐
│ 1. 用户提问                         │
│ 2. 模型: 分析任务,决定是否调工具    │
│ 3. 调用工具(如搜索/计算)           │
│ 4. 工具返回结果(ToolMessage)       │
│ 5. 模型: 基于结果继续推理           │
│    ├─ 需要更多信息 → 回到第3步     │
│    └─ 有答案 → 输出最终回答         │
└────────────────────────────────────┘

观察一次完整循环

@tool
def weather(city: str) -> str:
    """查询城市天气"""
    return f"{city}: 晴,28°C"

agent = create_agent(model, [weather])
result = agent.invoke({"messages": [("user", "北京适合穿什么?")]})

# 打印整个消息链,观察 Agent 的思考过程
for m in result["messages"]:
    role = type(m).__name__
    content = str(m.content)[:80]
    if m.tool_calls:
        content = f"调用工具: {m.tool_calls[0]['name']}({m.tool_calls[0]['args']})"
    print(f"[{role}] {content}")

关键设计原则

原则	说明
工具描述清晰	模型靠描述决定何时用、怎么用
结果可理解	工具返回结构化、自包含的结果
错误可恢复	工具失败给模型可修正的信息
限制循环次数	防止无限循环(超时/步数上限)

控制循环

# LangGraph 中可设置最大步数
from langgraph.graph import StateGraph, START, END

# create_agent 底层就是 LangGraph,
# 复杂需求可直接拿到 graph 扩展
graph = agent.get_graph()
print(graph.to_json())   # 查看图结构

LangChain AgentState

AgentState 是 Agent 运行时的状态对象：保存消息、中间结果等，LangGraph 用它驱动整个工作流。

什么是 AgentState

# create_agent 创建的 Agent,内部状态结构:
class AgentState(TypedDict):
    messages: list   # 消息列表(核心状态)

# 每次 invoke 传入 {"messages": [...]},
# 结束时也返回 {"messages": [...]}

自定义状态(进阶)

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class MyState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息自动累加
    user_name: str          # 自定义字段
    step_count: int         # 记录步数

# 在 LangGraph 中构建自定义状态流
from langgraph.graph import StateGraph, START, END

graph = StateGraph(MyState)
graph.add_node("agent", agent_node)
graph.add_edge(START, "agent")
graph.add_edge("agent", END)
app = graph.compile()

# 运行
result = app.invoke({
    "messages": [("user", "你好")],
    "user_name": "张三",
    "step_count": 0,
})

状态访问与更新

# 节点内读写状态
def agent_node(state: MyState):
    # 读
    msgs = state["messages"]
    name = state.get("user_name", "游客")
    # 返回更新(部分更新)
    return {
        "messages": [("ai", f"你好,{name}!")],
        "step_count": state["step_count"] + 1,
    }

为什么状态重要

能力	说明
消息持久化	多轮对话不丢上下文
跨节点传递	多个节点共享业务数据
可检查	任意时刻可查看状态调试
可恢复	断点续跑、人工介入

LangChain 提示词

提示词（Prompt）管理：模板化、变量注入、少样本示例，让模型输出更可控。

PromptTemplate

from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate.from_template(
    "你是{role}。请用{language}回答:{question}"
)

# 填充变量
final_prompt = prompt.format(
    role="资深前端工程师",
    language="中文",
    question="什么是 Flex 布局?",
)
print(final_prompt)

ChatPromptTemplate(推荐)

from langchain_core.prompts import ChatPromptTemplate

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role},回答风格{style}。"),
    ("human", "{question}"),
])

messages = chat_prompt.invoke({
    "role": "专业教练",
    "style": "简洁有力",
    "question": "如何坚持跑步?",
})
resp = model.invoke(messages)

少样本示例(Few-shot)

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "把输入翻译成英文"),
    ("human", "苹果"),
    ("ai", "apple"),
    ("human", "香蕉"),
    ("ai", "banana"),
    ("human", "{word}"),
])

resp = model.invoke(chat_prompt.invoke({"word": "橘子"}))
print(resp.content)   # orange

提示词与 Agent 结合

agent = create_agent(
    model,
    tools,
    system_prompt=(
        "你是客服助手。回答问题前,必须先调用 search_docs 查资料;"
        "资料不足时如实说明,不要编造。"
    ),
)

LangChain 流式输出

流式输出（Streaming）让模型逐字返回结果，用户看到打字机效果，也适合边生成边处理。

基础流式

for chunk in model.stream("写一段 100 字的自我介绍"):
    print(chunk.content, end="", flush=True)

# chunk 是 AIMessageChunk,content 为增量片段

流式回调(on_llm_new_token)

from langchain_core.callbacks import BaseCallbackHandler

class PrintToken(BaseCallbackHandler):
    def on_llm_new_token(self, token, **kwargs):
        print(token, end="", flush=True)

handler = PrintToken()
for chunk in model.stream("讲个笑话", config={"callbacks": [handler]}):
    pass   # 回调已打印

Agent 流式

# Agent 流式输出(含工具调用过程)
async for event in agent.astream_events(
    {"messages": [("user", "帮我查天气并算 5+3")]},
    version="v2",
):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        token = event["data"]["chunk"].content
        if token:
            print(token, end="", flush=True)
    elif kind == "on_tool_start":
        print("
[调用工具]", event["name"], event["data"].get("input"))

流式 SSE 到前端

# FastAPI 示例:SSE 推送流式内容
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/chat")
def chat(q: str):
    def gen():
        for chunk in model.stream(q):
            yield f"data: {chunk.content}

"
    return StreamingResponse(gen(), media_type="text/event-stream")

LangChain 结构化输出

结构化输出让模型返回固定格式数据（JSON、Pydantic 对象），方便程序消费。

1. with_structured_output

from pydantic import BaseModel, Field

class Movie(BaseModel):
    title: str = Field(description="电影名")
    year: int = Field(description="上映年份")
    rating: float = Field(description="评分,0-10")
    genres: list[str] = Field(description="类型列表")

model_structured = model.with_structured_output(Movie)

movie = model_structured.invoke("推荐一部 2023 年的科幻电影")
print(movie.title)      # 直接是 Movie 对象
print(movie.year, movie.rating, movie.genres)

2. 输出 JSON 字典

model_json = model.with_structured_output(dict)

data = model_json.invoke("《流浪地球2》的导演是谁?输出 {director: 名字}")
print(data)   # {'director': '郭帆'}

3. 列表输出

class Item(BaseModel):
    name: str
    price: float

model_list = model.with_structured_output(list[Item])

items = model_list.invoke("列出 3 种常见水果及大致价格")
for it in items:
    print(it.name, it.price)

4. 与 Agent 结合

# Agent 最终回答也可是结构化数据
result = agent.invoke({"messages": [("user", "查天气并给出 JSON")]})
# 配合 system_prompt 要求输出 JSON 即可

# 或直接给模型绑定输出模式
agent_model = create_agent(
    model.with_structured_output(dict),
    tools,
)

LangChain 输出策略

输出策略解决「怎么拿结果」：直接返回、带元数据、解析器后处理、只取部分字段。

1. 直接返回内容

resp = model.invoke("你好")
text = resp.content              # 纯文本
print(text)

2. 带元数据返回

resp = model.invoke("写一句话")
print(resp.response_metadata)    # token 用量、耗时等
usage = resp.usage_metadata
if usage:
    print(f"输入 {usage['input_tokens']} tokens")
    print(f"输出 {usage['output_tokens']} tokens")

3. 输出解析器

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
text = parser.invoke(model.invoke("你好"))
# 直接把 AIMessage 转成纯字符串

# 链式管道(LCEL)
chain = model | parser
result = chain.invoke("讲个冷笑话")
print(result)   # 直接是字符串

4. 组合策略示例

from langchain_core.output_parsers import JsonOutputParser

json_parser = JsonOutputParser()
json_chain = model | json_parser
data = json_chain.invoke('输出 JSON: {"name":"张三","age":25}')
print(data["name"])    # 已解析为 Python 字典

5. 链式调用(LCEL 管道)

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

prompt = ChatPromptTemplate.from_template(
    "把这句话翻译成英文: {sentence}"
)
chain = prompt | model | StrOutputParser()

result = chain.invoke({"sentence": "你好,世界"})
print(result)   # Hello, world

LangChain 中间件

中间件（Middleware）在 Agent 的模型调用、工具调用前后插入自定义逻辑：日志、限流、缓存、拦截、注入上下文。

什么是中间件

Agent 执行链:
消息 → [中间件A] → 模型调用 → [中间件B] → 工具调用 → ...

中间件可以:
- 在调用前:记录、校验、改写请求
- 在调用后:记录、处理结果、缓存

创建中间件

from langchain.agents.middleware import Middleware
from langchain_core.messages import AnyMessage
from typing import Callable, Iterator

class LogMiddleware(Middleware):
    """记录每次模型调用"""
    def before_model(self, messages: list[AnyMessage]) -> list[AnyMessage]:
        print(f"[模型调用前] 消息数: {len(messages)}")
        return messages

    def after_model(self, response: AnyMessage) -> AnyMessage:
        print(f"[模型调用后] 回复: {str(response.content)[:50]}...")
        return response

agent = create_agent(model, tools, middleware=[LogMiddleware()])

中间件方法

方法	时机
before_model	模型调用前(可改消息)
after_model	模型调用后(可改回复)
before_tool	工具执行前
after_tool	工具执行后

限流中间件示例

import time

class RateLimitMiddleware(Middleware):
    def __init__(self, min_interval: float = 1.0):
        self.min_interval = min_interval
        self.last_call = 0.0

    def before_model(self, messages):
        wait = self.min_interval - (time.time() - self.last_call)
        if wait > 0:
            time.sleep(wait)
        return messages

    def after_model(self, response):
        self.last_call = time.time()
        return response

LangChain 中间件钩子

钩子（Hooks）是中间件的细粒度拦截点：针对「某次具体调用」注入逻辑，比整个中间件更灵活。

钩子触发时机

before_model(messages)
    ↓ 模型返回
after_model(response)

before_tool(tool_call)
    ↓ 工具执行
after_tool(result)

前后钩子都可以:
- 修改输入/输出
- 提前返回(短路)
- 抛出异常中止

实现钩子

from langchain.agents.middleware import Middleware
from langchain_core.messages import ToolMessage, AIMessage

class AuditMiddleware(Middleware):
    """审计所有工具调用"""
    def before_tool(self, tool_call: dict) -> dict:
        print(f"[审计] 调用工具 {tool_call['name']}, 参数 {tool_call['args']}")
        return tool_call   # 原样放行

    def after_tool(self, result: str, tool_call: dict) -> str:
        print(f"[审计] 工具结果: {str(result)[:60]}")
        return result

修改模型回复(注入提示)

class FormatMiddleware(Middleware):
    """强制回复以特定格式结尾"""
    def after_model(self, response):
        if isinstance(response, AIMessage):
            content = response.content
            if not content.endswith("

—— 由 AI 助手生成"):
                response.content = content + "

—— 由 AI 助手生成"
        return response

短路提前返回

class CacheMiddleware(Middleware):
    """缓存命中则跳过模型调用"""
    cache = {}

    def before_model(self, messages):
        key = str(messages[-1].content)
        if key in self.cache:
            from langchain_core.messages import AIMessage
            # 构造"假"的模型回复,不再真正调用
            return self.cache[key]  # 返回消息列表则短路
        return messages

LangChain 模型调用拦截

模型调用拦截在请求到达模型之前捕获/改写：记录成本、注入系统上下文、实现缓存、模拟故障。

1. 拦截记录

class ModelLogMiddleware(Middleware):
    def before_model(self, messages):
        # 记录请求
        print(f"→ 请求模型, 最后一条: {messages[-1].content[:60]}")
        return messages

    def after_model(self, response):
        # 记录响应与用量
        usage = getattr(response, "usage_metadata", None)
        if usage:
            print(f"← 回复完成, tokens: {usage}")
        return response

agent = create_agent(model, tools, middleware=[ModelLogMiddleware()])

2. 注入系统上下文

class ContextInjectMiddleware(Middleware):
    """每次对话自动附加当前用户/时间上下文"""
    def __init__(self, user_name: str):
        self.user_name = user_name

    def before_model(self, messages):
        from langchain_core.messages import SystemMessage
        # 在消息头部注入系统信息
        return [
            SystemMessage(
                f"当前用户: {self.user_name}; 当前时间: {__import__('datetime').datetime.now()}"
            )
        ] + messages

3. 模型缓存

import hashlib

class ModelCacheMiddleware(Middleware):
    def __init__(self):
        self.cache = {}

    def before_model(self, messages):
        key = hashlib.md5(str(messages).encode()).hexdigest()
        if key in self.cache:
            from langchain_core.messages import AIMessage
            return [AIMessage(content=self.cache[key])]  # 短路
        self._key = key
        return messages

    def after_model(self, response):
        self.cache[self._key] = response.content
        return response

4. 失败注入(测试用)

class FaultInjectMiddleware(Middleware):
    """测试降级逻辑:模拟模型超时"""
    def __init__(self, fail: bool = False):
        self.fail = fail

    def before_model(self, messages):
        if self.fail:
            raise TimeoutError("模拟模型超时")
        return messages

LangChain 工具调用拦截

工具调用拦截在工具执行前后介入：参数校验、结果改写、失败重试、权限检查。

1. 基础拦截

class ToolAuditMiddleware(Middleware):
    def before_tool(self, tool_call: dict) -> dict:
        print(f"[工具] {tool_call['name']} args={tool_call['args']}")
        return tool_call

    def after_tool(self, result: str, tool_call: dict) -> str:
        print(f"[工具] 完成, 结果长度: {len(result)}")
        return result

2. 参数白名单校验

class SafeToolMiddleware(Middleware):
    """拦截危险工具调用"""
    DANGEROUS = {"delete_db", "send_email", "rm_file"}

    def before_tool(self, tool_call: dict) -> dict:
        if tool_call["name"] in self.DANGEROUS:
            # 直接抛错,阻止执行
            raise PermissionError(
                f"工具 {tool_call['name']} 被安全策略拦截"
            )
        return tool_call

3. 结果增强

class EnrichToolMiddleware(Middleware):
    """给工具结果附加来源信息"""
    def after_tool(self, result: str, tool_call: dict) -> str:
        return f"{result}
[数据来源: {tool_call['name']} @ {__import__('datetime').datetime.now():%H:%M}]"

4. 失败自动重试

import time

class RetryMiddleware(Middleware):
    def __init__(self, retries: int = 2):
        self.retries = retries

    def before_tool(self, tool_call: dict) -> dict:
        # 由于 before/after 是成对调用的,
        # 可以在 after 里检测失败并重试(简化示意)
        return tool_call

    def after_tool(self, result: str, tool_call: dict) -> str:
        if result.startswith("ERROR:") and self.retries > 0:
            self.retries -= 1
            time.sleep(1)
            # 重试(实际场景需重新执行工具)
            return f"{result}
[重试中...]"
        return result

LangChain @before_agent

@before_agent 是 Agent 的全局前置钩子：在整轮 Agent 执行开始之前运行一次（区别于每次模型调用前）。

基本用法

from langchain.agents.middleware import Middleware

class BeforeAgentMiddleware(Middleware):
    """Agent 开始前执行"""
    def before_agent(self, messages):
        print("=== Agent 开始 ===")
        print(f"用户消息: {messages[-1].content[:80]}")
        return messages

agent = create_agent(model, tools, middleware=[BeforeAgentMiddleware()])

典型场景

class GuardMiddleware(Middleware):
    """Agent 前置检查与初始化"""
    def before_agent(self, messages):
        # 1. 用户输入检查
        text = messages[-1].content
        if len(text) > 500:
            raise ValueError("输入过长")

        # 2. 初始化会话上下文
        # 3. 记录开始时间
        self.start_time = __import__("time").time()
        return messages

    def after_agent(self, result):
        # Agent 结束后执行
        elapsed = __import__("time").time() - self.start_time
        print(f"=== Agent 结束, 耗时 {elapsed:.2f}s ===")
        return result

拦截非法输入

class TopicGuardMiddleware(Middleware):
    """只允许特定主题"""
    ALLOWED = {"技术", "学习", "生活"}

    def before_agent(self, messages):
        text = messages[-1].content
        # 简单关键词判断
        if "违法" in text or "攻击" in text:
            raise ValueError("内容不在允许范围")
        return messages

与 before_model 的区别

钩子	时机	次数
before_agent	整个 Agent 开始前	每轮 invoke 一次
before_model	每次模型调用前	Agent 循环中多次
after_agent	整个 Agent 结束后	每轮 invoke 一次

LangChain 对话记忆

对话记忆（Memory）让 Agent 记住多轮上下文。核心思路：把历史消息保存起来，新请求时一起传给模型。

1. 手动维护消息历史

# 最简单:客户端保存所有消息,完整传给模型
history = []

def chat(user_text: str):
    global history
    history.append(("user", user_text))
    result = agent.invoke({"messages": history})
    reply = result["messages"][-1].content
    history.append(("ai", reply))
    return reply

print(chat("我叫张三"))
print(chat("我叫什么名字?"))   # 能记住 → 张三

2. 限制长度(窗口)

# 只保留最近 N 条,防止上下文超长
MAX_MESSAGES = 20

def trim_history(h):
    return h[-MAX_MESSAGES:]

def chat(user_text: str):
    global history
    history = trim_history(history + [("user", user_text)])
    result = agent.invoke({"messages": history})
    history.append(("ai", result["messages"][-1].content))
    return result["messages"][-1].content

3. 摘要压缩记忆

# 超长对话:把早期对话压缩成摘要
def summarize_old(history):
    if len(history) > 30:
        old = history[:-20]   # 取最早的部分
        recent = history[-20:]
        summary_prompt = f"把以下对话压缩成摘要:
{old}"
        summary = model.invoke(summary_prompt).content
        return [("system", f"对话摘要: {summary}")] + recent
    return history

4. 会话隔离

# 多用户并发:按 session_id 存独立历史
sessions: dict[str, list] = {}

def chat(session_id: str, user_text: str):
    hist = sessions.setdefault(session_id, [])
    hist.append(("user", user_text))
    result = agent.invoke({"messages": hist})
    hist.append(("ai", result["messages"][-1].content))
    # 限制长度
    sessions[session_id] = hist[-20:]
    return result["messages"][-1].content

LangChain 跨会话存储

跨会话存储把对话历史持久化到数据库/Redis，服务重启、多实例部署都不丢记忆。

1. 存储抽象

import json, sqlite3

class MessageStore:
    """简单的消息存储(SQLite 示例)"""
    def __init__(self, db="chat.db"):
        self.conn = sqlite3.connect(db)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS sessions (
                session_id TEXT PRIMARY KEY,
                messages TEXT
            )
        """)

    def load(self, session_id: str) -> list:
        row = self.conn.execute(
            "SELECT messages FROM sessions WHERE session_id=?",
            (session_id,)
        ).fetchone()
        return json.loads(row[0]) if row else []

    def save(self, session_id: str, messages: list):
        self.conn.execute(
            "INSERT OR REPLACE INTO sessions VALUES (?, ?)",
            (session_id, json.dumps(messages, ensure_ascii=False))
        )
        self.conn.commit()

store = MessageStore()

2. 使用持久化记忆

def chat(session_id: str, user_text: str):
    history = store.load(session_id)
    history.append(("user", user_text))
    result = agent.invoke({"messages": history})
    history.append(("ai", result["messages"][-1].content))
    store.save(session_id, history[-30:])   # 保留最近30条
    return result["messages"][-1].content

# 服务重启后:
print(chat("s1", "我叫李四"))
print(chat("s1", "我叫什么?"))   # 仍能记住

3. Redis 版本

import redis, json

r = redis.Redis(host="localhost", port=6379, db=0)

def load_messages(sid):
    data = r.get(f"chat:{sid}")
    return json.loads(data) if data else []

def save_messages(sid, msgs):
    r.setex(f"chat:{sid}", 3600 * 24 * 7,   # 7天过期
            json.dumps(msgs, ensure_ascii=False))

4. 关键设计

要点	说明
TTL 过期	会话太久自动清理
长度限制	存最近 N 条或摘要
原子性	读写用事务/原子命令
隔离	session_id 命名空间隔离

LangChain 人工介入

人工介入（Human-in-the-loop）让关键步骤需要人工确认才继续：危险操作确认、结果审核、双人复核。

1. 用中间件实现确认

class HumanApprovalMiddleware(Middleware):
    """危险工具执行前要求人工确认"""
    DANGEROUS = {"send_email", "delete_record", "pay_order"}

    def before_tool(self, tool_call: dict) -> dict:
        if tool_call["name"] in self.DANGEROUS:
            print(f"
⚠️ 请求确认: 调用 {tool_call['name']}")
            print(f"参数: {tool_call['args']}")
            ok = input("确认执行? (y/n): ").strip().lower()
            if ok != "y":
                raise PermissionError("用户取消操作")
        return tool_call

2. 手动触发 Agent(两步流程)

# 第一步:让模型分析并建议
plan = model.invoke("分析这个订单并给出处理建议")
print("模型建议:", plan.content)

# 人工确认后执行
ok = input("是否执行? (y/n): ")
if ok == "y":
    result = agent.invoke({"messages": [("user", "按建议执行")]})
    print("执行结果:", result["messages"][-1].content)

3. 异步人工审批(Web 场景)

# 思路:Agent 停在"等待审批"节点
# 1. Agent 生成操作请求 → 存 pending 表
# 2. 前端展示审批卡片
# 3. 审批通过 → 继续执行

pending_ops = {}   # 简化:内存存储

def request_approval(op_type: str, payload: dict) -> str:
    op_id = f"op_{len(pending_ops) + 1}"
    pending_ops[op_id] = {"type": op_type, "payload": payload, "status": "pending"}
    return op_id

def approve(op_id: str, decision: bool):
    op = pending_ops.get(op_id)
    if op:
        op["status"] = "approved" if decision else "rejected"
    return op

LangChain 多 Agent

多 Agent（Multi-Agent）让多个各司其职的智能体协作完成任务：一个主导、多个分工，或流水线接力。

1. 分工模式(角色分离)

from langchain.agents import create_agent

# 研究员 Agent
researcher = create_agent(
    model, [search_tool],
    system_prompt="你是研究员,负责搜集信息并输出要点。",
)

# 写作者 Agent
writer = create_agent(
    model, [],
    system_prompt="你是文案写手,根据要点写出文章。",
)

# 主编 Agent(汇总)
editor = create_agent(
    model, [],
    system_prompt="你是主编,负责审校与润色。",
)

2. 流水线接力

def research_to_write(topic: str) -> str:
    # 第一步:研究
    research = researcher.invoke({
        "messages": [("user", f"研究主题:{topic},输出3个要点")]
    })
    points = research["messages"][-1].content

    # 第二步:写作
    draft = writer.invoke({
        "messages": [("user", f"根据以下要点写 200 字短文:
{points}")]
    })
    article = draft["messages"][-1].content

    # 第三步:润色
    final = editor.invoke({
        "messages": [("user", f"润色这篇文章:
{article}")]
    })
    return final["messages"][-1].content

3. 团队协作(一个 Agent 调另一个)

@tool
def consult_writer(task: str) -> str:
    """把写作任务交给专业写手 Agent"""
    result = writer.invoke({"messages": [("user", task)]})
    return result["messages"][-1].content

# 主导 Agent 可调用"写手"工具
lead = create_agent(model, [consult_writer, search_tool])

4. 多 Agent 设计原则

原则	说明
职责单一	每个 Agent 只做一件事
明确交接	上一个输出 = 下一个输入,格式要规范
控制成本	Agent 越多调用越多,权衡收益
可观测	每个 Agent 的结果都记录可查

LangChain RAG

RAG（Retrieval-Augmented Generation，检索增强生成）：先从知识库检索相关内容，再让模型基于检索结果回答——解决「模型不知道你的私有数据」的问题。

RAG 流程

1. 离线:文档 → 切分 → 向量化 → 存入向量库
2. 在线:用户提问 → 向量化 → 检索最相关片段
3. 生成:问题 + 检索片段 → 模型 → 回答(带来源)

最小 RAG 实现

# 安装: pip install langchain-text-splitters chromadb langchain-community
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

# 1. 文档
docs = [
    "公司年假政策:工作满一年享5天年假。",
    "报销流程:先填单,经理审批,财务打款。",
    "远程办公:每周三可在家办公。",
]

# 2. 向量化+存库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = InMemoryVectorStore.from_texts(docs, embeddings)

# 3. 检索
query = "年假怎么算?"
hits = store.similarity_search(query, k=2)
context = "
".join(d.page_content for d in hits)

# 4. 生成
resp = model.invoke(
    f"基于以下资料回答问题,资料:
{context}

问题:{query}"
)
print(resp.content)

为什么用 RAG

方案	优点	缺点
直接问模型	简单	不知道私有数据,可能胡编
微调模型	掌握知识	贵、慢、更新难
RAG	便宜、实时、可溯源	需搭建检索链路

LangChain 文档加载

文档加载（Document Loading）把各种格式的文件（PDF、Word、网页、Markdown）读入 LangChain 文档对象，是 RAG 的第一步。

1. 加载各种格式

# 安装: pip install pypdf langchain-community beautifulsoup4

# 文本
from langchain_core.document_loaders import TextLoader
docs = TextLoader("notes.txt", encoding="utf-8").load()

# Markdown
from langchain_community.document_loaders import UnstructuredMarkdownLoader
docs = UnstructuredMarkdownLoader("readme.md").load()

# PDF
from langchain_community.document_loaders import PyPDFLoader
docs = PyPDFLoader("report.pdf").load()

# 网页
from langchain_community.document_loaders import WebBaseLoader
docs = WebBaseLoader("https://example.com/article").load()

# CSV
from langchain_community.document_loaders import CSVLoader
docs = CSVLoader("data.csv").load()

2. 文档对象结构

doc = docs[0]
doc.page_content    # 正文文本
doc.metadata        # 元数据(来源、页码等)
# {'source': 'report.pdf', 'page': 1, ...}

# 批量查看
for d in docs[:3]:
    print(f"[{d.metadata.get('source')}] {d.page_content[:50]}...")

3. 手动创建文档

from langchain_core.documents import Document

doc = Document(
    page_content="这是一段知识内容……",
    metadata={"source": "manual", "id": 1},
)

4. 递归加载目录

import os

def load_dir(path: str) -> list:
    all_docs = []
    for root, _, files in os.walk(path):
        for f in files:
            if f.endswith((".md", ".txt")):
                full = os.path.join(root, f)
                all_docs.extend(TextLoader(full, encoding="utf-8").load())
    return all_docs

LangChain RAG Agent

RAG Agent = Agent + 检索工具：让模型自主决定何时检索、检索什么，而不是每轮强制检索。

1. 把检索封装成工具

from langchain_core.tools import tool

# 提前建好向量库
store = InMemoryVectorStore.from_texts(docs, embeddings)

@tool
def search_kb(query: str) -> str:
    """从公司知识库检索信息。查询员工政策、流程时使用。"""
    hits = store.similarity_search(query, k=3)
    return "

".join(
        f"[来源{d.metadata.get('source','?')}] {d.page_content}"
        for d in hits
    )

2. 创建 RAG Agent

agent = create_agent(
    model,
    [search_kb],
    system_prompt=(
        "你是公司知识助手。回答问题前先检索知识库,"
        "答案必须基于检索到的资料,资料不足时说明。"
    ),
)

# 普通闲聊不检索
r1 = agent.invoke({"messages": [("user", "你好")]})

# 政策问题自动检索
r2 = agent.invoke({"messages": [("user", "报销流程是什么?")]})
print(r2["messages"][-1].content)

3. 多知识源

@tool
def search_hr(query: str) -> str:
    """检索 HR 政策(请假、考勤、福利)"""

@tool
def search_tech(query: str) -> str:
    """检索技术文档(API、部署、故障排查)"""

agent = create_agent(model, [search_hr, search_tech])

4. 与普通 RAG 对比

维度	普通 RAG	RAG Agent
检索时机	固定:每轮都检索	模型决定是否检索
多源支持	需手动合并	多工具自动路由
追问能力	弱	可多轮检索追问
复杂度	低	中

LangChain 智能客服

综合运用 RAG + Agent + 记忆，构建一个企业智能客服：查政策、查订单、人工转接。

1. 工具集

from langchain_core.tools import tool

@tool
def search_kb(query: str) -> str:
    """检索公司知识库(政策、流程、常见问题)"""
    # ... 向量检索实现

@tool
def query_order(order_id: str) -> str:
    """查询订单状态与物流"""
    # ... 调用订单系统 API
    return f"订单 {order_id}: 已发货,预计 3 天到达"

@tool
def transfer_to_human() -> str:
    """转接人工客服(当用户情绪激动或问题复杂时)"""
    # ... 创建工单/推送客服队列
    return "已为您转接人工客服,工单号: CS-2026-0817"

tools = [search_kb, query_order, transfer_to_human]

2. 客服 Agent

agent = create_agent(
    model,
    tools,
    system_prompt=(
        "你是XX商城客服小助手。规则:
"
        "1. 政策类问题必须检索知识库后回答
"
        "2. 查订单需要订单号,没有就问用户要
"
        "3. 用户投诉/情绪激动时调用转人工
"
        "4. 不知道就转人工,不要编造"
    ),
)

3. 会话管理

def customer_service(session_id: str, user_input: str) -> str:
    history = store.load(session_id)          # 取历史
    history.append(("user", user_input))
    result = agent.invoke({"messages": history})
    reply = result["messages"][-1].content
    history.append(("ai", reply))
    store.save(session_id, history[-30:])     # 存最近30条
    return reply

# 模拟对话
print(customer_service("u1", "我的订单 12345 到哪了?"))
print(customer_service("u1", "怎么申请退款?"))   # 记住上下文

4. 上线要点

要点	说明
知识库维护	政策更新后重新向量化
转人工兜底	明确触发条件与流程
敏感信息	订单信息需身份校验
数据记录	对话日志留存审计

LangChain 个人知识库

构建个人知识库问答：把你自己的笔记、文档、收藏变成可对话的知识库。这是 RAG 最经典的个人应用。

1. 加载并切分文档

# 安装: pip install langchain-text-splitters
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每片约500字符
    chunk_overlap=50,      # 片间重叠50字符(保持上下文)
)

# 加载所有笔记
raw_docs = load_dir("~/notes/")   # 前面定义的目录加载函数
chunks = splitter.split_documents(raw_docs)
print(f"共 {len(chunks)} 个片段")

2. 向量化并存储

from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = InMemoryVectorStore.from_documents(chunks, embeddings)

# 生产环境可用 Chroma / FAISS 持久化:
# from langchain_chroma import Chroma
# store = Chroma.from_documents(chunks, embeddings, persist_directory="./kb")

3. 检索问答

@tool
def search_my_notes(query: str) -> str:
    """检索我的个人笔记。问我的笔记内容时使用。"""
    hits = store.similarity_search(query, k=4)
    if not hits:
        return "笔记中没有相关内容"
    return "

".join(
        f"[片段{i+1}] {d.page_content}" for i, d in enumerate(hits)
    )

agent = create_agent(
    model, [search_my_notes],
    system_prompt=(
        "你是我的个人知识助手,根据笔记内容回答问题;"
        "笔记没有的内容要明确说明。"
    ),
)

print(agent.invoke({"messages": [("user", "我的笔记里关于 FastAPI 写了什么?")]})
      ["messages"][-1].content)

4. 维护知识库

# 新增笔记后重新构建(或增量):
# 1. 新笔记 → 切分 → 向量化 → 追加到库
# 2. 删除笔记 → 按 metadata 删除对应向量

def rebuild():
    """一键重建知识库"""
    raw = load_dir("~/notes/")
    chunks = splitter.split_documents(raw)
    return InMemoryVectorStore.from_documents(chunks, embeddings)

LangChain 个人助手

把之前学的全串起来，做一个真正的个人助手：日历、天气、计算、知识库，一个 Agent 全搞定。

1. 工具全家桶

from langchain_core.tools import tool
from datetime import datetime

@tool
def get_now() -> str:
    """获取当前日期时间"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate(expr: str) -> str:
    """数学计算,传入表达式如 '3.14*2"""'
    try:
        return str(eval(expr, {"__builtins__": {}}))
    except Exception as e:
        return f"计算错误: {e}"

@tool
def get_weather(city: str) -> str:
    """查询天气"""
    # 实际接天气 API
    return f"{city}: 多云 26°C, 东南风3级"

@tool
def search_kb(query: str) -> str:
    """检索我的知识库"""
    hits = store.similarity_search(query, k=3)
    return "
".join(d.page_content for d in hits)

2. 助手 Agent

assistant = create_agent(
    model,
    [get_now, calculate, get_weather, search_kb],
    system_prompt=(
        "你是我的个人AI助手。"
        "时间/计算/天气/知识库问题分别用对应工具,"
        "能一步算完的不要问用户。"
    ),
)

def ask(question: str):
    result = assistant.invoke({"messages": [("user", question)]})
    return result["messages"][-1].content

print(ask("现在几点?"))
print(ask("帮我算 2^10"))
print(ask("上海天气?"))
print(ask("我笔记里怎么部署 FastAPI?"))

3. 带记忆的助手

def chat(session_id, text):
    history = store.load(session_id)
    history.append(("user", text))
    result = assistant.invoke({"messages": history})
    history.append(("ai", result["messages"][-1].content))
    store.save(session_id, history[-30:])
    return result["messages"][-1].content

4. 扩展方向

能力	工具示例
邮件	send_email(to, subject, body)
日历	create_event(title, time)
任务	add_todo(item, due)
搜索	web_search(keyword)

LangChain LangSmith

LangSmith 是 LangChain 官方的可观测性平台：追踪每次运行、查看 token 成本、调试 Agent 行为、评估输出质量。

1. 接入 LangSmith

pip install langsmith
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="lsv2_xxx"     # 从 langsmith.com 获取
export LANGCHAIN_PROJECT="my-agent"     # 项目名(可选)

2. 自动追踪

# 配置后,所有 LangChain 调用自动上报
agent = create_agent(model, tools)
result = agent.invoke({"messages": [("user", "查天气")]})

# 无需额外代码,在 LangSmith 控制台可看到:
# - 完整调用链(模型/工具/中间件)
# - 输入输出详情
# - token 用量与成本
# - 延迟分析

3. 手动打点

from langsmith import traceable

@traceable(name="my_custom_step")
def preprocess(text: str) -> str:
    return text.strip()   # 自定义步骤也会被追踪

4. 评估数据集

# LangSmith 支持:创建数据集 → 批量跑 → 评分
# 评估指标示例:
# - 相关性(人工/LLM 评分)
# - 延迟与成本
# - 错误率

# 用 LangSmith 评估器
from langsmith.evaluation import evaluate

def my_evaluator(run, example):
    # run: 一次运行; example: 数据集样例
    score = 1 if "答案" in run.outputs["output"] else 0
    return {"key": "has_answer", "score": score}

5. 本地替代方案

# 不想上云,自己记录关键信息
def log_run(question, messages, cost_tokens):
    print(json.dumps({
        "q": question,
        "tokens": cost_tokens,
        "steps": len(messages),
    }, ensure_ascii=False, indent=2))

LangChain 错误处理

生产级 Agent 必须优雅处理错误：模型限流、工具失败、超时、解析失败。

1. 基础 try/except

try:
    result = agent.invoke({"messages": [("user", question)]})
    print(result["messages"][-1].content)
except Exception as e:
    print(f"调用失败: {e}")
    print("请稍后重试或换个问法")

2. 区分错误类型

from langchain_core.exceptions import (
    OutputParserException,
)

try:
    result = agent.invoke({"messages": [("user", question)]})
except OutputParserException as e:
    print("输出解析失败,模型可能没按格式返回")
except TimeoutError:
    print("请求超时,请重试")
except Exception as e:
    # 限流、网络等
    print(f"未知错误: {type(e).__name__}: {e}")

3. 重试机制

import time

def invoke_with_retry(agent, question, retries=3, delay=2):
    for attempt in range(retries):
        try:
            return agent.invoke({"messages": [("user", question)]})
        except Exception as e:
            print(f"第{attempt+1}次失败: {e}")
            if attempt < retries - 1:
                time.sleep(delay * (attempt + 1))   # 退避
    raise RuntimeError("多次重试仍失败")

result = invoke_with_retry(agent, "查天气")

4. 模型层错误处理

# 模型初始化时配置重试
model = init_chat_model(
    "deepseek-chat",
    model_provider="deepseek",
    max_retries=3,          # 自动重试
    timeout=60,             # 超时
)

# 限流降级:换模型
def get_model_with_fallback():
    try:
        return init_chat_model("deepseek-chat", model_provider="deepseek")
    except Exception:
        return init_chat_model("qwen-plus", model_provider="alibabacloud")

5. 用户友好错误提示

def safe_chat(text):
    try:
        result = agent.invoke({"messages": [("user", text)]})
        return result["messages"][-1].content
    except TimeoutError:
        return "😅 响应超时了,请再试一次"
    except Exception as e:
        print("详细日志:", e)   # 记录到日志系统
        return "🙏 出了点问题,请稍后再试"

LangChain Chat Model API

Chat Model API 速查手册：初始化、调用、参数、方法一览。

初始化

# 通用
from langchain.chat_models import init_chat_model
model = init_chat_model(
    "deepseek-chat",
    model_provider="deepseek",
    temperature=0.7,
    max_tokens=2048,
)

# 具体 SDK
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)

调用方法

resp = model.invoke("你好")            # 同步
resp = asyncio.run(model.ainvoke("你好"))  # 异步
for c in model.stream("写诗"): ...     # 流式
resps = model.batch([...])             # 批量

绑定

m = model.bind(temperature=0)             # 绑定参数
m = model.bind_tools([tool])              # 绑定工具
m = model.bind(response_format={"type":"json_object"})  # JSON模式
m = model.with_structured_output(Schema)  # 结构化输出

响应对象

resp.content              # 文本
resp.tool_calls           # 工具调用列表
resp.usage_metadata       # token 用量
resp.response_metadata    # 元数据(耗时等)
resp.role                 # 'assistant'

常见错误

错误	原因	解决
AuthenticationError	API Key 错/缺失	检查环境变量
RateLimitError	触发限流	加延迟/重试
BadRequestError	模型名错/参数非法	检查模型名
TimeoutError	请求超时	加大 timeout

LangChain Agent API

Agent API 速查：创建、调用、流式、事件。

创建 Agent

from langchain.agents import create_agent

agent = create_agent(model, tools)
agent = create_agent(model, tools, system_prompt="...")
agent = create_agent(model, tools, middleware=[mw])

调用

# 标准调用
result = agent.invoke({"messages": [("user", "问题")]})

# 异步
result = asyncio.run(agent.ainvoke({"messages": [...]}))

# 流式事件
async for event in agent.astream_events({"messages": [...]}, version="v2"):
    print(event["event"], event.get("name", ""))

# 多轮:带上历史消息
result = agent.invoke({"messages": history + [("user", "追问")]})

结果结构

result["messages"]           # 完整消息链
result["messages"][-1]       # 最终回答
result["messages"][-1].content   # 答案文本

# 查看工具调用过程
for m in result["messages"]:
    if m.tool_calls:
        print("调用了", m.tool_calls[0]["name"])

方法速查

方法	说明
invoke(input)	同步调用
ainvoke(input)	异步调用
stream(input)	流式
astream_events	事件流(模型/工具/消息)
get_graph()	获取底层 LangGraph 图

LangChain Messages API

Messages API 速查：消息类型、创建、操作。

消息类型

from langchain_core.messages import (
    SystemMessage,    # 系统
    HumanMessage,     # 用户
    AIMessage,        # 助手
    ToolMessage,      # 工具结果
    AIMessageChunk,   # 流式分块
)

SystemMessage("你是助手")
HumanMessage("你好")
AIMessage("你好!")
ToolMessage("结果内容", tool_call_id="call_1")

元组快捷写法

# ("role", "content") 元组会被自动转换
messages = [
    ("system", "系统设定"),
    ("human", "用户问题"),
    ("ai", "之前回答"),
    ("user", "新问题"),     # "user" 等价 "human"
]
model.invoke(messages)

常用属性

msg.content        # 文本内容
msg.role           # 角色名
msg.type           # 类型
msg.tool_calls     # (AI) 工具调用
msg.tool_call_id   # (Tool) 关联的工具调用ID

# 流式合并
full = AIMessageChunk(content="")
for c in model.stream("hi"):
    full += c

工具类

from langchain_core.messages import (
    AnyMessage,          # 任意消息类型
    is_ai_message,
    is_human_message,
    trim_messages,       # 裁剪历史
)

# 裁剪:保留最近 N 条 + 始终保留 System
from langchain_core.messages import trim_messages
trimmed = trim_messages(
    messages,
    max_tokens=1000,
    strategy="last",
    token_counter=len,      # 简单按条数
    start_on="human",
    include_system=True,    # 保留系统消息
)

LangChain Tools API

Tools API 速查：创建、配置、调用。

创建工具

from langchain_core.tools import tool

@tool
def my_tool(arg1: str, arg2: int = 1) -> str:
    """工具描述(给模型看)"""
    return "结果"

# 自定义名称
@tool("tool_name")
def f(x: str) -> str: ...

# 容错
@tool(handle_tool_error=True)
def safe_tool(x: int) -> int: ...

# 异步
@tool
async def async_tool(x: str) -> str: ...

工具属性

tool.name          # 名称
tool.description   # 描述
tool.args_schema   # 参数 Schema
tool.args          # 参数 JSON Schema 字典
tool.invoke({"x": "值"})   # 调用
tool.invoke("字符串")       # 单参数简写(若只有一个必填参数)

结构化工具

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class Args(BaseModel):
    query: str = Field(description="关键词")
    limit: int = Field(5, description="条数")

tool = StructuredTool.from_function(
    func=lambda query, limit: f"{query} x {limit}",
    name="search",
    description="搜索",
    args_schema=Args,
)

与模型/Agent 结合

# 绑定到模型
m = model.bind_tools([my_tool])
resp = m.invoke("...")
for call in resp.tool_calls:
    result = my_tool.invoke(call["args"])

# 加入 Agent
agent = create_agent(model, [my_tool, async_tool])

LangChain Middleware API

Middleware API 速查：中间件的生命周期方法与参数。

生命周期钩子

from langchain.agents.middleware import Middleware

class MyMiddleware(Middleware):
    # Agent 级
    def before_agent(self, messages): ...
    def after_agent(self, result): ...

    # 模型级(每次调用)
    def before_model(self, messages): ...
    def after_model(self, response): ...

    # 工具级(每次调用)
    def before_tool(self, tool_call): ...
    def after_tool(self, result, tool_call): ...

注册

agent = create_agent(model, tools, middleware=[
    LogMiddleware(),
    RateLimitMiddleware(),
])

修改输入/输出

class ModifyMiddleware(Middleware):
    def before_model(self, messages):
        # 返回修改后的消息
        return messages + [("system", "额外上下文")]

    def after_model(self, response):
        # 返回修改后的响应
        response.content += "
(由中间件追加)"
        return response

短路/中断

class ShortCircuit(Middleware):
    def before_model(self, messages):
        if "直接回答" in messages[-1].content:
            # 返回消息列表 = 跳过模型,直接用这些消息作为"模型输出"
            from langchain_core.messages import AIMessage
            return [AIMessage(content="(被中间件拦截)")]
        return messages

    def before_tool(self, tool_call):
        if tool_call["name"] in {"dangerous_op"}:
            raise PermissionError("禁止调用")   # 抛错中断

LangChain 配置与错误类

配置与错误类速查：模型配置、回调配置、异常体系。

环境变量配置

# API Keys
DEEPSEEK_API_KEY=sk-...
OPENAI_API_KEY=sk-...
DASHSCOPE_API_KEY=sk-...

# LangSmith
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=lsv2_...

# 代理(可选)
HTTP_PROXY=http://127.0.0.1:7890
HTTPS_PROXY=http://127.0.0.1:7890

运行时配置(config)

# 每次调用可传 config
result = agent.invoke(
    {"messages": [("user", "hi")]},
    config={
        "callbacks": [PrintToken()],   # 回调
        "tags": ["production"],        # 标签
        "metadata": {"user_id": 123},  # 元数据
        "run_name": "my-run",
    },
)

# 限流/并发
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(max_concurrency=5)

异常体系

from langchain_core.exceptions import (
    OutputParserException,    # 输出解析失败
)

# 模型层(来自 SDK):
# AuthenticationError  认证失败
# RateLimitError      限流
# BadRequestError     请求非法
# InternalServerError 服务端错误
# TimeoutError        超时

# 统一兜底
def safe_invoke(chain, **kwargs):
    try:
        return chain.invoke(**kwargs)
    except OutputParserException:
        return None, "输出格式异常"
    except Exception as e:
        return None, f"{type(e).__name__}: {e}"

模型厂商错误对照

厂商	包	典型异常
OpenAI 系	openai	AuthenticationError 等
DeepSeek	openai 兼容	同上
阿里百炼	dashscope	InvalidApiKey 等

LangChain

LangChain 教程总结：知识图谱、实战路径、下一步学习方向。

知识图谱

LangChain
├── 模型层: Chat Models(统一调用/流式/结构化)
├── 提示层: Prompt Template(角色+任务+约束+示例)
├── 工具层: @tool / StructuredTool(让模型调用函数)
├── 智能体: create_agent(模型+工具+循环)
│   ├── 中间件: before/after model & tool
│   ├── 记忆: 多轮/跨会话/摘要压缩
│   └── 多Agent: 分工/流水线/协作
├── RAG: 文档加载→切分→向量化→检索→生成
└── 可观测: LangSmith(追踪/评估)

实战路径建议

阶段1: 环境搭建 + 模型调用 + 提示词       (半天)
阶段2: @tool + create_agent + 中间件     (1天)
阶段3: RAG 个人知识库                    (1天)
阶段4: 完整应用: 智能客服/个人助手        (1-2天)
阶段5: LangSmith 调优 + 生产部署         (按需)

常见问题排查

问题	原因	解决
模型不调用工具	工具描述不清/模型不支持	优化 description;换支持工具调用的模型
Agent 死循环	工具结果不可用	限制步数;让工具返回明确信息
回答乱编	没有检索或提示约束弱	加 RAG;提示词强调"不知道就说不知道"
上下文超长	历史累积	trim_messages / 摘要压缩

继续学习

• LangGraph: 复杂状态机、条件分支、人工介入的高级实现
• LangSmith: 评估集、回归测试、生产监控
• 向量库: Chroma / FAISS / Milvus 生产级检索
• 多模态: 图片/语音/视频与 Agent 结合