LangChain 工具调用与 RAG 深度融合实战：打造智能知识型 Agent
当 LLM 不再只会「聊天」，而是能够主动查资料、调工具、做决策，AI Agent 的威力才真正被释放。本文将带你从工具调用机制到 RAG 深度融合，构建一个能搜、能算、能调用外部 API 的生产级智能 Agent。
目录
- 背景与概念：为什么 Tool Calling + RAG 是 Agent 的核心
- 核心知识：LangChain Tool Calling 机制详解
- 核心知识：RAG 原理与检索优化策略
- 实战演练：构建知识型 Agent
- 进阶技巧：多工具编排与智能路由
- 进阶技巧：混合检索与重排序实战
- 生产级考虑：性能、成本与安全
- 常见问题 FAQ
- 总结与展望
一、背景与概念：为什么 Tool Calling + RAG 是 Agent 的核心
2024 年以来，LLM 的能力边界已经从「纯文本生成」扩展到了「具身智能」。GPT-4 的函数调用（Function Calling）、Claude 的工具使用（Tool Use）、以及开源模型通过格式约束实现的工具调用能力，让大语言模型从「聊天机器人」进化为「智能代理」（Agent）。
但工具调用只是硬币的一面。LLM 受限于训练数据的截止时间和参数化的知识存储方式，在处理实时信息、私有数据、精确事实时往往力不从心。这就是 RAG（Retrieval-Augmented Generation，检索增强生成） 的价值所在——它让 Agent 在需要时「翻阅资料」，用外部知识弥补模型的短板。
为什么两者必须结合？
单一的 RAG 系统只能做「问答」——用户提问，系统检索，模型回答。它无法胜任需要多步推理、跨系统交互的复杂任务。比如：
- 「帮我查一下上周的销售数据，和上个月同期做个对比，然后把结果发到 Slack 通知团队」——这需要数据库查询工具 + 计算工具 + 消息工具 + RAG 检索
- 「阅读这份合同 PDF，找出违约条款，对照《民法典》相关法条，给出修改建议」——这需要文件工具 + RAG 知识库 + 推理链
Tool Calling 赋予 Agent 行动能力，RAG 赋予 Agent 知识底座。两者的深度融合，才是构建实用 AI Agent 的正解。
本文主要内容
我们将以 LangChain 框架为核心，从零构建一个「知识型 Agent」：
- 理解 LangChain 的工具系统设计哲学（@tool 装饰器、StructuredTool、BaseTool）
- 深入 AgentExecutor 的推理-行动循环（ReAct / Tool Calling Agent）
- 实现向量化 RAG 引擎（ChromaDB + 嵌入模型 + 多策略检索）
- 将 RAG 与工具调用深度整合：把「检索」本身变成一个 Tool
- 进阶：多工具编排、智能路由、混合检索、重排序
- 生产部署：流式输出、缓存优化、Token 成本控制、安全防护
在开始之前，请确保你已经安装了以下依赖：
pip install langchain langchain-openai langchain-community chromadb tiktoken pypdf
二、核心知识：LangChain Tool Calling 机制详解
2.1 Tool 的本质：函数的「结构化描述」
在 LangChain 中，一个 Tool 本质上是对 Python 函数的封装，附加了 LLM 可理解的描述信息。当 Agent 决定使用某个工具时，它实际上是在生成一个结构化的函数调用请求。
LangChain 提供了三种定义工具的方式：
方式一：@tool 装饰器（最简洁）
from langchain.tools import tool
@tool
def search_knowledge_base(query: str) -> str:
"""
在内部知识库中搜索相关信息。
当用户询问公司政策、产品文档、历史记录时使用此工具。
Args:
query: 搜索查询字符串
"""
# 实际检索逻辑
results = vector_store.similarity_search(query, k=5)
return "\n\n".join([doc.page_content for doc in results])
@tool
def calculate(expression: str) -> str:
"""
执行数学计算。支持加减乘除、幂运算、三角函数等。
当需要进行精确数值计算时使用此工具。
Args:
expression: 数学表达式，如 '2 + 3 * 4'
"""
import math
try:
result = eval(expression, {"__builtins__": {}}, {"math": math})
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
方式二：StructuredTool（带结构化参数）
from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
class WeatherInput(BaseModel):
city: str = Field(description="城市名称，如 '北京'、'上海'")
date: str = Field(default="today", description="日期，格式 YYYY-MM-DD")
def get_weather(city: str, date: str = "today") -> str:
"""获取指定城市和日期的天气信息"""
# 调用天气 API
return f"{city} {date} 天气：晴，22°C~30°C"
weather_tool = StructuredTool.from_function(
func=get_weather,
name="get_weather",
description="获取指定城市的天气信息",
args_schema=WeatherInput
)
方式三：BaseTool 子类（最灵活）
from langchain.tools import BaseTool
from typing import Optional, Type
class DatabaseQueryTool(BaseTool):
name: str = "database_query"
description: str = """
执行 SQL 查询。仅支持 SELECT 语句。
表结构：orders(id, customer, amount, date), products(id, name, price)
"""
args_schema: Type[BaseModel] = DatabaseQueryInput
def _run(self, sql: str, limit: int = 10) -> str:
# 安全检查：只允许 SELECT
if not sql.strip().upper().startswith("SELECT"):
return "错误：仅允许 SELECT 查询"
# 执行查询
return execute_sql(sql, limit)
2.2 Tool Calling 的底层流程
当 Agent 接到用户请求后，LangChain 的 AgentExecutor 进入一个推理-行动循环（Reasoning-Action Loop）：
步骤详解：
- 构建 Prompt：将系统提示、可用工具列表（含名称、描述、参数 schema）、对话历史一并组装成 prompt
- LLM 推理：模型分析用户意图，决定「直接回答」还是「调用工具」
- 解析输出：如果是工具调用，LangChain 解析模型输出的结构化参数（JSON/function call）
- 执行工具：调用对应的 Python 函数，获取结果
- 反馈观察：工具执行结果作为新的「观察」（Observation）注入上下文
- 循环判断：Agent 判断是否还需要更多信息，如果是则回到步骤 1；否则生成最终回答
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# 初始化 LLM 和工具
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_knowledge_base, calculate, weather_tool]
# 创建 Agent
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个智能助手，可以搜索知识库、计算和查天气。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5, # 最多循环 5 次
handle_parsing_errors=True,
)
# 执行
result = executor.invoke({
"input": "北京的天气怎么样？如果温度超过25度就帮我算一下空调一天的电费"
})
2.3 关键设计决策：工具粒度与描述策略
工具设计有两个核心原则：
原则一：单一职责 每个工具只做一件事。避免「万能工具」——它会让 LLM 困惑。
# ❌ 不好的设计：一个工具做太多事
@tool
def do_everything(action: str, data: str) -> str:
"""可以搜索、计算、发邮件、查数据库"""
...
# ✅ 好的设计：原子化工具
@tool
def search_docs(query: str) -> str: ...
@tool
def send_email(to: str, subject: str, body: str) -> str: ...
@tool
def query_database(sql: str) -> str: ...
原则二：描述即文档
LLM 完全依赖工具的 description 字段来理解何时调用。描述应该：
- 说明什么时候使用这个工具
- 说明什么时候不该用
- 包含参数含义和返回值格式
@tool
def get_stock_price(symbol: str) -> str:
"""
获取股票实时价格。
使用场景：
- 用户询问某只股票的当前价格
- 需要计算投资组合价值时
不适用场景：
- 历史价格走势（请使用 get_stock_history）
- 公司基本面分析（请使用 get_company_info）
Args:
symbol: 股票代码，如 AAPL、TSLA、600519.SH
Returns:
JSON 格式：{"symbol": "...", "price": ..., "change": "..."}
"""
三、核心知识：RAG 原理与检索优化策略
3.1 RAG 的工作流程
RAG 的核心思想是「先检索，再生成」。整个流程分为离线（索引构建）和在线（查询处理）两个阶段：
离线阶段：
- 文档加载：从 PDF、网页、数据库等来源读取文档
- 文本分割：将长文档切分为适当大小的「块」（chunk）
- 向量化：使用嵌入模型（如 text-embedding-3-small）将每个 chunk 转为向量
- 索引存储：将向量存入向量数据库（如 ChromaDB、Pinecone、Qdrant）
在线阶段：
- 查询改写：对用户问题进行优化（扩展、纠错、多角度重写）
- 向量检索：将改写后的查询转为向量，在向量数据库中搜索相似 chunk
- 重排序：对检索结果进行二次排序，提高相关性
- 上下文组装：将检索到的文档片段拼接进 LLM 的 prompt
- 生成回答：LLM 综合上下文和用户问题生成最终答案
3.2 向量存储与嵌入模型选型
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, PyPDFLoader
# 1. 加载文档
loader = PyPDFLoader("company_policy.pdf")
documents = loader.load()
# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每块 1000 字符
chunk_overlap=200, # 块间重叠 200 字符
separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""],
)
chunks = text_splitter.split_documents(documents)
print(f"共生成 {len(chunks)} 个文本块")
# 3. 向量化 + 存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db",
)
嵌入模型对比：
| 模型 | 维度 | 最大输入 | 价格(每1M token) | 适合场景 |
|---|---|---|---|---|
| text-embedding-3-small | 512 | 8191 | $0.02 | 通用检索，成本敏感 |
| text-embedding-3-large | 3072 | 8191 | $0.13 | 高精度语义匹配 |
| bge-large-zh-v1.5 | 1024 | 512 | 免费（本地） | 中文场景 |
| m3e-base | 768 | 512 | 免费（本地） | 中文，轻量级 |
3.3 关键词检索的补充价值
纯向量检索虽然强大，但有其固有的盲区。嵌入模型将文本映射为语义向量，在这个过程中会丢失精确的关键词匹配能力。例如，用户搜索「SKU-2024-Q3-88492」时，向量检索可能返回包含「SKU-2024-Q2-88493」的文档——语义接近但并非用户要的产品编号。
这就是为什么生产级 RAG 系统通常采用混合检索策略——同时运行向量检索和关键词检索（如 BM25），然后通过融合算法合并结果。BM25 是经典的 TF-IDF 变体，擅长精确字符串匹配和罕见词检索。LangChain 的 EnsembleRetriever 可以无缝整合两种检索器。
在实践中，推荐的初始权重配比是向量检索 0.7、BM25 0.3。对于包含大量技术编号、API 端点、错误码的知识库，可以适当提高 BM25 的权重到 0.4-0.5。
3.4 Chunk 策略：RAG 性能的关键杠杆
Chunk 的大小和重叠度直接影响检索质量：
| 策略 | Chunk 大小 | 重叠 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|---|---|
| 固定大小 | 500 | 100 | 通用场景 | 简单可靠 | 可能截断语义 |
| 语义分割 | 按段落 | 0 | 结构化文档 | 保留语义完整 | 依赖文档格式 |
| 小 Chunk | 256 | 64 | 精准问答 | 精确定位 | 上下文不足 |
| 大 Chunk | 2000 | 400 | 复杂推理 | 丰富上下文 | 噪声增加 |
实践建议：对同一份文档，使用多种 chunk 策略生成不同粒度的索引（小块用于精准检索，大块用于上下文补充），即「父子文档」（Parent Document）模式。
# 父子文档模式
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
# 子文档（用于检索）
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
# 父文档（用于上下文）
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
store = InMemoryStore()
retriever = ParentDocumentRetriever(
vectorstore=vector_store,
docstore=store,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
四、实战演练：构建知识型 Agent
现在我们将 Tool Calling 和 RAG 结合，构建一个完整的知识型 Agent。这个 Agent 可以：
- 在本地知识库中搜索信息（RAG）
- 调用外部 API（天气、股价）
- 进行数学计算
- 根据检索结果做决策
4.1 项目结构
knowledge_agent/
├── main.py # Agent 入口
├── tools/
│ ├── __init__.py
│ ├── search_tool.py # RAG 检索工具
│ ├── api_tools.py # 外部 API 工具
│ └── calc_tool.py # 计算工具
├── rag/
│ ├── loader.py # 文档加载
│ ├── indexer.py # 向量索引
│ └── retriever.py # 检索器
└── config.py # 配置文件
4.2 将 RAG 封装为 Tool
这是深度融合的关键步骤——把检索能力变成一个标准 Tool：
# tools/search_tool.py
from langchain.tools import tool
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from typing import Optional
# 全局变量（生产环境用依赖注入）
_vector_store = None
_llm = None
def init_search_tool(persist_dir: str = "./chroma_db"):
"""初始化检索工具"""
global _vector_store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
_vector_store = Chroma(
persist_directory=persist_dir,
embedding_function=embeddings,
)
@tool
def search_knowledge_base(query: str, k: int = 5) -> str:
"""
在内部知识库中搜索信息。用于回答公司政策、产品文档、技术规范等问题。
检索策略：
- 对用户查询自动进行同义词扩展
- 使用 MMR（最大边际相关性）算法去重
- 返回 Top-K 最相关文档片段
Args:
query: 搜索查询
k: 返回结果数量（默认 5）
"""
if _vector_store is None:
return "错误：知识库未初始化"
# 使用 MMR 检索（兼顾相关性和多样性）
results = _vector_store.max_marginal_relevance_search(
query, k=k, fetch_k=20, lambda_mult=0.7
)
if not results:
return "未找到相关信息。建议尝试不同的关键词。"
formatted = []
for i, doc in enumerate(results, 1):
source = doc.metadata.get("source", "未知来源")
formatted.append(f"[{i}] (来源: {source})\n{doc.page_content}")
return "\n\n---\n\n".join(formatted)
4.3 创建外部 API 工具
# tools/api_tools.py
from langchain.tools import tool
import requests
from datetime import datetime
@tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
"""获取当前日期和时间"""
now = datetime.now()
return f"当前时间: {now.strftime('%Y-%m-%d %H:%M:%S')} (时区: {timezone})"
@tool
def web_search_tool(query: str) -> str:
"""
搜索互联网获取最新信息。
适用于知识库中没有的实时信息。
Args:
query: 搜索查询
"""
# 这里接入实际的搜索 API（如 Tavily、SerpAPI）
return f"[模拟搜索结果] 关于 '{query}' 的最新信息..."
4.4 组装 Agent
# main.py
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from tools.search_tool import search_knowledge_base, init_search_tool
from tools.api_tools import get_current_time, web_search_tool
from tools.calc_tool import calculate
# 初始化
init_search_tool("./chroma_db")
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [
search_knowledge_base,
get_current_time,
web_search_tool,
calculate,
]
# 精心设计的系统提示
system_prompt = """你是一个知识型 AI Agent，具备以下能力：
1. **知识检索**：可以搜索内部知识库获取公司政策、产品文档等信息
2. **联网搜索**：对于知识库中没有的实时信息，可以搜索互联网
3. **数学计算**：可以进行精确的数值计算
4. **时间感知**：知道当前日期和时间
工作原则：
- 优先使用内部知识库，找不到再联网搜索
- 当用户问题涉及多个步骤时，逐步执行
- 计算结果要给出推导过程
- 如果信息不足以回答问题，诚实说明"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=8,
handle_parsing_errors=True,
return_intermediate_steps=True,
)
# 测试运行
if __name__ == "__main__":
response = executor.invoke({
"input": "公司今年的休假政策是什么？帮我算一下如果我有15天年假，平均每个月可以休几天？"
})
print(f"\n最终回答:\n{response['output']}")
4.5 运行轨迹分析
让我们看看一个典型的运行轨迹：
> Entering new AgentExecutor chain...
Invoking: search_knowledge_base with {'query': '公司休假政策 年假'}
[返回 3 条文档片段]
Invoking: calculate with {'expression': '15 / 12'}
计算得到: 1.25
根据公司休假政策，员工每年享有 15 天带薪年假...
平均到每个月，您可以休 15/12 = 1.25 天...
这个例子展示了 Agent 如何自主决策工具调用顺序：先检索政策文档获取事实依据，再调用计算工具进行数值推导，最后综合两个结果生成回答。
五、进阶技巧：多工具编排与智能路由
当一个 Agent 拥有越来越多的工具时，简单地把所有工具都扔给 LLM 会带来两个问题：
- 性能下降：prompt 中工具描述越多，推理越慢，Token 消耗越大
- 选择失误：工具越多，LLM 选错工具的概率越高
解决方案是工具路由（Tool Routing）——根据用户意图动态选择工具子集。
5.1 意图识别 + 工具分组
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
class Intent(BaseModel):
"""用户意图分类"""
category: Literal["knowledge_search", "calculation", "api_call", "general"]
confidence: float = Field(ge=0, le=1)
# 意图识别 prompt
intent_prompt = ChatPromptTemplate.from_messages([
("system", """分析用户消息，判断意图类别：
- knowledge_search: 需要查询知识库、文档、政策
- calculation: 需要数学计算、数据分析
- api_call: 需要调用外部 API（天气、股价等）
- general: 普通对话，不需要工具
返回 JSON 格式。"""),
("human", "{input}"),
])
# 带结构化输出的 LLM
intent_chain = intent_prompt | llm.with_structured_output(Intent)
# 工具分组
tool_groups = {
"knowledge_search": [search_knowledge_base, web_search_tool],
"calculation": [calculate],
"api_call": [get_current_time, weather_tool, get_stock_price],
}
def select_tools(user_input: str):
"""根据意图选择工具子集"""
intent = intent_chain.invoke({"input": user_input})
print(f"识别意图: {intent.category} (置信度: {intent.confidence})")
selected = tool_groups.get(intent.category, [])
# 始终包含通用工具
selected.append(get_current_time)
return selected
5.2 多 Agent 协作模式
对于更复杂的任务，可以使用「路由 Agent + 专家 Agent」的模式：
from langchain.agents import AgentExecutor
# 创建多个专家 Agent
knowledge_agent = AgentExecutor(
agent=create_tool_calling_agent(llm, [search_knowledge_base], prompt),
tools=[search_knowledge_base],
)
calc_agent = AgentExecutor(
agent=create_tool_calling_agent(llm, [calculate], prompt),
tools=[calculate],
)
# 路由 Agent（Supervisor）
supervisor_tools = [
StructuredTool.from_function(
func=lambda query: knowledge_agent.invoke({"input": query})["output"],
name="delegate_to_knowledge_expert",
description="将知识检索类问题委派给知识专家处理"
),
StructuredTool.from_function(
func=lambda expr: calc_agent.invoke({"input": expr})["output"],
name="delegate_to_calc_expert",
description="将计算类问题委派给计算专家处理"
),
]
supervisor = AgentExecutor(
agent=create_tool_calling_agent(llm, supervisor_tools, prompt),
tools=supervisor_tools,
)
5.3 动态工具注册
生产环境中，工具可能需要动态加载（如按用户权限加载不同工具集）：
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self._tools: Dict[str, BaseTool] = {}
self._permissions: Dict[str, List[str]] = {} # 用户 -> 工具列表
def register(self, tool: BaseTool, required_role: str = "user"):
self._tools[tool.name] = tool
if required_role not in self._permissions:
self._permissions[required_role] = []
self._permissions[required_role].append(tool.name)
def get_tools_for_user(self, user_role: str) -> List[BaseTool]:
allowed = self._permissions.get(user_role, [])
return [self._tools[name] for name in allowed if name in self._tools]
# 使用
registry = ToolRegistry()
registry.register(search_knowledge_base, required_role="user")
registry.register(query_database, required_role="admin")
registry.register(send_email, required_role="user")
# 不同用户看到不同的工具集
user_tools = registry.get_tools_for_user("user") # 不含 query_database
admin_tools = registry.get_tools_for_user("admin") # 包含所有
六、进阶技巧：混合检索与重排序实战
6.1 为什么需要混合检索？
纯向量检索在某些场景下效果不佳：
- 精确关键词匹配：如产品编号 "SKU-88492"，向量检索可能返回语义相近但编号不同的结果
- 专有名词：公司内部术语、缩写，嵌入模型可能无法准确理解
- 数字/日期查询：「2024年Q3财报」——向量检索可能返回 Q2 或 Q4 的内容
混合检索 = 向量检索 + 关键词检索，取二者的交集或加权融合。
6.2 实现 BM25 + 向量检索的混合引擎
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# 1. BM25 关键词检索器（基于文档集合）
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5
# 2. 向量检索器
vector_retriever = vector_store.as_retriever(
search_type="mmr",
search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.7}
)
# 3. 集成检索器（RRF 融合）
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.3, 0.7], # BM25 权重 30%，向量检索 70%
)
6.3 重排序（Re-ranking）提升精度
检索到的文档可能相关但排序不佳。重排序模型可以对初检结果精排：
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
# 使用 Cross-Encoder 进行重排序
model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-large")
compressor = CrossEncoderReranker(model=model, top_n=5)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=ensemble_retriever,
)
# 现在检索的结果经过了 BM25 → 向量 → 重排序 三阶段过滤
results = compression_retriever.invoke("公司今年的休假政策")
6.4 查询优化：让检索更聪明
在检索前对用户查询进行处理，可以显著提升检索质量：
from langchain_core.prompts import ChatPromptTemplate
# 查询扩展：将短查询扩展为多个相关查询
query_expansion_prompt = ChatPromptTemplate.from_messages([
("system", """将用户的查询扩展为3个不同角度的搜索查询。
原始查询: {query}
输出每行一个查询，不要编号。"""),
])
def expand_query(query: str) -> list[str]:
response = llm.invoke(
query_expansion_prompt.format_messages(query=query)
)
queries = [q.strip() for q in response.content.split("\n") if q.strip()]
return queries[:3] # 最多 3 个
# 多查询融合检索
from langchain.retrievers.multi_query import MultiQueryRetriever
multi_query_retriever = MultiQueryRetriever.from_llm(
retriever=vector_store.as_retriever(),
llm=llm,
)
6.5 带来源引用的回答
高质量 RAG 的标配是回答带上来源引用：
def format_with_citations(query: str, retriever) -> str:
docs = retriever.invoke(query)
# 构建带编号的上下文
context_parts = []
sources = {}
for i, doc in enumerate(docs, 1):
context_parts.append(f"[{i}] {doc.page_content}")
sources[i] = doc.metadata.get("source", "未知")
context = "\n\n".join(context_parts)
prompt = f"""基于以下检索到的文档回答问题。在回答中引用来源编号（如 [1]、[2]）。
文档：
{context}
问题：{query}
要求：
- 每个关键事实都要标注来源编号
- 如果文档之间信息有冲突，请指出
- 如果没有足够信息，请明确说明"""
response = llm.invoke(prompt)
return response.content, sources
七、生产级考虑：性能、成本与安全
7.1 流式输出与用户体验
from langchain.callbacks import StreamingStdOutCallbackHandler
from langchain_core.runnables import RunnableConfig
async def stream_agent_response(user_input: str):
"""流式返回 Agent 响应"""
async for event in executor.astream_events(
{"input": user_input},
version="v2",
):
kind = event["event"]
if kind == "on_tool_start":
print(f"\n🔧 调用工具: {event['name']}")
elif kind == "on_tool_end":
print(f"✅ 工具完成: {event['name']}")
elif kind == "on_chat_model_stream":
content = event["data"]["chunk"].content
if content:
yield content # 实时返回给前端
7.2 Token 成本优化策略
| 策略 | 效果 | 实现难度 |
|---|---|---|
| 压缩工具描述 | 节省 20-30% 系统 prompt token | ⭐ 低 |
| 对话历史摘要 | 节省 50%+ 上下文 token | ⭐⭐ 中 |
| 工具结果截断 | 节省 30-50% 观察 token | ⭐ 低 |
| 缓存常用检索 | 减少重复 API 调用 | ⭐⭐⭐ 高 |
| 意图路由减少工具集 | 节省 40-60% 工具描述 token | ⭐⭐ 中 |
# 对话历史摘要
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(
llm=llm,
max_token_limit=500, # 摘要最大 token 数
return_messages=True,
)
# 工具结果截断
@tool
def search_with_truncation(query: str) -> str:
results = vector_store.similarity_search(query, k=10)
formatted = []
total_chars = 0
for doc in results:
chunk = doc.page_content[:500] # 每段最多 500 字符
formatted.append(chunk)
total_chars += len(chunk)
if total_chars > 3000: # 总字符数上限
formatted.append("... (更多结果已截断)")
break
return "\n\n".join(formatted)
7.3 安全防护
# SQL 注入防护
def sanitize_sql(sql: str) -> bool:
dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE", "EXEC"]
sql_upper = sql.upper()
for keyword in dangerous:
if keyword in sql_upper:
return False
return sql_upper.strip().startswith("SELECT")
# 工具执行沙箱
@tool
def safe_python_exec(code: str) -> str:
"""在受限环境中执行 Python 代码"""
allowed_builtins = {
"abs": abs, "len": len, "range": range,
"int": int, "float": float, "str": str,
"list": list, "dict": dict, "sum": sum,
"min": min, "max": max, "sorted": sorted,
}
try:
result = eval(code, {"__builtins__": allowed_builtins}, {})
return str(result)
except Exception as e:
return f"代码执行错误: {str(e)}"
# 内容过滤
def filter_sensitive_info(text: str) -> str:
"""过滤敏感信息（手机号、身份证、银行卡）"""
import re
text = re.sub(r'1[3-9]\d{9}', '[手机号已隐藏]', text)
text = re.sub(r'\d{17}[\dXx]', '[身份证号已隐藏]', text)
text = re.sub(r'\d{16,19}', '[银行卡号已隐藏]', text)
return text
7.4 监控与可观测性
在生产环境中运行 Agent，看不见的问题才是最危险的。建议对每个 Agent 请求记录以下指标：
import time
from langchain.callbacks import BaseCallbackHandler
class MetricsCallback(BaseCallbackHandler):
"""收集 Agent 运行指标"""
def __init__(self):
self.tool_calls = []
self.total_tokens = 0
self.start_time = None
def on_agent_start(self, *args, **kwargs):
self.start_time = time.time()
def on_tool_start(self, serialized, input_str, **kwargs):
self.tool_calls.append({
"tool": serialized.get("name", "unknown"),
"input": input_str,
"start": time.time(),
})
def on_tool_end(self, output, **kwargs):
if self.tool_calls:
self.tool_calls[-1]["end"] = time.time()
self.tool_calls[-1]["duration"] = (
self.tool_calls[-1]["end"] - self.tool_calls[-1]["start"]
)
def get_report(self) -> dict:
elapsed = time.time() - self.start_time if self.start_time else 0
return {
"total_duration": f"{elapsed:.2f}s",
"tool_calls": len(self.tool_calls),
"tool_details": [
f"{tc['tool']}: {tc.get('duration', 0):.2f}s"
for tc in self.tool_calls
],
}
# 使用
callback = MetricsCallback()
executor.invoke({"input": "..."}, config={"callbacks": [callback]})
print(callback.get_report())
7.5 错误处理与降级策略
Agent 在生产中最常见的失败模式是工具调用超时或返回异常。一个健壮的 Agent 应该实现优雅降级：
@tool
def resilient_search(query: str) -> str:
"""带降级的搜索工具"""
try:
# 主路径：向量检索
return vector_store.similarity_search(query, k=5)
except Exception as e:
try:
# 降级路径 1：简单关键词匹配
return keyword_search(query)
except Exception:
# 降级路径 2：返回预设的兜底回答
return "搜索服务暂时不可用，请稍后重试或尝试其他关键词。"
对于外部 API 工具，建议加上超时和重试：用 tenacity 库实现指数退避重试（首次 1s、二次 2s、三次 5s），最多重试 3 次后返回友好的降级信息。这样即使部分依赖不可用，Agent 仍能给出部分回答而非完全报错。
八、常见问题 FAQ
Q1: Tool Calling 和 Function Calling 有什么区别？
A: Tool Calling 是 LangChain 的概念，Function Calling 是 OpenAI 的原生能力。两者本质相同：让 LLM 输出结构化参数来调用外部函数。LangChain 的 Tool Calling Agent 在底层可以适配多种 LLM 的工具调用机制（OpenAI function calling、Anthropic tool use、Ollama JSON mode），提供统一的接口。
Q2: 如何选择嵌入模型？中文场景用哪个？
A: 推荐选择顺序：
- 有预算且追求精度 →
text-embedding-3-large（OpenAI） - 成本敏感但需要好效果 →
bge-large-zh-v1.5（开源，中文 SOTA） - 本地部署、轻量级 →
m3e-base或bge-small-zh-v1.5
中文场景优先选择在中文数据集上微调过的模型，如 BGE 系列、M3E 系列。
Q3: Agent 频繁调用同一个工具怎么办？
A: 这通常是工具描述不够精确导致的。解决方案：
- 设置
max_iterations限制最大循环次数 - 在
return_intermediate_steps=True模式下分析哪些步骤是冗余的 - 优化工具描述，明确「何时停止」
- 添加一个
final_answer工具让 LLM 显式声明完成
Q4: 向量数据库怎么选？ChromaDB、Pinecone、Qdrant 哪个好？
A: 取决于场景：
| 向量数据库 | 类型 | 适用规模 | 优势 |
|---|---|---|---|
| ChromaDB | 嵌入式 | <100K 向量 | 零配置，适合原型和中小项目 |
| Qdrant | 独立服务 | 100K-10M | 高性能，过滤功能强 |
| Pinecone | 云服务 | 任意规模 | 全托管，无需运维 |
| Milvus | 分布式 | 10M+ | 大规模生产环境首选 |
| FAISS | 库 | <1M | 极快，适合离线批处理 |
Q5: 如何处理文档更新？RAG 索引如何保持同步？
A: 常见策略：
- 全量重建：适合文档量小（<1000）的场景，定时全量重建索引
- 增量更新：使用文档 ID 追踪，只更新变化的文档。ChromaDB 支持
update_document - 版本管理：为索引打版本号，每次更新切换版本（蓝绿部署）
- 实时同步：通过 Webhook 或消息队列监听文档变更事件
Q6: Tool Calling 的延迟太大了，怎么优化？
A: 延迟来源主要是：LLM 推理 + 工具执行。优化方向：
- 使用更快的模型（如 GPT-4o-mini 代替 GPT-4o 做意图路由）
- 并行执行独立的工具调用（LangChain 支持
RunnableParallel） - 缓存常见的工具调用结果
- 对知识检索类工具使用更小的向量维度
Q7: 如何防止 Agent 执行危险操作？
A: 多层防护：
- 工具层：每个工具内部做参数校验和权限检查
- 执行层：设置
max_iterations、超时时间 - 内容层：敏感信息过滤、输出审计
- 人机协同：高风险操作（如发送邮件、修改数据）增加人工确认步骤
Q8: 使用 LangChain 的 Agent 和使用 LangGraph 有什么区别？
A: LangChain Agent 适合简单的「单轮推理-行动」循环，代码量少、上手快。LangGraph 提供了图结构编排能力，支持条件分支、并行执行、循环、人工介入等复杂工作流。如果你的 Agent 需要：
- 多步骤且有条件分支 → 使用 LangGraph
- 需要并行调用多个工具 → LangGraph 的
SendAPI 更方便 - 需要持久化状态和断点续跑 → LangGraph 内置
checkpointer - 简单的工具调用问答 → LangChain Agent 完全够用
建议：从 LangChain Agent 开始，当发现 AgentExecutor 难以表达你的工作流时再迁移到 LangGraph。
Q9: 本地部署 RAG 的最低硬件配置是什么？
A: 取决于嵌入模型和向量数据库：
- 嵌入模型：
bge-small-zh-v1.5（384 维）在 CPU 上即可运行，内存 < 500MB - 向量数据库：ChromaDB 在 10 万条向量以下仅需 < 2GB 内存
- LLM：7B 量化模型（如 Qwen2.5-7B-Instruct-GGUF q4）需 6-8GB 显存/内存
- 推荐最低配置：16GB RAM + 8GB VRAM 的消费级 GPU（或 Apple Silicon 16GB 统一内存）
九、总结与展望
核心要点回顾
本文从零构建了一个 LangChain 驱动的知识型 Agent，涵盖了：
- Tool Calling 机制：三种定义方式、推理-行动循环、工具设计原则
- RAG 深度整合：向量存储、Chunk 策略、父子文档模式
- 工具路由与编排：意图识别、动态工具加载、多 Agent 协作
- 混合检索优化：BM25 + 向量融合、Cross-Encoder 重排序、查询扩展
- 生产级考量：流式输出、成本优化、安全防护、可观测性
下一步学习建议
- LangGraph：LangChain 的图编排框架，适合复杂的多步骤 Agent 工作流
- CrewAI / AutoGen：多 Agent 框架，构建 Agent 团队协作
- Agent 记忆系统：长短期记忆管理，让 Agent 记住用户偏好和上下文
- RLHF 对齐：通过人类反馈强化学习优化 Agent 的行为策略
技术趋势展望
2025 年，Agent 技术正在向以下方向发展：
- MCP 协议（Model Context Protocol）：Anthropic 提出的标准化工具接口协议，让 Agent 可以即插即用地连接各种工具
- Agent-to-Agent 通信：多个 Agent 之间通过标准协议协作，形成 Agent 网络
- 本地优先：更多工具和模型支持本地运行，降低延迟和成本，保护数据隐私
AI Agent 的黄金时代才刚刚开始。掌握 Tool Calling + RAG 这两项核心能力，你就拥有了构建智能应用的最强武器。
本文由 MarkShareX AI 自动创作，分类：Agent，方向：LangChain 工具调用与 RAG 深度融合实战