LangChain 实战：从零搭建你的第一个 AI Agent 应用

2026-05-29 Agent 0 次阅读

从 Chain 到 Agent，从 Prompt 到 Tool，从 Memory 到 RAG——手把手带你掌握 LangChain 核心开发范式，打造能自主思考、会调用工具、有记忆能力的智能应用。

为什么你需要 LangChain
快速上手：5 分钟跑通第一个 Chain
Prompt 工程：模板化你的 AI 指令
模型抽象层：灵活切换 LLM 提供商
Chain 组合技：串联多个 AI 调用
Tool 与 Agent：让 AI 学会调用外部工具
Memory 模块：带记忆的对话系统
RAG 实战：构建知识库问答系统
生产环境部署要点
性能优化与成本控制
常见问题解答
总结与展望

一、为什么你需要 LangChain

1.1 大模型时代的开发痛点

2024-2025 年，大语言模型能力飞速进化，但将 LLM 集成到实际应用中，开发者面临一系列共性问题。

想象你要开发一个「智能客服系统」——用户问「我的订单什么时候发货？」，AI 需要：理解用户意图 → 查询订单数据库 → 获取物流信息 → 用自然语言回复。这个过程涉及：

API 调用的样板代码：HTTP 请求、鉴权、超时重试、流式输出
Prompt 的碎片化拼接：模板参数、条件分支、多轮对话上下文的字符串拼接地狱
工具调用的解析博弈：让 AI 输出 JSON、解析动作、执行函数、回传结果
多模型适配噩梦：OpenAI、Anthropic、Google、开源模型各有不同 API 格式
对话状态的维护：每次调用无状态，需自己管理对话历史和上下文窗口

这些痛点不是个例，而是整个 LLM 应用开发领域的通用挑战。LangChain 正是为解决这些问题而生的。

1.2 LangChain 是什么：乐高积木式开发

LangChain 是一个开源的 LLM 应用开发框架，GitHub Star 数超过 95k，是目前最受欢迎的 AI 应用开发工具之一。它的核心设计理念是组件化——把 LLM 开发中的常见模式抽象为可复用的「组件」，像搭乐高积木一样组合它们。

LangChain 有两大核心编程范式：

范式	说明	控制流	适用场景	类比
Chain（链）	预定义的调用序列，A→B→C 依次执行	确定性	翻译、摘要、固定流程	工厂流水线
Agent（代理）	AI 自主决定调用哪些工具、按什么顺序	非确定性	客服、搜索、复杂任务	自动驾驶

简单来说：Chain 是「你告诉 AI 每一步做什么」，Agent 是「你给 AI 一堆工具，让它自己决定怎么做」。

1.3 生态全景图

LangChain 已经发展出一个完整的生态系统：

LangChain 生态地图：
┌────────────────────────────────────────────────────────────┐
│                      LangChain Core                        │  ← Python/JS 核心库
│     (Prompt · Model · Chain · Agent · Memory · RAG)        │
├──────────────┬──────────────┬──────────────┬───────────────┤
│  LangSmith   │  LangServe   │  LangGraph   │ LangChain Hub │
│  (调试/监控)  │  (API 部署)  │  (状态机编排)  │  (Prompt 市场) │
├──────────────┴──────────────┴──────────────┴───────────────┤
│                LangChain Community (70+ 集成)               │
│   OpenAI · Anthropic · Google · Ollama · Pinecone · Chroma │
│   Qdrant · Weaviate · FAISS · Redis · PostgreSQL · Neo4j  │
└────────────────────────────────────────────────────────────┘

读完本文你将掌握：

LangChain 六大核心组件（Prompt / Model / Chain / Agent / Memory / RAG）的实战用法
从简单 Chain 到复杂 Agent 的完整开发路径，每一步都有可运行代码
一个完整的 RAG 知识库问答系统（含文档加载、分割、向量化、检索、生成全流程）
生产环境部署的注意事项、性能优化、成本控制策略
常见踩坑点和避坑指南

二、快速上手：5 分钟跑通第一个 Chain

2.1 环境搭建

开始之前，先搭建一个干净的环境。推荐使用虚拟环境避免依赖冲突：

# 创建虚拟环境
python -m venv langchain-env
source langchain-env/bin/activate  # Windows 用户: langchain-env\Scripts\activate

# 安装核心包（版本锁定避免兼容性问题）
pip install langchain==0.3.* langchain-openai==0.2.* python-dotenv

# 验证安装
python -c "import langchain; print(f'LangChain version: {langchain.__version__}')"

配置 API Key（推荐用 .env 文件管理敏感信息）：

echo 'OPENAI_API_KEY=sk-your-actual-key-here' > .env
echo 'OPENAI_BASE_URL=https://api.openai.com/v1' >> .env  # 可选：代理地址

⚠️ 安全提醒：永远不要把 .env 文件提交到 Git 仓库。在 .gitignore 中添加 .env。

2.2 Hello World：你的第一个 Chain

下面是最简单的 LangChain 应用——不到 15 行代码：

# hello_langchain.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

# 1. 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 2. 创建 Prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}专家，用{style}风格回答问题。回答要专业、准确。"),
    ("human", "{question}")
])

# 3. 构建 Chain（使用 LCEL 语法）=20
chain = prompt | llm | StrOutputParser()

# 4. 执行
response = chain.invoke({
    "role": "Python 编程",
    "style": "简洁、带代码示例",
    "question": "如何在 Python 中实现异步文件读取？"
})

print(response)

运行结果示例：

在 Python 中实现异步文件读取，推荐使用 `aiofiles` 库：

