小鹅通1. AI Agent开发实战：从零构建具备规划与工具调用能力的智能系统
最近在开发一个能处理复杂任务的AI Agent时，踩了不少坑，也积累了一些实战经验。与普通聊天机器人不同，真正的AI Agent需要具备环境感知、自主决策和工具调用的能力。今天就来分享下如何用LangChain框架构建一个实用的AI Agent系统。
这个系统最核心的特点是能够根据任务需求，自主决定何时调用外部工具（如计算器或数据库），并将多个工具调用串联起来完成复杂任务。比如当询问"公司预算增加46%后是多少"时，Agent会先查询原始预算，再调用计算器完成百分比计算。
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. AI Agent核心架构解析
2.1 Agent四大核心组件
一个完整的AI Agent系统通常包含以下四个关键部分：
-
LLM核心：负责基础的语言理解和生成。我们选用通义千问(qwen-plus)作为基础模型，它在中文场景表现优异。
-
记忆系统：
- 短期记忆：保存当前对话历史，通常存储最近的5-7轮对话
- 长期记忆：通过RAG(检索增强生成)实现，这里用FAISS向量数据库存储公司内部文档
-
规划能力：Agent需要自主决定任务执行流程。例如：
- 是否需要调用工具？
- 按什么顺序调用？
- 如何处理工具返回结果？
-
工具集：Agent可调用的外部函数。本案例中实现了：
- 精确计算器(处理数学运算)
- RAG搜索(查询公司内部文档)
2.2 工具调用实现原理
工具调用的核心流程其实是一个多轮对话过程：
- 用户提问 → LLM分析是否需要工具 → 如需工具，返回工具名和参数
- 系统执行工具 → 将结果封装为ToolMessage → 再次发送给LLM
- LLM整合信息 → 返回最终答案或继续调用其他工具
这个循环最多执行5次(可配置)，避免无限循环。每次循环都会打印详细的调试信息，方便开发者理解Agent的思考过程。
3. 代码实现详解
3.1 工具函数定义规范
在LangChain中，工具函数必须遵循特定格式：
python复制@tool
def tool_name(parameters) -> str:
"""
工具功能描述(LLM靠这个理解工具用途)
参数说明:
param1: 参数描述+示例
返回:
str: 返回值描述+示例
"""
# 工具实现逻辑
return "结果字符串"
特别注意：
- 必须使用@tool装饰器
- 文档字符串要详细准确，这是LLM理解工具的关键
- 返回值必须是字符串
- 参数建议使用基本类型(str/int/float等)
3.2 RAG知识库实现
我们的RAG系统采用以下技术栈：
- 文本分块：RecursiveCharacterTextSplitter
- 向量化：DashScopeEmbeddings
- 向量存储：FAISS
python复制def rag_search(query: str) -> str:
# 初始化向量数据库
raw_text = """【公司内部文档】..."""
docs = [Document(page_content=raw_text)]
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=25, # 每个文本块的大小
chunk_overlap=5 # 块之间的重叠部分
)
split_docs = text_splitter.split_documents(docs)
# 使用DashScope的嵌入模型
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
# 加载或创建FAISS索引
RAG_PATH = "faiss_index"
if os.path.exists(RAG_PATH):
ragdb = FAISS.load_local(RAG_PATH, embeddings,
allow_dangerous_deserialization=True)
else:
ragdb = FAISS.from_documents(split_docs, embeddings)
ragdb.save_local(RAG_PATH)
# 执行相似度搜索
return "\n\n".join(doc.page_content
for doc in ragdb.similarity_search(query, k=2))
关键参数说明：
- chunk_size：影响检索精度，一般25-200之间
- chunk_overlap：避免关键信息被切断
- k=2：返回最相关的2个文档片段
3.3 计算器工具的安全实现
原始版本使用eval存在安全风险，改进方案：
python复制import operator
import re
@tool
def safe_calculator(expression: str) -> str:
"""
安全计算数学表达式(仅支持基础运算)
参数:
expression: 如 "2 + 2" 或 "(5 * 3) / 2"
返回:
str: 计算结果或错误信息
"""
# 白名单校验
if not re.match(r'^[\d\s+\-*/().]+$', expression):
return "错误: 包含非法字符"
# 安全运算字典
allowed_ops = {
'+': operator.add,
'-': operator.sub,
'*': operator.mul,
'/': operator.truediv
}
try:
# 使用ast安全解析
node = ast.parse(expression, mode='eval')
for sub_node in ast.walk(node):
if isinstance(sub_node, ast.BinOp):
if not isinstance(sub_node.op, (ast.Add, ast.Sub, ast.Mult, ast.Div)):
return "错误: 不支持的运算符"
elif not isinstance(sub_node, (ast.Expression, ast.Num, ast.BinOp)):
return "错误: 不支持的语法"
# 执行安全计算
result = eval(expression, {'__builtins__': None}, allowed_ops)
return str(round(result, 2))
except Exception as e:
return f"计算错误: {e}"
这个安全版本：
- 使用正则表达式过滤非法字符
- 通过AST解析检查语法结构
- 限制可用的运算符
- 隔离执行环境
4. 多轮工具调用流程剖析
4.1 核心执行循环
python复制def run_agent(query: str):
# 初始化工具和模型
tool_maps = {"rag_search": rag_search, "calculator": safe_calculator}
llm = ChatTongyi(model_name="qwen-plus")
tool_llm = llm.bind_tools(tools=list(tool_maps.values()))
# 初始化消息历史
message = [HumanMessage(content=query)]
# 最多5轮工具调用
for i in range(5):
print(f"\n{'='*20}\n第{i+1}轮\n{query}\n{'='*20}")
# 调用LLM获取响应
response = tool_llm.invoke(message)
message.append(response)
# 检查是否需要调用工具
if not response.tool_calls:
print("最终结果：" + response.content)
return response.content
# 处理每个工具调用
for tool_call in response.tool_calls:
func_name = tool_call["name"]
if func_name in tool_maps:
# 执行工具并获取结果
tool_output = tool_maps[func_name].invoke(tool_call["args"])
print(f"工具调用：{func_name}，参数：{tool_call['args']}，结果：{tool_output}")
# 将结果加入对话历史
message.append(
ToolMessage(
content=tool_output,
tool_call_id=tool_call["id"],
name=func_name,
)
)
else:
message.append(
ToolMessage(
content=f"错误: 工具 {func_name} 不存在",
tool_call_id=tool_call["id"],
name=func_name,
)
)
return "达到最大迭代次数"
4.2 典型执行案例
案例1：查询公司计划
code复制输入: "公司计划是什么"
第1轮:
LLM决定调用rag_search工具，参数: {"query": "公司计划"}
工具返回:
【公司内部机密：代号"深蓝计划"】
1. 项目目标：开发一款能听懂猫语的翻译器...
2. 核心技术：基于Transformer的"喵声波"分析算法...
第2轮:
LLM直接返回最终答案
案例2：预算计算
code复制输入: "公司的经费预算是多少，如果预算提高46%后多少"
第1轮:
LLM调用rag_search，参数: {"query": "经费预算"}
工具返回:
5. 经费预算：仅剩50元人民币...
第2轮:
LLM调用calculator，参数: {"expression": "50 * 1.46"}
工具返回: 73.0
第3轮:
LLM整合信息返回最终答案
5. 实战经验与避坑指南
5.1 工具设计最佳实践
-
文档字符串要详细：
- 包含清晰的示例
- 说明参数格式要求
- 注明可能的错误情况
-
工具功能要单一：
不好的设计：python复制@tool def finance_tool(query): """既能查预算又能计算"""好的设计：
python复制@tool def get_budget():... @tool def calculate_percentage():... -
错误处理要完善：
- 返回可读的错误信息
- 记录详细的调试日志
- 考虑设置重试机制
5.2 常见问题排查
问题1：工具不被调用
- 检查@tool装饰器是否遗漏
- 确认工具描述是否清晰完整
- 测试直接调用工具是否能正常工作
问题2：无限循环
- 设置最大迭代次数(通常5-10次)
- 在工具返回中添加"最终答案已足够"的标记
- 监控工具调用深度并报警
问题3：工具调用顺序不合理
- 在工具描述中添加优先级提示
- 使用Chain-of-Thought引导LLM思考
- 考虑显式定义工作流
5.3 性能优化技巧
-
缓存工具结果：
python复制from functools import lru_cache @lru_cache(maxsize=100) @tool def rag_search(query):... -
批量处理工具调用：
- 识别可以并行执行的工具调用
- 使用asyncio实现并发
-
精简对话历史：
- 只保留最近3轮关键对话
- 对长文本进行摘要
6. 安全加固方案
6.1 输入验证层
python复制def sanitize_input(text: str) -> str:
# 移除敏感字符
text = re.sub(r"[;\\'\"]", "", text)
# 限制长度
return text[:500]
6.2 工具调用监控
python复制def tool_middleware(func):
def wrapper(*args, **kwargs):
start = time.time()
try:
result = func(*args, **kwargs)
log_tool_usage(func.__name__, "success", time.time()-start)
return result
except Exception as e:
log_tool_usage(func.__name__, str(e), time.time()-start)
raise
return wrapper
6.3 权限控制系统
python复制TOOL_PERMISSIONS = {
"calculator": ["user", "admin"],
"rag_search": ["admin"]
}
def check_permission(user_role, tool_name):
if tool_name not in TOOL_PERMISSIONS:
return False
return user_role in TOOL_PERMISSIONS[tool_name]
在实际项目中，我们还会添加：
- 速率限制
- 敏感词过滤
- 操作审计日志
开发AI Agent系统最关键的不仅是实现功能，更要考虑安全性、可靠性和可维护性。特别是在工具调用环节，一定要做好输入验证和权限控制。经过几个项目的实践，我发现良好的工具设计能显著提升Agent的可用性。建议从简单场景开始，逐步扩展功能，同时建立完善的监控体系。