LangChain 实战：高级 RAG 与工具增强 Agent 深度构建指南
从 Multi-Query RAG 到多工具协同 Agent——掌握 LangChain 生产级 AI 应用开发的核心范式，构建真正"能查、能用、能推理"的智能体
目录
- 背景与概念：为什么你的 RAG 不够"聪明"
- RAG 核心模式深度解析
- 工具增强 Agent 架构设计
- 实战一：构建多策略自适应 RAG 系统
- 实战二：工具增强 Agent 完整实现
- 进阶：LangGraph 多 Agent 协作模式
- 生产部署与性能优化
- 常见问题 FAQ
- 总结与展望
一、背景与概念
1.1 RAG 的进化之路
检索增强生成（RAG）已经成为大语言模型（LLM）落地最主流的范式之一。它的核心思想并不复杂：在 LLM 回答用户问题之前，先从外部知识库中检索相关文档片段，然后将这些片段作为上下文注入 Prompt，让模型基于可靠信源生成回答。
但如果你已经在生产环境中跑过 RAG 系统，你一定遇到过这些问题：
- 检索不准：用户问"Python 中
async和await的区别"，检索系统却返回了一篇关于 JavaScript 异步编程的文章 - 上下文缺失：返回的文档片段太短，缺少必要的背景信息，模型只能"盲人摸象"
- 工具脱节：RAG 只能"查文档"，无法执行实际操作（查数据库、调 API、发邮件）
- 多步推理困难：复杂问题需要拆解成多个子任务，单次检索+生成的模式力不从心
这些痛点催生了 LangChain 在 RAG 和 Agent 领域的一系列深度能力。本篇文章将带你跳出"向量库 + LLM"的初级组合，探索以下高级范式：
| 范式 | 解决的核心问题 | 关键技术 |
|---|---|---|
| 多策略 RAG | 检索精度不足 | Multi-Query、Parent-Document、Contextual Compression |
| 工具增强 Agent | RAG 只能查不能做 | Tool/Function Calling、ReAct Agent |
| 多 Agent 协作 | 复杂任务拆解 | LangGraph、StateGraph、条件路由 |
| 生产级优化 | 延迟、成本、可靠性 | Streaming、Caching、Fallback |
1.2 本文读者收益
读完本文，你将能够：
- 理解并实现 4 种高级 RAG 检索策略，知道何时用哪种
- 构建 一个能调用多种外部工具的 Agent（搜索引擎、数据库、API、代码执行器）
- 掌握 LangGraph 的基础用法，搭建多 Agent 协作流水线
- 部署 一个支持流式输出、缓存、容错的生产级 AI 应用
二、RAG 核心模式深度解析
2.1 Multi-Query RAG：一个问题，多种问法
问题场景：用户的原始问题措辞模糊或不完整，单次检索难以命中正确的文档。
解决思路：让 LLM 从不同角度重新表述用户问题，生成 3-5 个变体查询，然后合并所有检索结果，去重后作为上下文。
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
# 初始化向量库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
collection_name="tech_docs",
embedding_function=embeddings,
persist_directory="./chroma_db"
)
# 创建基础检索器
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# 包装为 Multi-Query 检索器
llm = ChatOpenAI(model="gpt-4o", temperature=0)
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=base_retriever,
llm=llm,
include_original=True # 保留原始查询
)
# 使用示例
question = "怎么让 AI 应用跑得更快"
docs = multi_query_retriever.invoke(question)
# LLM 会自动生成类似以下变体：
# - "如何优化 AI 应用的推理速度"
# - "AI 应用性能调优的方法有哪些"
# - "加速大模型应用响应的技术方案"
# - "减少 LLM API 调用延迟的最佳实践"
# - "AI 应用吞吐量提升策略"
适用场景：
- 用户问题简短模糊（如"怎么优化"、"出错了怎么办"）
- 知识库覆盖领域广，同一概念有多种表述方式
- 需要确保检索召回率（宁可多召回也不遗漏）
注意事项：
- 每次检索会调用 LLM 生成变体 + N 次向量搜索，延迟和成本增加 N 倍
- 建议对变体数量做限制（3-5 个为宜），配合结果去重减少冗余
2.2 Parent-Document RAG：小块检索，大块回答
问题场景：为了检索精度，文档通常被切成较小的 chunk（如 256-512 tokens），但这导致检索到的片段缺少上下文，模型无法理解完整语义。
解决思路：建立一个"双层索引"——检索时用小块（child），返回结果时反查其所属的大块（parent），将完整的父文档作为上下文。
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 子文档分割器（小块，用于检索）
child_splitter = RecursiveCharacterTextSplitter(
chunk_size=300,
chunk_overlap=50
)
# 父文档分割器（大块，用于返回）
parent_splitter = RecursiveCharacterTextSplitter(
chunk_size=2000,
chunk_overlap=200
)
# 文档存储（用于子→父映射）
store = InMemoryStore()
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
search_kwargs={"k": 3} # 返回 3 个父文档
)
# 添加文档
retriever.add_documents(raw_documents)
# 检索时返回完整的父文档
results = retriever.invoke("Rust 的所有权系统如何工作")
# 每个 result 包含约 2000 tokens 的完整段落
效果对比：
| 方案 | 检索精度 | 上下文完整性 | 存储开销 |
|---|---|---|---|
| 小块检索 + 小块返回 | ⭐⭐⭐⭐⭐ | ⭐⭐ | 低 |
| 大块检索 + 大块返回 | ⭐⭐ | ⭐⭐⭐⭐⭐ | 低 |
| 小块检索 + 大块返回（本文方案） | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | 中（双份向量） |
2.3 Contextual Compression RAG：检索后再"提纯"
问题场景：即使检索到了相关文档，其中也包含大量与当前问题无关的信息，白白占用上下文窗口。
解决思路：在检索结果返回后，再用 LLM 对每个文档片段进行"压缩"——提取与问题相关的核心句子，丢弃无关内容。
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
# 创建压缩器
compressor = LLMChainExtractor.from_llm(llm)
# 包装检索器
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever
)
# 检索 + 自动压缩
compressed_docs = compression_retriever.invoke(
"如何在 Kubernetes 中配置 HPA 自动扩缩容"
)
# 返回的每个文档只保留了与 HPA 配置相关的句子
更高效的替代方案：如果你觉得每篇文档都调一次 LLM 成本太高，可以使用 LLMChainFilter——它不提取内容，只是判断文档是否相关，做二分类过滤：
from langchain.retrievers.document_compressors import LLMChainFilter
filter_compressor = LLMChainFilter.from_llm(llm)
filter_retriever = ContextualCompressionRetriever(
base_compressor=filter_compressor,
base_retriever=base_retriever
)
2.4 Self-Query RAG：元数据感知检索
问题场景：知识库中的文档带有丰富的元数据（日期、作者、类别、版本号等），但传统 RAG 只用语义相似度检索，完全忽略了这些结构化信息。
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
# 定义元数据字段
metadata_field_info = [
AttributeInfo(
name="source",
description="文档来源：官方文档、社区博客、学术论文",
type="string",
),
AttributeInfo(
name="published_date",
description="文档发布日期，格式 YYYY-MM-DD",
type="string",
),
AttributeInfo(
name="version",
description="框架/库的版本号",
type="string",
),
]
document_content_description = "技术文档和教程"
self_query_retriever = SelfQueryRetriever.from_llm(
llm=llm,
vectorstore=vectorstore,
document_contents=document_content_description,
metadata_field_info=metadata_field_info,
verbose=True
)
# 用户自然语言查询 → LLM 自动解析出语义+元数据过滤条件
docs = self_query_retriever.invoke(
"找 2025 年之后发布的关于 LangChain v0.3 的官方文档"
)
# LLM 内部生成的查询结构（自动推断）：
# {
# "query": "LangChain v0.3",
# "filter": "and(eq('source', '官方文档'), gte('published_date', '2025-01-01'), eq('version', '0.3'))"
# }
2.5 四种策略对比总结
| 策略 | 擅长场景 | 额外成本 | 实现复杂度 |
|---|---|---|---|
| Multi-Query | 问题模糊、需要高召回 | LLM 调用 × (N+1) | ⭐⭐ |
| Parent-Document | 需要完整上下文 | 双份向量存储 | ⭐⭐ |
| Contextual Compression | 上下文窗口紧张 | LLM 调用 × 文档数 | ⭐ |
| Self-Query | 元数据丰富的知识库 | LLM 调用 × 1 | ⭐⭐⭐ |
2.6 组合策略：在实际项目中如何选择
在实际项目中，很少只用单一策略。更常见的做法是根据场景组合使用。以下是三种典型的组合模式：
模式一：高精度知识库（文档 + FAQ）
Self-Query（按版本/日期过滤） → Parent-Document（返回完整上下文）
适合企业知识库、API 文档等结构化内容丰富的场景。先用 Self-Query 精准定位到特定版本的文档，再用 Parent-Document 返回完整章节，确保上下文完整。
模式二：通用问答助手
Multi-Query（提高召回） → Contextual Compression（压缩提纯）
适合对话式 AI 助手。Multi-Query 确保不遗漏相关文档，Compression 确保不浪费上下文窗口。这个组合在 LangChain 社区中被广泛验证为"最佳性价比组合"。
模式三：研究分析型
Multi-Query → Parent-Document → LLM 重排序（Re-rank） → Top-K
适合需要深度分析和交叉验证的场景（如学术研究、竞品分析）。Triple-check 机制最大化信息完整性，代价是延迟较高。
2.7 RAG 评估：你怎么知道策略选对了
选择 RAG 策略不是凭感觉，需要数据说话。推荐使用 RAGAS（RAG Assessment）框架进行评估：
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 答案是否基于提供的上下文
answer_relevancy, # 答案是否切题
context_recall, # 检索是否召回了必要信息
context_precision, # 检索结果中相关文档的比例
)
# 准备测试数据集
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": ["什么是 RAG？", "如何优化向量检索速度？", "Chroma 和 Pinecone 的区别是什么？"],
"answer": ["RAG 是检索增强生成...", "可以通过索引优化...", "Chroma 是开源嵌入式数据库..."],
"contexts": [
["RAG (Retrieval-Augmented Generation) 是一种结合检索和生成的技术..."],
["向量检索优化方法包括：使用 HNSW 索引、量化压缩、预过滤..."],
["Chroma 是开源向量数据库，Pinecone 是托管向量数据库..."],
],
})
# 执行评估
results = evaluate(eval_dataset, metrics=[faithfulness, answer_relevancy, context_recall, context_precision])
print(f"忠实度: {results['faithfulness']:.2f}")
print(f"切题度: {results['answer_relevancy']:.2f}")
print(f"召回率: {results['context_recall']:.2f}")
print(f"精确率: {results['context_precision']:.2f}")
评估的黄金法则：至少准备 50 个真实用户问题作为测试集，分别用不同策略跑一遍，对比各指标。不要凭直觉选策略，用数据决策才能真正优化你的 RAG 系统。
三、工具增强 Agent 架构设计
3.1 Agent 的"双手"：从只读到读写
传统 RAG 像一个"只读"的图书管理员——只能查资料，不能做任何实际操作。工具增强 Agent 则赋予了 AI"双手"，让它能够：
- 🔍 搜索网页：获取最新信息，弥补 LLM 知识截止日期
- 🗄️ 查询数据库：直接执行 SQL，获取结构化数据
- 🌐 调用 API：操作外部服务（发送邮件、创建工单、触发流水线）
- 💻 执行代码：运行 Python 脚本，做数据分析和可视化
- 📊 操作文件：读写本地或云存储中的文件
3.2 ReAct Agent：推理与行动的交织
ReAct（Reasoning + Acting）是 LangChain 中最经典的 Agent 范式。它的核心循环是：
Thought → Action → Observation → Thought → Action → ... → Final Answer
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool
from langchain_community.tools import DuckDuckGoSearchRun
import sqlite3
# 定义工具集
# 工具1：网页搜索
search_tool = DuckDuckGoSearchRun()
# 工具2：数据库查询
def query_database(sql: str) -> str:
"""执行 SQL 查询并返回结果。"""
conn = sqlite3.connect("knowledge.db")
cursor = conn.cursor()
try:
cursor.execute(sql)
results = cursor.fetchall()
return str(results)
except Exception as e:
return f"错误: {e}"
finally:
conn.close()
db_tool = Tool(
name="database_query",
func=query_database,
description="执行 SQL 查询数据库。输入应为完整的 SQL 语句。"
)
# 工具3：Python 代码执行器
def execute_python(code: str) -> str:
"""执行 Python 代码并返回 stdout 输出。"""
import subprocess
result = subprocess.run(
["python3", "-c", code],
capture_output=True, text=True, timeout=10
)
return result.stdout or result.stderr
python_tool = Tool(
name="python_executor",
func=execute_python,
description="执行 Python 代码。用于计算、数据处理、图表生成。"
)
# 工具4：当前时间
from datetime import datetime
time_tool = Tool(
name="current_time",
func=lambda _: datetime.now().isoformat(),
description="获取当前日期和时间。无输入参数。"
)
tools = [search_tool, db_tool, python_tool, time_tool]
# 创建 Agent
from langchain_openai import ChatOpenAI
from langchain import hub
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 使用 LangChain Hub 中的 ReAct prompt 模板
prompt = hub.pull("hwchase17/react")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=10,
handle_parsing_errors=True
)
# 使用 Agent
response = agent_executor.invoke({
"input": "帮我查一下今天几号，然后搜索 LangChain 今天有没有什么新闻，最后用 Python 计算一下 2026 年还剩下多少天"
})
3.3 OpenAI Functions Agent：更稳定、更精确
ReAct Agent 依赖 Prompt 工程来解析 LLM 输出中的 Action、Action Input 等标记，偶尔会出现解析失败。OpenAI Functions Agent 利用 Function Calling 原生能力，工具调用更可靠：
from langchain.agents import create_openai_functions_agent
# 同样的工具列表，不同的 Agent 类型
functions_agent = create_openai_functions_agent(llm, tools, prompt)
functions_executor = AgentExecutor(
agent=functions_agent,
tools=tools,
verbose=True,
max_iterations=5
)
# Functions Agent 内部使用 OpenAI 的 tool_choice 机制
# LLM 返回的不是文本标记，而是结构化的 function_call JSON
response = functions_executor.invoke({
"input": "搜索 LangGraph 最新版本号，用 Python 判断它是否大于 0.2.0"
})
两种 Agent 对比：
| 特性 | ReAct Agent | Functions Agent |
|---|---|---|
| 模型要求 | 任意 LLM | 支持 Function Calling 的模型 |
| 调用稳定性 | 偶尔解析失败 | 结构化输出，极其稳定 |
| 并行工具调用 | 不支持 | 支持（一次返回多个 tool_calls） |
| Prompt 控制力 | 高（可定制推理链） | 低（依赖模型内部机制） |
| 推荐场景 | 开源模型、需要精细控制 | OpenAI/Azure/兼容接口 |
四、实战一：构建多策略自适应 RAG 系统
4.1 系统架构
我们将构建一个自适应 RAG 路由器：根据用户问题的特征（模糊度、复杂度、元数据需求），自动选择最优检索策略。
用户问题 → 问题分类器 → [多策略路由]
├─ 简单事实查询 → 基础 RAG
├─ 模糊问题 → Multi-Query RAG
├─ 需要上下文 → Parent-Document RAG
├─ 元数据过滤 → Self-Query RAG
└─ 复杂综合查询 → Multi-Query + Compress
4.2 完整实现
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
# === 步骤 1：问题分类器 ===
class QueryAnalysis(BaseModel):
"""LLM 对用户问题的结构化分析"""
complexity: Literal["simple", "fuzzy", "contextual", "metadata", "complex"] = Field(
description="问题类型：simple=简单事实，fuzzy=表述模糊，contextual=需要上下文，metadata=元数据过滤，complex=综合复杂"
)
needs_compression: bool = Field(
description="是否需要对检索结果进行压缩提纯"
)
analysis_prompt = ChatPromptTemplate.from_messages([
("system", """分析用户问题的特征。
- simple: 直接的事实查询，如"X 是什么"
- fuzzy: 问题表述模糊、简短，如"怎么优化"
- contextual: 答案需要大量上下文才能理解
- metadata: 问题包含时间、版本、来源等过滤条件
- complex: 需要综合多种策略
只返回 JSON。"""),
("human", "{question}")
])
analyzer = analysis_prompt | llm.with_structured_output(QueryAnalysis)
# === 步骤 2：多策略检索器工厂 ===
class AdaptiveRAGRetriever:
def __init__(self, vectorstore, llm):
self.vectorstore = vectorstore
self.llm = llm
self._init_retrievers()
def _init_retrievers(self):
"""初始化所有检索策略"""
self.base = self.vectorstore.as_retriever(search_kwargs={"k": 5})
self.multi_query = MultiQueryRetriever.from_llm(
retriever=self.base, llm=self.llm
)
self.parent_doc = ParentDocumentRetriever(
vectorstore=self.vectorstore,
docstore=InMemoryStore(),
child_splitter=RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50),
parent_splitter=RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200),
)
compressor = LLMChainExtractor.from_llm(self.llm)
self.compressed = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=self.multi_query
)
def retrieve(self, question: str, analysis: QueryAnalysis):
"""根据问题分析结果选择策略"""
strategy_map = {
"simple": self.base,
"fuzzy": self.multi_query,
"contextual": self.parent_doc,
"metadata": self.base, # Self-Query 需要额外配置，此处简化
"complex": self.compressed,
}
retriever = strategy_map.get(analysis.complexity, self.base)
print(f"📊 问题类型: {analysis.complexity}, 压缩: {analysis.needs_compression}")
docs = retriever.invoke(question)
# 额外压缩（如果需要）
if analysis.needs_compression and analysis.complexity != "complex":
docs = compressor.compress_documents(docs, question)
return docs
# === 步骤 3：完整的 RAG 链 ===
def build_adaptive_rag_chain(vectorstore, llm):
retriever_engine = AdaptiveRAGRetriever(vectorstore, llm)
rag_prompt = ChatPromptTemplate.from_messages([
("system", """你是一个技术助手。根据提供的上下文回答问题。
如果上下文不足以回答，请明确说明。
上下文：
{context}"""),
("human", "{question}")
])
def adaptive_retrieve(inputs):
question = inputs["question"]
analysis = analyzer.invoke({"question": question})
docs = retriever_engine.retrieve(question, analysis)
return {"context": docs, "question": question, "analysis": analysis}
chain = (
RunnablePassthrough.assign(context_and_question=adaptive_retrieve)
| (lambda x: {
"context": "\n\n---\n\n".join(
d.page_content for d in x["context_and_question"]["context"]
),
"question": x["context_and_question"]["question"]
})
| rag_prompt
| llm
| StrOutputParser()
)
return chain
# === 使用示例 ===
chain = build_adaptive_rag_chain(vectorstore, llm)
# 简单问题 → 基础 RAG
result = chain.invoke({"question": "什么是 VectorStore"})
# 模糊问题 → Multi-Query RAG
result = chain.invoke({"question": "怎么加速"})
# 复杂问题 → Multi-Query + Compression
result = chain.invoke({"question": "比较 Chroma、Pinecone 和 Weaviate 在性能、成本和易用性上的差异，并给出生产环境选型建议"})
五、实战二：工具增强 Agent 完整实现
5.1 场景设定
构建一个技术问答助手 Agent，它能：
- 从向量知识库检索技术文档（RAG 工具）
- 实时搜索网页获取最新信息（搜索工具）
- 执行 Python 代码做计算和验证（代码工具）
- 查询数据库获取统计数据（数据库工具）
- 智能选择工具组合完成任务
5.2 工具定义与注册
from langchain.tools import tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
# === 工具 1：知识库检索（RAG 工具） ===
@tool
def search_knowledge_base(query: str) -> str:
"""从内部技术知识库中检索相关文档。适合查找概念、教程、最佳实践。"""
docs = vectorstore.similarity_search(query, k=3)
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "未知来源")
results.append(f"[{i}] 来源: {source}\n{doc.page_content[:500]}")
return "\n\n---\n\n".join(results) if results else "未找到相关文档"
# === 工具 2：网页搜索 ===
@tool
def web_search(query: str) -> str:
"""搜索互联网获取最新信息。适合查找新闻、最新版本、社区讨论。"""
# 使用免费搜索引擎（需安装 duckduckgo-search）
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=3))
return "\n\n".join(
f"[{i+1}] {r['title']}\n{r['body'][:300]}\n链接: {r['href']}"
for i, r in enumerate(results)
)
# === 工具 3：Python 执行器 ===
@tool
def run_python(code: str) -> str:
"""执行 Python 代码并返回结果。用于计算、数据分析、格式转换。
代码中可用 print() 输出结果。"""
import io, sys, traceback
old_stdout = sys.stdout
sys.stdout = buffer = io.StringIO()
try:
# 安全限制：禁用危险操作
safe_globals = {
"__builtins__": {
k: v for k, v in __builtins__.items()
if k not in ["__import__", "eval", "exec", "open", "compile"]
},
"print": print, "range": range, "len": len,
"int": int, "float": float, "str": str, "list": list,
"dict": dict, "set": set, "tuple": tuple, "bool": bool,
"sum": sum, "max": max, "min": min, "sorted": sorted,
"abs": abs, "round": round, "enumerate": enumerate,
"zip": zip, "map": map, "filter": filter,
}
exec(code, safe_globals, {})
return buffer.getvalue() or "代码执行完毕（无输出）"
except Exception as e:
return f"执行错误: {traceback.format_exc()}"
finally:
sys.stdout = old_stdout
# === 工具 4：数据库查询 ===
@tool
def query_stats(sql: str) -> str:
"""查询统计数据库。表结构：
- articles(id, title, category, views, published_date)
- users(id, name, join_date, active)
输入完整的 SQL SELECT 语句。"""
import sqlite3
conn = sqlite3.connect("stats.db")
try:
cur = conn.cursor()
cur.execute(sql)
rows = cur.fetchall()
cols = [d[0] for d in cur.description]
# 格式化为表格
header = " | ".join(cols)
separator = "-+-".join("-" * len(c) for c in cols)
body = "\n".join(" | ".join(str(v) for v in row) for row in rows[:20])
return f"{header}\n{separator}\n{body}\n\n共 {len(rows)} 条记录"
except Exception as e:
return f"SQL 错误: {e}"
finally:
conn.close()
# 工具注册
all_tools = [search_knowledge_base, web_search, run_python, query_stats]
5.3 Agent 创建与对话管理
from langchain.agents import create_openai_functions_agent
from langchain.agents import AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
# 增强版 System Prompt
system_prompt = """你是一个技术问答助手，拥有以下能力：
1. **知识库检索**：查询内部技术文档和教程
2. **网页搜索**：获取互联网上的最新信息
3. **Python 执行**：运行代码进行计算和分析
4. **数据库查询**：查询统计数据
工作原则：
- 技术概念和教程优先使用知识库，实时信息使用网页搜索
- 需要计算时使用 Python，不要手动心算
- 如果知识库找不到答案，用网页搜索补充
- 每一步都向用户说明你在做什么
- 如果工具返回错误，换一种方式重试，最多 2 次"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = create_openai_functions_agent(llm, all_tools, prompt)
executor = AgentExecutor(
agent=agent,
tools=all_tools,
verbose=True,
max_iterations=8,
handle_parsing_errors=True,
return_intermediate_steps=True
)
# === 多轮对话示例 ===
conversations = [
"LangChain 中 LCEL 是什么？和传统 Chain 有什么区别？",
"用 Python 计算一下：如果每天写 500 行代码，一年能写多少行？",
"搜索一下今天 AI 领域有什么大新闻",
]
for query in conversations:
print(f"\n{'='*60}")
print(f"👤 用户: {query}")
result = executor.invoke({"input": query})
print(f"🤖 助手: {result['output']}")
六、进阶：LangGraph 多 Agent 协作模式
6.1 为什么需要多 Agent
单个 Agent 面对复杂任务时，容易陷入"什么都会但什么都不精"的困境——工具太多导致选择困难，Prompt 太长降低准确率，错误容易级联放大。
多 Agent 架构的核心思想是分工协作：每个 Agent 只负责自己擅长的子任务，通过消息传递协调工作。
6.2 LangGraph 基础
LangGraph 是 LangChain 生态中的状态图编排框架，它把 Agent 看作图中的节点，把消息传递看作边，可以灵活地定义 Agent 之间的控制流。
from typing import TypedDict, Annotated, Sequence
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
# === 定义全局状态 ===
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
next_agent: str
task_result: dict
# === 定义各 Agent 节点 ===
def researcher_node(state: AgentState) -> AgentState:
"""研究员 Agent：负责信息检索"""
researcher_prompt = ChatPromptTemplate.from_messages([
("system", "你是研究员。搜索并整理与任务相关的信息。"),
MessagesPlaceholder(variable_name="messages"),
])
researcher_chain = researcher_prompt | llm.bind_tools([web_search, search_knowledge_base])
response = researcher_chain.invoke({"messages": state["messages"]})
return {"messages": [response], "next_agent": "analyst"}
def analyst_node(state: AgentState) -> AgentState:
"""分析师 Agent：负责数据分析和计算"""
analyst_prompt = ChatPromptTemplate.from_messages([
("system", "你是数据分析师。基于研究员提供的信息，进行深度分析和计算。"),
MessagesPlaceholder(variable_name="messages"),
])
analyst_chain = analyst_prompt | llm.bind_tools([run_python, query_stats])
response = analyst_chain.invoke({"messages": state["messages"]})
return {"messages": [response], "next_agent": "writer"}
def writer_node(state: AgentState) -> AgentState:
"""写手 Agent：负责整合输出"""
writer_prompt = ChatPromptTemplate.from_messages([
("system", """你是技术写手。基于前面的研究和分析，撰写一份结构清晰的技术报告。
格式要求：标题 → 摘要 → 详细分析 → 结论。"""),
MessagesPlaceholder(variable_name="messages"),
])
response = llm.invoke(writer_prompt.format_messages(messages=state["messages"]))
return {"messages": [response], "next_agent": END}
# === 构建状态图 ===
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("researcher", researcher_node)
workflow.add_node("analyst", analyst_node)
workflow.add_node("writer", writer_node)
# 定义边：线性流水线
workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "analyst")
workflow.add_edge("analyst", "writer")
workflow.add_edge("writer", END)
# 编译运行
app = workflow.compile()
result = app.invoke({
"messages": [
HumanMessage(content="分析 Rust 语言在 2025-2026 年的发展趋势，包括社区增长、企业采用率和主要用例变化")
],
"next_agent": "researcher",
"task_result": {}
})
print(result["messages"][-1].content)
6.3 条件路由：智能任务分发
线性流水线适合固定流程，但更灵活的方案是让"调度员"根据任务类型动态路由：
def supervisor_node(state: AgentState) -> AgentState:
"""调度员：分析任务并决定下一步交给谁"""
supervisor_prompt = ChatPromptTemplate.from_messages([
("system", """你是任务调度员。分析当前任务，决定下一步执行者：
- 需要搜索/检索 → researcher
- 需要计算/分析 → analyst
- 需要综合输出 → writer
- 任务完成 → FINISH
只回复角色名：researcher / analyst / writer / FINISH"""),
MessagesPlaceholder(variable_name="messages"),
])
response = llm.invoke(supervisor_prompt.format_messages(messages=state["messages"]))
next_agent = response.content.strip()
return {"messages": [response], "next_agent": next_agent}
# 重新构建图（条件路由版）
workflow_v2 = StateGraph(AgentState)
workflow_v2.add_node("supervisor", supervisor_node)
workflow_v2.add_node("researcher", researcher_node)
workflow_v2.add_node("analyst", analyst_node)
workflow_v2.add_node("writer", writer_node)
workflow_v2.set_entry_point("supervisor")
# 条件边：根据 supervisor 决策路由
def route_next(state: AgentState) -> str:
return state["next_agent"] if state["next_agent"] != "FINISH" else END
workflow_v2.add_conditional_edges("supervisor", route_next, {
"researcher": "researcher",
"analyst": "analyst",
"writer": "writer",
END: END
})
# 所有 worker 完成后回到 supervisor
workflow_v2.add_edge("researcher", "supervisor")
workflow_v2.add_edge("analyst", "supervisor")
workflow_v2.add_edge("writer", "supervisor")
app_v2 = workflow_v2.compile()
6.4 人机协作（Human-in-the-Loop）
生产环境中，纯自动化的 Agent 流程有时需要人类介入——比如涉及敏感操作（发送邮件、修改数据库）时需要人类审批。LangGraph 原生支持打断点：
from langgraph.checkpoint.sqlite import SqliteSaver
# 使用 SQLite 持久化状态（支持暂停/恢复）
memory = SqliteSaver.from_conn_string("checkpoints.db")
workflow_with_human = StateGraph(AgentState)
# ... 添加节点 ...
# 在关键节点前设置打断点
workflow_with_human.add_node("draft_email", draft_email_node)
workflow_with_human.add_node("human_approval", human_approval_node)
workflow_with_human.add_node("send_email", send_email_node)
workflow_with_human.add_edge("draft_email", "human_approval")
# 在 human_approval 前打断
app_with_human = workflow_with_human.compile(
checkpointer=memory,
interrupt_before=["human_approval"] # 🔴 在此节点前暂停
)
# 第一次运行：会在 human_approval 前暂停
config = {"configurable": {"thread_id": "user-123"}}
for event in app_with_human.stream(input_data, config):
print(event)
# 获取当前状态，供人类审查
state = app_with_human.get_state(config)
print(f"待审批内容: {state.values}")
# 人类审查通过后，继续执行
app_with_human.update_state(config, {"approved": True})
for event in app_with_human.stream(None, config): # None = 无新输入，继续
print(event)
6.5 并行执行：多路搜索 + 结果聚合
当任务可以拆分为独立并行子任务时，LangGraph 的 Send API 可以同时派发多个节点：
from langgraph.graph import StateGraph, END
from langgraph.constants import Send
from typing import List, TypedDict
import operator
class ParallelState(TypedDict):
queries: List[str] # 需要搜索的子问题列表
search_results: List[str] # 聚合后的搜索结果
final_answer: str
def split_to_queries(state: ParallelState):
"""将复杂问题拆分为多个子查询"""
# 这里用 LLM 拆解，示例中直接硬编码
return {"queries": [
"LangChain 最新版本特性",
"LangGraph 多 Agent 模式",
"LangSmith 监控最佳实践"
]}
def search_single_query(state: dict):
"""搜索单个子查询"""
query = state["query"]
result = web_search(query) # 假设 web_search 是已定义的工具
return {"search_results": [result]}
def merge_results(state: ParallelState):
"""合并所有搜索结果并生成最终答案"""
all_results = "\n\n".join(state["search_results"])
final = llm.invoke(f"基于以下搜索结果回答: {all_results}")
return {"final_answer": final.content}
# 关键：用 Send 并行派发
def continue_to_searches(state: ParallelState):
"""为每个子查询创建一个 Send"""
return [Send("search", {"query": q}) for q in state["queries"]]
workflow_parallel = StateGraph(ParallelState)
workflow_parallel.add_node("splitter", split_to_queries)
workflow_parallel.add_node("search", search_single_query)
workflow_parallel.add_node("merger", merge_results)
workflow_parallel.set_entry_point("splitter")
workflow_parallel.add_conditional_edges("splitter", continue_to_searches)
workflow_parallel.add_edge("search", "merger")
workflow_parallel.add_edge("merger", END)
app_parallel = workflow_parallel.compile()
# 3 个子查询并行搜索，总耗时 ≈ 最慢的那个
result = app_parallel.invoke({})
这个模式的威力在于：子查询数量和复杂度可以动态变化，LangGraph 自动管理并行执行。对于"比较 A、B、C 三个框架"这类任务，并行搜索比串行快 2-3 倍。
6.6 多 Agent 架构选型指南
七、生产部署与性能优化
7.1 流式输出
生产环境必须支持流式输出，让用户看到逐字生成的响应：
# Agent 流式输出
async for chunk in executor.astream_events(
{"input": "搜索 LangChain 最新版本"},
version="v2"
):
kind = chunk["event"]
if kind == "on_chat_model_stream":
content = chunk["data"]["chunk"].content
if content:
print(content, end="", flush=True) # 实时输出
# LangGraph 流式输出
async for event in app.astream_events(
{"messages": [HumanMessage(content="...")]},
version="v2"
):
# event 包含节点名称、输入输出、中间步骤
print(f"节点: {event.get('name')}, 事件: {event['event']}")
7.2 缓存策略
from langchain_core.caches import InMemoryCache
from langchain_core.globals import set_llm_cache
import hashlib
# LLM 响应缓存
set_llm_cache(InMemoryCache())
# 自定义检索结果缓存
from functools import lru_cache
class CachedRetriever:
def __init__(self, retriever, maxsize=128):
self.retriever = retriever
self._cache = {}
def invoke(self, query: str):
cache_key = hashlib.md5(query.encode()).hexdigest()
if cache_key in self._cache:
return self._cache[cache_key]
result = self.retriever.invoke(query)
self._cache[cache_key] = result
return result
7.3 错误处理与降级
from tenacity import retry, stop_after_attempt, wait_exponential
class RobustAgent:
def __init__(self, executor, fallback_chain=None):
self.executor = executor
self.fallback_chain = fallback_chain
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def invoke(self, input_text: str) -> str:
try:
result = self.executor.invoke({"input": input_text})
return result["output"]
except Exception as e:
print(f"Agent 执行失败 (尝试降级): {e}")
if self.fallback_chain:
return self.fallback_chain.invoke(input_text)
raise
# 降级链：不依赖工具，纯 LLM 回答
fallback_prompt = ChatPromptTemplate.from_messages([
("system", "你是技术助手。在不使用外部工具的情况下尽力回答。如果不确定，请诚实说明。"),
("human", "{input}")
])
fallback_chain = fallback_prompt | llm | StrOutputParser()
robust_agent = RobustAgent(executor, fallback_chain)
7.4 可观测性：LangSmith 集成
追踪 Agent 的每一步行为是生产运维的基本要求。LangSmith 提供了开箱即用的解决方案：
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your_key_here"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-agent-prod"
# 此后所有 LangChain/LangGraph 调用自动上报 trace
# 在 app.langchain.com 查看：
# - 每次调用的完整链路（LLM → Retriever → Tool）
# - 每步的输入/输出/耗时/Token 消耗
# - Agent 推理轨迹（Thought → Action → Observation）
# - 错误堆栈和重试记录
# 也可以手动添加自定义 trace
from langsmith import traceable
@traceable(name="my_custom_rag_step")
def custom_rag_query(query: str) -> str:
# 你的自定义逻辑
return result
7.5 成本优化：选择合适的模型组合
Agent 系统中最大的成本来自 LLM 调用。一个优化策略是大小模型混合：
# 复杂推理（Agent 决策）用大模型
agent_llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 简单任务（检索问题变体生成、文档过滤）用小模型
light_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 在 Multi-Query Retriever 中使用小模型
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=base_retriever,
llm=light_llm, # 变体生成不需要 4o
include_original=True
)
# 在 Contextual Compression 中也用小模型
compressor = LLMChainExtractor.from_llm(light_llm)
月成本估算（10万次查询/月）：
| 策略 | 模型组合 | 月成本 | 质量 |
|---|---|---|---|
| 全量 gpt-4o | 4o × 所有步骤 | ~$2000 | ⭐⭐⭐⭐⭐ |
| 混合策略 | 4o × 决策 + mini × 检索 | ~$600 | ⭐⭐⭐⭐ |
| 全量 mini | mini × 所有步骤 | ~$200 | ⭐⭐⭐ |
| 混合 + 缓存 | 同上 + 热点缓存 | ~$400 | ⭐⭐⭐⭐ |
7.6 性能基准参考
| 指标 | 基础 RAG | Multi-Query RAG | Agent (3 tools) | Agent (6+ tools) |
|---|---|---|---|---|
| 首次响应延迟 | 1.2s | 3.8s | 2.5s | 4.2s |
| 完整响应时间 | 3.5s | 8.2s | 6.0s | 15.0s |
| LLM 调用次数 | 1 | 5 (1 生成 + 4 检索) | 2-4 | 4-10 |
| Token 消耗 | ~800 | ~3200 | ~2000 | ~6000 |
| 推荐模型 | gpt-4o-mini | gpt-4o | gpt-4o | gpt-4o |
八、常见问题 FAQ
Q1：Multi-Query RAG 变体数量设多少最合适？
A：根据经验，3-5 个变体是最佳平衡点。少于 3 个和直接用原问题差别不大，多于 5 个收益递减明显，但成本线性增长。如果你的领域术语变体特别多（如医学术语、法律条文），可以适当增加到 7 个。
Q2：Parent-Document RAG 的父子块大小怎么定？
A：子块建议 200-400 tokens（保证检索精度），父块建议 1500-3000 tokens（保证上下文完整性）。可以用以下经验公式：父块大小 ≈ 子块大小 × 8。如果文档本身短小（如 FAQ 条目），直接用全文作文档即可。
Q3：Functions Agent 和 ReAct Agent 怎么选？
A：如果你用 OpenAI/Azure 的模型，无脑选 Functions Agent——更稳定、更快速、支持并行工具调用。如果你用开源模型（Llama、Qwen 等），且该模型不支持原生 Function Calling，那就用 ReAct Agent，配合 few-shot examples 提高解析成功率。
Q4：Agent 老是调用错误的工具怎么办？
A：三个优化方向：(1) 工具描述写得更精确——不仅描述功能，还要说明"什么情况下用、什么情况下不用"；(2) 给 Agent 添加 few-shot examples 展示正确用法；(3) 在 System Prompt 中明确工具选择规则。如果还不行，考虑拆分成多个专业 Agent，缩小每个 Agent 的工具集。
Q5：LangGraph 和 LangChain Agent 的关系是什么？
A：LangChain Agent 是"单个 Agent + 工具"的封装，适合相对简单的问答任务。LangGraph 是更底层的状态图框架，你可以用它实现任意复杂的多 Agent 协作模式（循环、条件分支、人机协作）。可以理解为：Agent 是 LangGraph 的一个特例——单节点 + 工具循环。
Q6：生产环境中怎么监控 Agent 的行为？
A：推荐 LangSmith（LangChain 官方的可观测性平台），设置 LANGCHAIN_TRACING_V2=true 即可自动记录每次调用链路的完整 trace。也可以自建监控：在每个工具调用前后打日志，记录工具名、输入参数、执行耗时、返回结果。特别关注 max_iterations 触发的截断——这说明 Agent 陷入了循环。
Q7：上下文窗口总是不够用怎么办？
A：(1) 优先使用 Contextual Compression，在检索后就压缩无关内容；(2) 使用更小的 Embedding 模型（如 text-embedding-3-small，性价比高）；(3) 对历史对话做摘要压缩（Summarization Memory），而不是保留原始消息；(4) 如果用的是 ChatGPT，尽量用 gpt-4o（128K 上下文），而非 gpt-4o-mini。
Q8：如何确保 Agent 不会陷入无限循环？
A：这是 Agent 部署中最常见的坑。三道防线：(1) max_iterations 硬限制——设置一个合理的上限（通常 8-15 次），超过后强制终止并返回部分结果；(2) 循环检测——记录最近 N 次 Action 的哈希值，如果出现重复模式（连续 3 次相同 Action + Input），主动打断；(3) 渐进式信心衰减——在 System Prompt 中加入"如果连续 2 次尝试失败，请降级为不使用工具直接回答"的指令。LangSmith 中有一个专门的 loop_detected 事件可以订阅。
Q9：多 Agent 架构 vs 单 Agent + 多工具，什么时候该升级？
A：当出现以下任一信号时，就该考虑多 Agent 架构：(1) 工具集超过 8-10 个——单个 Agent 的选择困难导致频繁误用工具；(2) System Prompt 超过 500 字——提示词太长会稀释关键指令；(3) 不同任务需要的推理风格差异大——比如数据分析需要严谨性、客服需要亲和力，共用一个 System Prompt 难以兼顾；(4) 需要并行的子任务——比如同时在不同数据源搜索，单 Agent 只能串行。反之，如果工具少于 5 个、任务类型单一，单 Agent + 多工具就是最简方案。
Q10：RAG 系统的冷启动问题怎么解决——新知识库没有任何数据？
A：冷启动是每个 RAG 项目的第一道坎。建议：(1) 种子文档策略——手动准备 20-30 篇高质量 FAQ 或文档作为种子，不求多但求精；(2) 从用户日志中挖掘——如果有旧系统（如客服工单、论坛帖子），用 LLM 把"用户问题 + 最佳回复"转成 Q&A 文档对；(3) 渐进式上线——先在小范围（如内部团队）跑 1-2 周，收集 feedback loop，把用户问过但系统答不好的问题手工补充进知识库；(4) 不要一上来就追求全自动——第一版可以先做"检索推荐"模式，让人类审核检索结果后再生成，既保证质量又积累训练数据。
九、总结与展望
9.1 核心要点回顾
本文从 RAG 的四大痛点出发，系统性地介绍了 LangChain 生态中的高级解决方案：
- Multi-Query RAG 解决了"问题表述模糊"的问题——用 LLM 生成多角度变体查询，提高检索召回率
- Parent-Document RAG 解决了"上下文不完整"的问题——小块检索保精度，大块返回保上下文
- Contextual Compression 解决了"上下文窗口浪费"的问题——检索后自动提纯，只保留相关句子
- Self-Query RAG 解决了"元数据无法利用"的问题——让 LLM 自动将自然语言转成结构化过滤条件
- 工具增强 Agent 让 AI 从"只读"进化为"可执行"——搜索、计算、查询、操作，无所不能
- LangGraph 多 Agent 通过分工协作解决复杂任务——每个 Agent 专注自己的领域，通过状态图灵活编排
9.2 技能路径图
入门：基础 RAG（向量库 + LLM）
↓
进阶：高级 RAG 模式（Multi-Query, Parent-Doc, Compression）
↓
高级：工具增强 Agent（ReAct, Function Calling）
↓
专家：多 Agent 协作（LangGraph, 条件路由, 人机协作）
↓
大师：生产级部署（流式、缓存、监控、A/B 测试）
9.3 下一步行动
如果你想继续深入，建议按以下顺序学习：
- 📖 阅读 LangChain 官方文档的 RAG 和 Agents 教程
- 🔧 用本文的代码模板搭建一个自己的多策略 RAG 系统，替换为你实际业务的知识库
- 🧪 尝试 LangGraph 的 Multi-Agent Supervisor 官方示例
- 📊 用 LangSmith 监控你的 Agent 调用链路，观察哪些工具用得最多、哪些步骤耗时最长
- 🚀 将你的 RAG + Agent 系统部署到生产环境，收集真实用户反馈迭代优化
本文由 MarkShareX AI 自动创作，分类：Agent，方向：LangChain 实战（高级 RAG 与工具增强 Agent）