```python
import asyncio
import aiofiles

async def read_file_async(filepath: str) -> str:
    """异步读取文件全部内容"""
    async with aiofiles.open(filepath, mode='r', encoding='utf-8') as f:
        content = await f.read()
    return content

async def read_lines_async(filepath: str):
    """异步逐行读取"""
    async with aiofiles.open(filepath, mode='r') as f:
        async for line in f:
            yield line.strip()

# 使用方式
async def main():
    content = await read_file_async('example.txt')
    print(content)

asyncio.run(main())

相比传统的同步 I/O，异步方式在处理大量文件时性能提升显著（通常 3-5 倍）。


### 2.3 深入理解 LCEL 管道操作符 `|`

LCEL（LangChain Expression Language）是 LangChain 的核心语法。使用 `|` 操作符（类似 Unix 管道的概念）连接组件：

prompt | llm | output_parser ↓ ↓ ↓ 输入 AI 处理格式化输出


数据流向非常直观：上游组件的输出自动成为下游组件的输入。每个组件都实现了 `Runnable` 接口，支持三种调用方式：

| 调用方式 | 方法 | 适用场景 | 示例 |
| :--- | :--- | :--- | :--- |
| 同步单次 | `invoke()` | 简单对话、问答 | `chain.invoke({"topic": "Python"})` |
| 批量处理 | `batch()` | 批量翻译、分类 | `chain.batch([{"t": "A"}, {"t": "B"}])` |
| 流式输出 | `stream()` | 聊天、实时生成 | `for chunk in chain.stream(...)` |

> **设计哲学**：声明式写法让代码清晰、易于单元测试、便于可视化。同一个 Chain，既是可执行的逻辑，也是自文档化的流程图。

---

## 三、Prompt 工程：模板化你的 AI 指令

Prompt 是你与 LLM 对话的核心界面。精心设计的 Prompt 能让 AI 的输出质量从 60 分提升到 90 分。LangChain 提供了强大的 Prompt 模板系统。

### 3.1 ChatPromptTemplate：多角色结构化对话

最常用的模板——支持 system、human、ai 三种角色消息：

```python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

# 带历史记录的多轮对话模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是 EVA，一个专业的技术问答 AI 助手。回答风格：简洁、准确、友好。"),
    MessagesPlaceholder(variable_name="history"),  # =E5=AF=B9=E8=AF=9D=E5=8E=86=E5=8F=B2=E5=8D=A0=E4=BD=8D=E7=AC=A6
    ("human", "{input}")
])

# 模拟两轮对话
conversation_history = [
    HumanMessage(content="什么是 Docker？"),
    AIMessage(content="Docker 是一个开源的容器化平台，它能将应用及其依赖打包到一个轻量级、可移植的容器中。"),
    HumanMessage(content="那和虚拟机有什么区别？"),
]

result = chat_prompt.invoke({
    "history": conversation_history,
    "input": "和虚拟机有什么区别？"
})

# 查看格式化后的消息
for msg in result.to_messages():
    role = "System" if msg.type == "system" else "AI" if msg.type == "ai" else "Human"
    print(f"[{role}] {msg.content[:80]}...")

3.2 Few-Shot Prompting：用示例教 AI 输出格式

当需要 AI 以特定格式输出时（如 JSON、Markdown 表格、分类标签），少样本提示是最有效的技术：

from langchain_core.prompts import FewShotChatMessagePromptTemplate, ChatPromptTemplate

# 定义格式示例
examples = [
    {
        "input": "用户反馈：登录页面的「记住我」复选框勾选后关闭浏览器重新打开还是需要重新登录",
        "output": "BUG | P1 | 前端 | Cookie 持久化策略失效 | 需要立即修复"
    },
    {
        "input": "用户建议：在搜索结果页增加按价格区间筛选的功能",
        "output": "FEATURE | P3 | 前端 | 搜索体验优化 | 可排期到下个 Sprint"
    },
    {
        "input": "数据库查询：用户列表接口在大数据量下响应时间超过 5 秒",
        "output": "PERFORMANCE | P0 | 后端 | SQL 查询缺少索引 | 紧急优化"
    },
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

# 组装最终 Prompt
final_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的 Bug/需求分类助手。
根据用户描述，严格按照以下格式输出（不要输出其他内容）：
类型 | 优先级 | 模块 | 根因判断 | 处理建议

类型可选：BUG, FEATURE, PERFORMANCE, SECURITY
优先级可选：P0(紧急), P1(高), P2(中), P3(低)"""),
    few_shot_prompt,
    ("human", "{input}")
])

print(final_prompt.format(input="上传文件接口没有校验文件大小，用户上传了 500MB 的文件导致服务器 OOM"))

3.3 Pipeline Prompt：多步推理链

复杂任务可以拆分为多个 Prompt 步骤，每步聚焦一个子任务：

from langchain_core.prompts import PromptTemplate

# 步骤 1：问题分解
decompose_prompt = PromptTemplate.from_template(
    """你是一个资深的软件架构师。请将以下复杂需求拆解为 3-5 个可独立实现的子任务。
    
需求：{requirement}

请用列表格式输出每个子任务，格式：子任务名称 - 简要说明"""
)

# 步骤 2：方案设计
design_prompt = PromptTemplate.from_template(
    """基于以下子任务分解，为每个子任务设计具体的技术方案。

{decomposition}

对每个子任务，请说明：技术选型、关键实现难点、预估工作量（人天）"""
)

# 步骤 3：风险评估
risk_prompt = PromptTemplate.from_template(
    """基于以下技术方案，识别潜在风险并给出缓解措施。

{design}

用表格格式输出：风险描述 | 影响程度(高/中/低) | 缓解措施"""
)

# 串联执行
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
decomposition = (decompose_prompt | llm | StrOutputParser()).invoke({
    "requirement": "构建一个支持百万级用户的实时消息推送系统"
})
design = (design_prompt | llm | StrOutputParser()).invoke({
    "decomposition": decomposition
})
risks = (risk_prompt | llm | StrOutputParser()).invoke({
    "design": design
})

print("=== 问题分解 ===")
print(decomposition)
print("\n=== 技术方案 ===")
print(design)
print("\n=== 风险评估 ===")
print(risks)

3.4 Prompt 模板类型速查

模板类型	类名	适用场景	灵活性
简单文本	`PromptTemplate`	文本补全、代码生成	⭐⭐
多角色对话	`ChatPromptTemplate`	聊天机器人、客服	⭐⭐⭐⭐
少样本	`FewShotChatMessagePromptTemplate`	格式控制、分类	⭐⭐⭐⭐⭐
流水线	`PipelinePromptTemplate`	多步推理	⭐⭐⭐⭐
动态选择	`LengthBasedExampleSelector`	上下文窗口管理	⭐⭐⭐

四、模型抽象层：灵活切换 LLM 提供商

LangChain 最大的工程价值之一就是统一的模型接口。无论是 OpenAI GPT-4、Anthropic Claude、Google Gemini 还是本地 Ollama 模型，切换只需改一行代码——这让你可以在开发阶段用便宜的模型快速迭代，上线时切换到最强模型。

4.1 初始化不同提供商的模型

# === OpenAI ===
from langchain_openai import ChatOpenAI
openai_llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,           # 0 = 确定性输出, 0.7-1 = 创意性输出
    max_tokens=4096,
    timeout=30,
    max_retries=2,
)

# === Anthropic Claude ===
from langchain_anthropic import ChatAnthropic
claude_llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0,
    max_tokens=4096,
)

# === Google Gemini ===
from langchain_google_genai import ChatGoogleGenerativeAI
gemini_llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-pro",
    temperature=0,
)

# === 本地 Ollama ===
from langchain_ollama import ChatOllama
local_llm = ChatOllama(
    model="qwen2.5:14b",
    temperature=0,
    base_url="http://localhost:11434",
)

4.2 策略模式：运行时动态切换

import os
from typing import Literal

ProviderType = Literal["openai", "anthropic", "gemini", "ollama"]

def get_llm(provider: ProviderType = "openai"):
    """根据配置动态切换模型提供商，支持通过环境变量注入配置"""
    configs = {
        "openai": {
            "factory": lambda: ChatOpenAI(
                model=os.getenv("OPENAI_MODEL", "gpt-4o"),
                temperature=float(os.getenv("LLM_TEMPERATURE", "0.3")),
                timeout=int(os.getenv("LLM_TIMEOUT", "30")),
            ),
            "api_key_env": "OPENAI_API_KEY",
        },
        "anthropic": {
            "factory": lambda: ChatAnthropic(
                model=os.getenv("ANTHROPIC_MODEL", "claude-3-5-sonnet-20241022"),
                temperature=float(os.getenv("LLM_TEMPERATURE", "0.3")),
            ),
            "api_key_env": "ANTHROPIC_API_KEY",
        },
        "gemini": {
            "factory": lambda: ChatGoogleGenerativeAI(
                model=os.getenv("GEMINI_MODEL", "gemini-1.5-pro"),
                temperature=float(os.getenv("LLM_TEMPERATURE", "0.3")),
            ),
            "api_key_env": "GOOGLE_API_KEY",
        },
    }

    config = configs.get(provider)
    if not config:
        raise ValueError(f"不支持的提供商: {provider}。可选: {list(configs.keys())}")

    if config["api_key_env"] not in os.environ:
        raise ValueError(
            f"使用 {provider} 需要设置环境变量 {config['api_key_env']}"
        )

    return config["factory"]()

# 一行切换
llm = get_llm("anthropic")  # 从 OpenAI 切换到 Claude

4.3 缓存机制：避免重复 API 调用

在实际应用中，相同的问题可能被重复提问。启用缓存可以节省 30-50% 的 API 成本：

from langchain_core.caches import InMemoryCache
from langchain_core.globals import set_llm_cache

# 内存缓存（进程内，适合开发环境）
set_llm_cache(InMemoryCache())

# 生产环境推荐 SQLite 缓存（持久化，跨进程共享）
from langchain_community.caches import SQLiteCache
set_llm_cache(SQLiteCache(database_path=".langchain_cache.db"))

# 现在重复调用相同 Prompt 会直接返回缓存结果
for _ in range(3):
    result = llm.invoke("总结一下 LangChain 的核心组件")
    # 第一次：调用 API
    # 第二次、第三次：直接返回缓存（0 Token 消耗）

4.4 回退策略：高可用架构

将多个模型组成回退链，主模型不可用时自动降级：

# 模型回退策略
from langchain_openai import ChatOpenAI

# 主力模型：能力强，成本高
primary = ChatOpenAI(model="gpt-4o", max_retries=1, timeout=15)

# 降级模型 1：能力中，成本低
fallback_1 = ChatOpenAI(model="gpt-4o-mini", max_retries=2)

# 降级模型 2：本地模型，零成本（需 Ollama 运行中）
fallback_2 = ChatOllama(model="qwen2.5:7b", timeout=10)

# 构建回退链
robust_llm = primary.with_fallbacks([fallback_1, fallback_2])

# 使用——自动容错
response = robust_llm.invoke("这个问题的答案是什么？")
# 如果 gpt-4o 超时 → 自动切到 gpt-4o-mini
# 如果 gpt-4o-mini 也失败 → 自动切到本地模型

实战数据：在某企业级项目中，启用三级回退后，API 调用成功率达到 99.97%（原 99.1%），同时总成本下降约 35%（约 15% 的请求被降级到 mini 模型）。

模型	适用任务	相对成本	推荐场景
GPT-4o	复杂推理、代码生成	$$$$	核心业务逻辑
Claude 3.5 Sonnet	长文分析、多步推理	$$$	文档处理
GPT-4o-mini	分类、翻译、摘要	$	预处理、路由
Gemini 1.5 Flash	批量处理、快速响应	$	高吞吐场景
Qwen 2.5 (本地)	格式转换、简单问答	0	离线/隐私场景

五、Chain 组合技：串联多个 AI 调用

单个 AI 调用能做的事情有限，真正的魔力在于把多个调用串联成管道。LangChain 提供了丰富的 Chain 组合模式。

5.1 Sequential Chain：顺序执行

典型的「分析→方案→代码」三阶段流水线：

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

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

# 阶段 1：分析问题
analyze_prompt = ChatPromptTemplate.from_template(
    """分析以下问题，拆解为技术要点：
问题：{problem}
输出：用列表列出所有技术要点"""
)
analyze_chain = analyze_prompt | llm | StrOutputParser()

# 阶段 2：设计方案（接收阶段 1 的输出）
design_prompt = ChatPromptTemplate.from_template(
    """基于以下技术分析，设计详细方案：
{analysis}
输出格式：每个方案包含「技术选型」「核心逻辑」「伪代码」三个部分"""
)
design_chain = design_prompt | llm | StrOutputParser()

# 阶段 3：生成代码（接收阶段 2 的输出）
code_prompt = ChatPromptTemplate.from_template(
    """根据以下设计方案，编写完整的 Python 代码：
{design}
要求：代码可运行、有注释、包含错误处理"""
)
code_chain = code_prompt | llm | StrOutputParser()

# 串联执行
analysis = analyze_chain.invoke({"problem": "实现一个 LRU 缓存"})
design = design_chain.invoke({"analysis": analysis})
code = code_chain.invoke({"design": design})

print("=== 技术分析 ===")
print(analysis[:300] + "...")
print("\n=== 设计方案 ===")
print(design[:300] + "...")
print("\n=== 生成代码 ===")
print(code[:500] + "...")

5.2 RunnableParallel：并行处理

当多个子任务互不依赖时，并行处理能大幅缩短总耗时。以下示例同时生成中英文两个版本的代码注释：

from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda

# 中文注释生成器
cn_comment_chain = (
    ChatPromptTemplate.from_template(
        "给以下 Python 代码添加详细的中文注释（只输出带注释的代码）：\n{code}"
    )
    | llm
    | StrOutputParser()
)

# 英文注释生成器
en_comment_chain = (
    ChatPromptTemplate.from_template(
        "Add detailed English comments to the following Python code (output only commented code):\n{code}"
    )
    | llm
    | StrOutputParser()
)

# 并行执行（两个 LLM 调用同时进行，总耗时 = max(单个耗时)）
parallel_chain = RunnableParallel(
    chinese=cn_comment_chain,
    english=en_comment_chain,
)

result = parallel_chain.invoke("""
def merge_sort(arr):
    if len(arr) <= 1:
        return arr
    mid = len(arr) // 2
    left = merge_sort(arr[:mid])
    right = merge_sort(arr[mid:])
    return merge(left, right)
""")

print("=== 中文注释版 ===")
print(result["chinese"][:300])
print("\n=== 英文注释版 ===")
print(result["english"][:300])

5.3 Router Chain：条件路由

根据输入内容自动分发到不同的处理分支——这是构建「智能客服系统」的核心模式：

# 第一步：分类器
router_prompt = ChatPromptTemplate.from_template(
    """判断以下用户输入属于哪个类别。只回复类别代码，不要回复其他内容。
    
类别代码：
TECH - 技术问题（编程、系统、架构等）
SALES - 售前咨询（价格、功能、对比等）
SUPPORT - 售后支持（Bug 报修、使用问题等）
CHAT - 闲聊/其他

用户输入：{input}
类别代码："""
)

classifier = router_prompt | llm | StrOutputParser()

# 第二步：三个处理分支
tech_handler = (
    ChatPromptTemplate.from_template(
        "作为技术专家，简洁专业地解答：{input}\n如果涉及代码，给出完整可运行的示例。"
    ) | llm | StrOutputParser()
)

sales_handler = (
    ChatPromptTemplate.from_template(
        "作为销售顾问，热情专业地介绍产品：{input}\n突出核心优势，适时引导进一步沟通。"
    ) | llm | StrOutputParser()
)

support_handler = (
    ChatPromptTemplate.from_template(
        "作为售后支持，耐心专业地帮用户解决问题：{input}\n优先提供自助解决方案，必要时引导提交工单。"
    ) | llm | StrOutputParser()
)

# 第三步：路由函数
from langchain_core.runnables import RunnableLambda

def route_to_handler(input_dict):
    category = classifier.invoke({"input": input_dict["input"]}).strip()
    print(f"[路由] 分类结果: {category}")
    
    handler_map = {
        "TECH": tech_handler,
        "SALES": sales_handler,
        "SUPPORT": support_handler,
        "CHAT": chat_handler,
    }
    handler = handler_map.get(category, chat_handler)  # 默认走闲聊
    return handler.invoke({"input": input_dict["input"]})

router = RunnableLambda(route_to_handler)

# 测试不同输入
for query in [
    "Python 的 GIL 是什么？有什么影响？",
    "你们的产品多少钱一个月？",
    "登录后页面一直白屏，怎么解决？",
]:
    print(f"\n{'='*50}")
    print(f"用户: {query}")
    print(f"系统: {router.invoke({'input': query})[:150]}...")

5.4 Chain 类型对比总结

Chain 类型	执行模式	耗时	适用场景	复杂度
简单 Chain	单步 LLM	1s	翻译、分类	⭐
Sequential	A→B→C 顺序	3s	分析→方案→代码	⭐⭐
Parallel	A/B/C 同时	1.5s	多语言、多角度	⭐⭐
Router	分支择一	1s	客服路由	⭐⭐⭐
Conditional	动态分支	变化	复杂决策	⭐⭐⭐⭐

六、Tool 与 Agent：让 AI 学会调用外部工具

Chain 是固定的流程，Agent 则是「自主决策」的智能体。AI 不再按部就班执行预设步骤，而是根据任务自主选择工具、决定执行顺序、判断何时完成。

6.1 定义自定义 Tool

使用 @tool 装饰器，把一个普通 Python 函数变成 AI 可调用的工具：

from langchain_core.tools import tool
from datetime import datetime
from zoneinfo import ZoneInfo

@tool
def search_knowledge(query: str) -> str:
    """在内部知识库中搜索公司制度、政策、流程等信息。
    
    Args:
        query: 搜索关键词或自然语言问题
    """
    # 模拟知识库（实际项目中这里连接向量数据库）
    knowledge = {
        "请假": "员工每年享有 15 天带薪年假，需提前 3 个工作日在 OA 系统提交请假申请，经直属上级审批。",
        "报销": "差旅报销需在返回后 7 个工作日内提交，需附电子发票和行程单。住宿标准：一线城市 500 元/晚，其他 350 元/晚。",
        "加班": "工作日加班按 1.5 倍时薪计算，周末 2 倍，法定节假日 3 倍。加班需提前在 OA 提交加班申请。",
        "入职": "新员工入职需携带身份证、学历证书、离职证明。入职当天签署劳动合同，HR 系统自动开通企业邮箱和 OA 账号。",
        "离职": "离职需提前 30 天书面申请，完成工作交接和资产归还后办理离职手续。",
    }
    for key, answer in knowledge.items():
        if key in query:
            return f"[知识库查询结果] {answer}"
    return f"[知识库] 未找到关于「{query}」的相关信息。建议联系 HR 部门获取最新政策。"

@tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
    """获取指定时区的当前日期和时间。
    
    Args:
        timezone: 时区名称，例如 Asia/Shanghai, America/New_York, Europe/London
    """
    try:
        now = datetime.now(ZoneInfo(timezone))
        weekday_cn = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
        return f"{now.strftime('%Y-%m-%d %H:%M:%S')} {weekday_cn[now.weekday()]} ({timezone})"
    except Exception as e:
        return f"获取时间失败: {e}"

@tool
def calculate_math(expression: str) -> str:
    """安全的数学计算器，支持基本运算。
    
    Args:
        expression: 数学表达式，例如 "2 + 3 * 4", "sqrt(16)", "100 / 3"
    """
    import math
    safe_builtins = {
        "abs": abs, "round": round, "min": min, "max": max,
        "sum": sum, "pow": pow, "sqrt": math.sqrt, "log": math.log,
    }
    try:
        result = eval(expression, {"__builtins__": safe_builtins}, safe_builtins)
        return f"计算: {expression} = {result}"
    except Exception as e:
        return f"计算失败: {e}。支持的运算：+, -, *, /, sqrt(), log(), pow() 等"

# 注册工具列表
tools = [search_knowledge, get_current_time, calculate_math]
print(f"✅ 已注册 {len(tools)} 个工具")

6.2 创建 React Agent（LangGraph 方式）

LangGraph 是 LangChain 官方推荐的 Agent 构建方式，支持复杂的循环和条件分支：

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# 系统提示词（决定 Agent 的行为模式）
system_prompt = """你是一个企业智能助手。你可以使用以下工具：

工具说明：
1. search_knowledge - 查询公司制度（请假、报销、入职等）
2. get_current_time - 获取当前时间
3. calculate_math - 数学计算

行为规则：
- 遇到公司制度类问题，必须先用 search_knowledge 查询
- 涉及日期的计算，先获取当前时间
- 不确定的事情，明确告知用户，不要编造信息
- 回答要友善、专业，必要时使用列表格式"""

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_react_agent(
    model=llm,
    tools=tools,
    prompt=system_prompt
)

# 执行 Agent
test_questions = [
    "我今年还有几天年假？",
    "下周五是几号？",
    "我出差去上海，住 3 晚可以报销多少？加上往返机票 2500，总共能报销多少？",
]

for q in test_questions:
    print(f"\n{'='*50}")
    print(f"👤 用户: {q}")
    response = agent.invoke({"messages": [{"role": "user", "content": q}]})
    
    # 打印 Agent 的思考过程
    for msg in response["messages"]:
        role = msg.type.upper()
        content = msg.content[:120] if msg.content else "(tool call)"
        print(f"  [{role}] {content}...")

6.3 Agent 决策循环（ReAct 模式）

Agent 的核心是经典的 **ReAct（Reasoning + Acting）**循环：

用户输入："今年还有几天年假？下周五是几号？"
    │
    ▼
┌─────────────────┐
│  Thought        │ "需要查制度获取年假天数，还需要获取当前日期"
│  (Reasoning)    │
└───────┬─────────┘
        │
        ▼
┌─────────────────┐    ┌─────────────────┐
│  Action 1       │    │  Action 2       │
│  search_knowl() │    │  get_time()     │ ← 并行调用
│  参数: "年假"    │    │  时区: Shanghai  │
└───────┬─────────┘    └───────┬─────────┘
        │                      │
        ▼                      ▼
┌─────────────────┐    ┌─────────────────┐
│  Observation 1  │    │  Observation 2  │
│  "15天年假"     │    │  "2025-06-06"   │
└───────┬─────────┘    └───────┬─────────┘
        │                      │
        └──────────┬───────────┘
                   ▼
          ┌─────────────────┐
          │  Thought 2      │ "已有信息充足，可以回答了"
          └───────┬─────────┘
                  ▼
          ┌─────────────────┐
          │  Final Answer   │ "您今年有 15 天年假。下周五是 6 月 6 日。"
          └─────────────────┘

6.4 进阶技巧：结构化 Tool 输出

当 Tool 的输出需要被程序化消费（而非只是给 AI 看），使用结构化输出：

from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List, Optional

class SearchResult(BaseModel):
    """知识库搜索结果"""
    found: bool = Field(description="是否找到匹配结果")
    content: str = Field(description="匹配的知识库内容")
    confidence: float = Field(description="匹配置信度 (0-1)")
    sources: List[str] = Field(default_factory=list, description="信息来源")

class ToolOutput(BaseModel):
    """统一的工具输出格式"""
    success: bool
    data: Optional[dict] = None
    error: Optional[str] = None

@tool(args_schema=SearchResult)
def structured_search(query: str) -> str:
    """带结构化输出的知识搜索"""
    # 实际搜索逻辑...
    result = SearchResult(
        found=True,
        content="员工每年享有 15 天带薪年假",
        confidence=0.95,
        sources=["员工手册 v3.2", "OA 制度公告 2024-01"]
    )
    return result.json()

6.5 工具设计最佳实践

原则	说明	好例子	坏例子
职责单一	一个工具只做一件事	`search_docs(query)`	`universal_handler()`
描述精确	docstring 就是 prompt	"搜索公司制度文档"	"处理数据"
类型安全	定义清晰的参数 schema	`query: str, limit: int`	`**kwargs`
错误友好	返回可读的错误信息	"未找到结果，建议..."	`raise Exception()`
幂等设计	读操作可无限重复	`get_time()`	重复调写 API

七、Memory 模块：带记忆的对话系统

LLM 天然是无状态的——每次调用都像第一次见面。Memory 模块为对话注入上下文，让 AI「记住」之前的对话内容。

7.1 ConversationBufferMemory：完整记忆

存储全部对话历史，适合短对话场景：

from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

memory = ConversationBufferMemory(
    return_messages=True,  # 返回 Message 对象而非字符串
    memory_key="history"   # Prompt 中引用此 key
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的 AI 助手，叫小智。回答简洁、有温度。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | llm | StrOutputParser()

# 模拟多轮对话
conversation = [
    "你好！我叫张三，是一名 Python 后端工程师。",
    "我最近在做一个 FastAPI 项目，有什么建议吗？",
    "你还记得我的名字吗？",
]

for user_input in conversation:
    response = chain.invoke({
        "input": user_input,
        "history": memory.load_memory_variables({})["history"]
    })
    print(f"👤: {user_input}")
    print(f"🤖: {response}\n")
    memory.save_context(
        {"input": user_input},
        {"output": response}
    )

7.2 ConversationSummaryMemory：摘要记忆

当对话变长，完整历史会撑爆上下文窗口。SummaryMemory 自动生成摘要，只保留关键信息：

from langchain.memory import ConversationSummaryMemory

summary_memory = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),  # 用便宜模型做摘要
    memory_key="history",
    return_messages=True,
)

# 长对话后，memory 中的内容会变成：
# "用户名叫张三，是 Python 后端工程师。讨论了 FastAPI 项目建议，包括
#  使用依赖注入、Pydantic 模型验证、异步数据库操作等要点。"

7.3 ConversationBufferWindowMemory：滑动窗口

只保留最近 K 轮对话，固定 Token 消耗：

from langchain.memory import ConversationBufferWindowMemory

window_memory = ConversationBufferWindowMemory(
    k=3,               # 只保留最近 3 轮
    return_messages=True,
    memory_key="history"
)
# 第 4 轮对话时，第 1 轮会被自动丢弃

7.4 组合策略：混合记忆

实际项目中最常用的模式——短期对话用 Window，长期记忆用 Summary：

from langchain.memory import (
    ConversationBufferWindowMemory,
    ConversationSummaryMemory,
    CombinedMemory,
)

# 短期记忆：最近 3 轮完整对话
short_term = ConversationBufferWindowMemory(
    k=3, memory_key="recent_history", return_messages=True
)

# 长期记忆：全对话摘要
long_term = ConversationSummaryMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    memory_key="summary",
    return_messages=False,  # 摘要用字符串
)

# 组合记忆
combined_memory = CombinedMemory(memories=[short_term, long_term])

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是小智。对话摘要：{summary}"),
    MessagesPlaceholder(variable_name="recent_history"),
    ("human", "{input}")
])

7.5 记忆策略选择指南

记忆类型	Token 消耗	信息保真度	适用场景	推荐指数
Buffer	无限增长	100%	短对话（< 10 轮）	⭐⭐⭐
Summary	固定（摘要长度）	80-90%	长对话、教育辅导	⭐⭐⭐⭐
Window (K=5)	固定（5轮）	近期 100%	实时客服、闲聊	⭐⭐⭐⭐⭐
混合（Window+Summary）	可控	90%+	企业级应用	⭐⭐⭐⭐⭐

八、RAG 实战：构建知识库问答系统

RAG（Retrieval-Augmented Generation，检索增强生成）是 LangChain 最经典、最实用的应用模式。核心思路：先检索相关文档，再让 LLM 基于文档回答——既利用了 LLM 的语言能力，又避免了幻觉。

8.1 完整 RAG 系统实现

以下是一个可运行的生产级 RAG 系统，涵盖文档加载、分割、向量化、存储、检索和生成的完整流程：

# === rag_system.py ===
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

load_dotenv()

# === 步骤 1：加载文档 ===
loader = DirectoryLoader(
    "./docs/",
    glob="**/*.md",           # 支持 Markdown 文件
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"},
    show_progress=True
)
documents = loader.load()
print(f"✅ 加载了 {len(documents)} 个文档，总大小: {sum(len(d.page_content) for d in documents)} 字符")

# === 步骤 2：文档分割 ====20
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,           # 每块最大 800 字符
    chunk_overlap=150,         # 块间重叠 150 字符（保持语义连续性）
    separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
    length_function=len,
)
chunks = text_splitter.split_documents(documents)
print(f"✅ 分割为 {len(chunks)} 个文本块")

# === 步骤 3：向量化与存储 ===
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
    collection_name="tech_docs"
)
print(f"✅ 向量数据库已创建并持久化到 ./chroma_db")

# === 步骤 4：构建 RAG Chain ===
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 语义相似度搜索
    search_kwargs={"k": 4}     # 检索最相关的 4 个片段
)

# 格式化函数
def format_docs(docs):
    """将检索到的文档片段格式化为 LLM 可读的上下文"""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "未知来源").split("/")[-1]
        formatted.append(
            f"[参考片段 {i}] 来源: {source}\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(formatted)

# RAG Prompt 模板（关键！决定回答质量）
rag_template = """你是一个专业的技术文档问答助手。请根据以下检索到的文档片段回答用户问题。

核心规则：
1. 回答必须基于提供的文档内容，不要添加文档中没有的信息
2. 如果文档中找不到答案，明确告知用户「文档中未找到相关信息」，不要编造
3. 回答要简洁、有结构（使用列表或表格），并在文末注明引用的片段编号
4. 如果检索到的内容不相关，说明文档可能不包含用户需要的信息

<检索到的文档片段>
{context}
</检索到的文档片段>

用户问题：{question}

请回答："""

rag_prompt = ChatPromptTemplate.from_template(rag_template)
llm = ChatOpenAI(model="gpt-4o", temperature=0.2)

# 组装 RAG Chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

# === 步骤 5：查询测试 ===
test_questions = [
    "系统支持哪些数据库？版本要求是什么？",
    "如何进行 API 认证？支持哪些认证方式？",
    "部署到生产环境有哪些注意事项？",
]

for i, q in enumerate(test_questions, 1):
    print(f"\n{'='*60}")
    print(f"Q{i}: {q}")
    print(f"{'='*60}")
    
    # 显示检索到的文档（调试用）
    retrieved = retriever.invoke(q)
    print(f"[检索到 {len(retrieved)} 个相关片段]")
    for j, doc in enumerate(retrieved, 1):
        print(f"  片段 {j}: {doc.metadata.get('source', '?')[-40:]} ({len(doc.page_content)} 字符)")
    
    # 生成回答
    answer = rag_chain.invoke(q)
    print(f"\n📝 回答:")
    print(answer)

8.2 RAG 系统架构

8.3 文档分割策略深入

分割策略直接影响 RAG 的检索质量。不同场景选择不同分割器：

# 策略 1：固定大小（最通用）
from langchain_text_splitters import RecursiveCharacterTextSplitter
generic_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
)

# 策略 2：按 Markdown 标题层级（技术文档最佳）
from langchain_text_splitters import MarkdownHeaderTextSplitter

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ],
    strip_headers=False,  # 保留标题在 chunk 中
)

# 策略 3：代码语义分割
from langchain_text_splitters import (
    Language,
    RecursiveCharacterTextSplitter,
)

python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=500,
    chunk_overlap=50,
)

# 策略 4：语义分割（基于 embedding 相似度变化）
# 当相邻句子的语义相似度下降超过阈值时在此处断开
from langchain_experimental.text_splitter import SemanticChunker

semantic_splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile"  # 按百分位
)

分割策略	最佳适用	语义保真度	实现复杂度
递归字符	通用文本	⭐⭐⭐	⭐
Markdown 标题	API 文档、技术手册	⭐⭐⭐⭐⭐	⭐⭐
代码语言	Python/JS/TS 等源码	⭐⭐⭐⭐	⭐⭐
语义分割	长文、叙事	⭐⭐⭐⭐⭐	⭐⭐⭐

8.4 高级 RAG 优化技术

# 1. 多路混合检索（Hybrid Search：BM25 + 向量）
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

bm25_retriever = BM25Retriever.from_documents(chunks, k=5)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

ensemble = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # BM25 30% + 向量 70%
)

# 2. 查询重写（Query Rewriting）
rewrite_template = """将用户的原始问题重写为更适合向量检索的关键词查询。
原始问题：{question}
重写后的查询（只输出查询文本，不要解释）："""

rewrite_chain = (
    ChatPromptTemplate.from_template(rewrite_template)
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

# 优化后的检索流程
optimized_retriever = rewrite_chain | retriever

# 3. 上下文压缩（只保留与问题相关的部分）
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(
    ChatOpenAI(model="gpt-4o-mini", temperature=0)
)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)
# 这个 retriever 会自动去掉文档中与问题无关的部分

九、生产环境部署要点

开发阶段在本地跑通 Demo 只是第一步，部署到生产环境需要考虑稳定性、性能、监控和成本。

9.1 流式输出（提升用户体验）

用户不能干等 5 秒看一个完整回复——流式输出让用户看到 AI「正在打字」：

# 同步流式
for chunk in rag_chain.stream("什么是 Kubernetes？"):
    print(chunk, end="", flush=True)

# 异步流式（FastAPI 场景）
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

@app.post("/chat/stream")
async def chat_stream(question: str):
    async def generate():
        async for chunk in rag_chain.astream(question):
            yield chunk
    return StreamingResponse(generate(), media_type="text/plain")

9.2 错误处理与指数退避

import time
import logging

logger = logging.getLogger(__name__)

def robust_invoke(chain, input_data, max_retries: int = 3):
    """带指数退避的重试调用"""
    last_exception = None
    
    for attempt in range(1, max_retries + 1):
        try:
            return chain.invoke(input_data)
        except Exception as e:
            last_exception = e
            wait = 2 ** attempt  # 2s, 4s, 8s
            
            if "rate_limit" in str(e).lower():
                wait *= 2  # 限流错误翻倍等待
            
            logger.warning(
                f"LLM 调用失败 (第 {attempt}/{max_retries} 次尝试), "
                f"{wait}s 后重试: {str(e)[:100]}"
            )
            
            if attempt < max_retries:
                time.sleep(wait)
            else:
                logger.error(f"所有重试失败: {str(last_exception)[:200]}")
                raise RuntimeError(
                    f"LLM 调用在 {max_retries} 次重试后仍然失败"
                ) from last_exception

9.3 LangServe：一行代码部署 API

# serve.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from langserve import add_routes

app = FastAPI(title="LangChain RAG API", version="1.0.0")

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 一行代码将 Chain 暴露为 REST API
add_routes(
    app,
    rag_chain,
    path="/rag",
    # 自动生成 OpenAPI 文档 + /rag/playground 交互式 UI
)

# 启动命令: uvicorn serve:app --host 0.0.0.0 --port 8000 --workers 4

LangServe 自动提供：/rag/playground（在线调试界面）、/rag/docs（OpenAPI 文档）、/rag/input_schema、/rag/output_schema

9.4 LangSmith 监控集成

# 在脚本开头添加这三行，所有 Chain/Agent 调用会自动上报
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your-project-key"
os.environ["LANGCHAIN_PROJECT"] = "production-rag"

# 无需修改任何业务代码！
# LangSmith 仪表盘会自动显示：
# - 每次调用的完整链路（Prompt → Model → Parser）
# - Token 消耗统计
# - 延迟 P50/P99
# - 错误率趋势
# - 用户反馈收集

十、性能优化与成本控制

10.1 Token 成本速算

理解 Token 消耗是控制成本的第一步：

模型	输入价格 ($/1M tokens)	输出价格 ($/1M tokens)	上下文窗口
GPT-4o	$2.50	$10.00	128K
GPT-4o-mini	$0.15	$0.60	128K
Claude 3.5 Sonnet	$3.00	$15.00	200K
Gemini 1.5 Pro	$1.25	$5.00	2M

例：用 GPT-4o 处理一次 RAG 查询（输入 3000 tokens + 输出 500 tokens）= $0.0125。每天 10000 次查询 ≈ $125/天 ≈ $3750/月。

10.2 成本优化策略

# 优化 1：分层使用模型
from langchain_openai import ChatOpenAI

class CostAwareRouter:
    """根据任务复杂度选择不同模型"""
    
    def __init__(self):
        self.cheap_llm = ChatOpenAI(model="gpt-4o-mini")     # $0.15/1M in
        self.strong_llm = ChatOpenAI(model="gpt-4o")         # $2.50/1M in
    
    def route(self, query: str) -> str:
        # 简单任务走 cheap 模型
        if len(query) < 50 and any(kw in query for kw in ["今天", "时间", "天气"]):
            return self.cheap_llm.invoke(query)
        # 复杂任务走 strong 模型
        return self.strong_llm.invoke(query)

# 优化 2：Prompt 压缩
def compress_prompt(prompt: str) -> str:
    """删除多余空白、缩短示例、移除冗余说明"""
    import re
    # 合并多余空行
    prompt = re.sub(r'\n{3,}', '\n\n', prompt)
    # 移除过长的示例（保留前 2 个）
    lines = prompt.split('\n')
    example_count = 0
    result = []
    for line in lines:
        if line.strip().startswith('示例') or line.strip().startswith('Example'):
            example_count += 1
            if example_count > 2:
                continue
        result.append(line)
    return '\n'.join(result)

# 优化 3：启用缓存（见 4.3 节）
# 可节省 30-50% 重复查询成本

10.3 性能优化检查清单

优化项	预期提升	实现难度	副作用
启用 LLM 缓存	30-50% 成本↓	⭐	无
使用 mini 模型处理简单任务	60-80% 成本↓	⭐⭐	复杂任务质量下降
Prompt 压缩	20-40% 延迟↓	⭐⭐	可能丢失上下文
并行调用（RunnableParallel）	2-3x 吞吐↑	⭐⭐	功耗/并发限制
Batch API（OpenAI）	50% 成本↓	⭐⭐⭐	延迟增加（24h 窗口）
语义缓存（GPTCache）	40-60% 成本↓	⭐⭐⭐	需要额外基础设施
本地部署（vLLM/Ollama）	90%+ 成本↓	⭐⭐⭐⭐	需要 GPU / 维护

十一、常见问题解答

Q1：LangChain 和 LlamaIndex 有什么区别？

A：LangChain 是通用 LLM 应用框架（Agent、Chain、Memory、Tool），LlamaIndex 专注数据索引和检索（RAG）。两者可搭配使用——用 LlamaIndex 构建检索层，用 LangChain 构建 Agent。如果项目以 RAG 为主，LlamaIndex 更专精；需要 Agent、多步推理、工具集成，LangChain 更全面。

Q2：LangGraph 是什么？必须学吗？

A：LangGraph 是 LangChain 团队开发的状态机编排库，用于构建复杂的多步骤 Agent 工作流（循环、条件分支、人工审批节点）。简单说——Chain 是直线（A→B→C），LangGraph 是状态图（可以回头、循环）。如果你的 Agent 需要「查资料→分析→编写→审查→修改→再审查」这种带反馈循环的流程，LangGraph 是最佳选择。如果只是简单的问答 Chain，不需要 LangGraph。

Q3：本地模型（Ollama/Llama.cpp）能用 LangChain 吗？

A：完全可以。通过 ChatOllama 连接本地 Ollama 实例：

from langchain_ollama import ChatOllama
llm = ChatOllama(model="qwen2.5:14b", temperature=0)

# 注意：本地模型通常 Function Calling 能力有限
# 建议配合手动 JSON 解析的 Prompt 策略来模拟 Tool 调用

Q4：Token 消耗太快，如何控制？

A：四个优先级——① 用 ConversationSummaryMemory 或 WindowMemory 替代 BufferMemory；② Prompt 中删除不必要的 Few-Shot 示例，压缩系统提示词；③ 用 gpt-4o-mini 处理分类/路由/预处理任务，复杂生成才用 gpt-4o；④ 启用语义缓存（GPTCache），相似问题直接返回缓存结果。

Q5：Agent 经常调用错误的工具怎么办？

A：三步排查——① 检查工具的 docstring 是否清晰描述了使用场景和参数含义（Agent 只通过 docstring 理解工具）；② 在系统提示词中添加 2-3 个正确使用工具的 Few-Shot 示例；③ 使用结构化输出（with_structured_output）替代自由文本，让工具调用更可靠。如果还不行，考虑缩小工具数量（3-5 个工具效果最好）。

Q6：LangChain 版本更新太快，代码经常失效怎么办？

A：锁定版本是唯一解：

# requirements.txt
langchain==0.3.*
langchain-openai==0.2.*
langchain-community==0.3.*
langgraph==0.2.*

关注 LangChain 官方 Blog 的迁移指南，大版本升级前先在测试环境验证。建议每季度审视一次依赖更新。

Q7：RAG 检索效果不好，如何系统优化？

A：按优先级尝试——① 调整 chunk_size（500-1500）和 chunk_overlap（10-20%），这是影响最大的参数；② 使用 Hybrid Search（BM25 + 向量，权重建议 3:7）；③ 添加 Re-ranking 步骤（Cohere Rerank 或 BGE-Reranker），将检索到的 20 个候选精排为 4 个；④ 优化文档预处理（去除 HTML 噪声、提取结构化信息）；⑤ 尝试更好的 Embedding 模型（text-embedding-3-large 比 small 效果好 5-10%）；⑥ 实现查询重写（Query Rewriting），将用户口语化问题转为搜索友好的关键词。

Q8：如何评估 RAG 系统的质量？

A：推荐使用 RAGAS 框架（pip install ragas）自动化评估，关注三个指标——上下文召回率（Context Recall）、答案忠实度（Faithfulness）、答案相关性（Answer Relevancy）。建议建立 50-100 个标注测试用例，每次改动后跑评估。

Q9：LangChain 适合做生产环境的聊天机器人吗？

A：适合。需注意——① 用 LangServe 部署 API，自带 playground；② 启用 LangSmith 监控延迟和 Token 消耗；③ 实现三级回退保证可用性；④ Prompt 注入防护；⑤ 速率限制和并发控制。许多企业已用 LangChain 支撑日活 10 万+ 的应用。

Q10：学习 LangChain 的最佳路径是什么？

A：建议按以下路线——

第 1 天：跑通 Hello World Chain → 理解 Prompt + Model + Parser
第 2-3 天：掌握 Agent + 自定义 Tool（重点，这是 LangChain 的核心价值）
第 4-5 天：Memory 模块 + 多轮对话系统
第 6-8 天：RAG 完整系统（文档→分割→向量化→检索→生成）
第 9-10 天：LangGraph 状态机编排
第 11-14 天：生产部署 + 监控 + 优化

不要试图一开始就学全部——先掌握 Chain 和 Agent，80% 的应用场景用这两个就够了。

十二、总结与展望

核心要点回顾

组件	一句话总结	关键 API	学习优先级
Prompt	结构化你的 AI 指令，告别字符串拼接	`ChatPromptTemplate`	⭐⭐⭐⭐⭐
Model	统一多 LLM 接口，一行代码切换提供商	`ChatOpenAI` / `ChatAnthropic`	⭐⭐⭐⭐⭐
Chain	用管道操作符 `	` 串联多个 AI 调用	`prompt \| llm \| parser`
Agent	AI 自主选择工具、决定执行顺序	`create_react_agent`	⭐⭐⭐⭐⭐
Memory	为无状态 LLM 注入对话上下文	`Buffer` / `Window` / `Summary`	⭐⭐⭐⭐
RAG	检索+生成：知识库问答的标准范式	`retriever + prompt + llm`	⭐⭐⭐⭐⭐

学习路线图

入门阶段（第 1-3 天）
    ↓ Simple Chain + Prompt Template
进阶阶段（第 4-7 天）  
    ↓ Agent + Custom Tool + Memory
实战阶段（第 8-12 天）
    ↓ RAG 系统 + Hybrid Search + Re-ranking
高阶阶段（第 13-18 天）
    ↓ LangGraph 状态机 + Multi-Agent 协作
生产阶段（第 19-25 天）
    ↓ LangServe 部署 + LangSmith 监控 + 成本优化

未来趋势展望

到 2025 年下半年，AI Agent 开发正在经历几个重要变化：

Multi-Agent 协作：多个 Agent 像团队一样分工合作（规划者、执行者、审查者），LangGraph 和 CrewAI 都在积极拥抱这个方向
MCP 标准化：Anthropic 提出的 MCP 协议正在成为 AI 与外部工具通信的标准，LangChain 开始原生支持
本地 Agent 能力提升：Qwen 3、Llama 4 等开源模型在 Function Calling 能力上快速追赶闭源模型
Agent 安全：随着 Agent 获得更多工具权限，Prompt 注入防护、权限控制、审计日志变得至关重要

无论技术如何演进，LangChain 作为「LLM 应用开发的 Swift/React」这一地位短时间内不会改变。掌握它，你就掌握了构建下一代 AI 应用的钥匙。🚀

本文由 MarkShareX AI 自动创作，分类：AI Agent，方向：LangChain 实战