1. AI Agent开发实战:从零构建具备工具调用能力的智能体
最近在开发一个能自动处理公司内部查询的AI助手时,深入研究了AI Agent的实现方式。与普通聊天机器人不同,真正的AI Agent应该具备自主决策、工具调用和任务规划能力。下面分享我的开发经验和踩过的坑。
2. AI Agent核心架构解析
2.1 四大核心组件
一个完整的AI Agent需要包含以下组件:
- LLM核心:负责基础对话和推理能力
- 记忆系统:
- 短期记忆:保存对话上下文
- 长期记忆:通过RAG接入知识库
- 规划模块:拆解复杂任务为可执行步骤
- 工具集:可调用的外部函数接口
在我的实现中,使用LangChain框架将这些组件有机整合。相比直接调用大模型API,这种架构更适合构建生产级应用。
2.2 工具调用机制详解
工具调用是Agent区别于普通聊天机器人的关键能力。实现要点包括:
- 工具函数必须用
@tool装饰器标记 - 函数文档字符串需要详细描述功能、参数和返回值
- 工具返回值必须是字符串格式
- 需要实现多轮对话机制处理工具调用链
特别注意:工具描述文档的质量直接影响大模型能否正确调用工具。建议参考OpenAI的Function Calling文档规范编写。
3. 实战开发:公司内部查询Agent
3.1 环境准备
bash复制pip install langchain langchain-community faiss-cpu
需要准备:
- 通义千问API Key(或其他兼容OpenAI的LLM服务)
- FAISS向量数据库(本地运行无需额外服务)
3.2 核心工具实现
3.2.1 精确计算器
python复制@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式,如 "2 + 2" 或 "500 * 0.8"
返回:
str: 计算结果,如 "4.0" 或 "400.0"
示例:
>>> calculator("2 + 2")
"4.0"
>>> calculator("500 * 0.8")
"400.0"
"""
print(f"[工具调用] 计算器正在计算: {expression}")
try:
# 安全计算实现(替代eval)
from ast import literal_eval
return str(literal_eval(expression))
except Exception as e:
return f"计算错误: {e}"
3.2.2 RAG知识库搜索
python复制@tool
def rag_search(query: str) -> str:
"""
从公司内部数据库搜索文档,包括项目计划、预算等信息
参数:
query: 搜索关键词,如"深蓝计划 预算"
返回:
str: 匹配的文档内容
示例:
>>> rag_search("深蓝计划")
"【公司内部机密:代号'深蓝计划'】..."
"""
# 初始化向量数据库
if not os.path.exists(RAG_PATH):
# 文档预处理和向量化
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200, # 更大的分块尺寸
chunk_overlap=50
)
docs = text_splitter.split_documents([Document(page_content=raw_text)])
ragdb = FAISS.from_documents(docs, embeddings)
ragdb.save_local(RAG_PATH)
else:
ragdb = FAISS.load_local(RAG_PATH, embeddings)
# 执行相似度搜索
results = ragdb.similarity_search(query, k=3)
return "\n\n".join(doc.page_content for doc in results)
3.3 Agent运行逻辑
python复制def run_agent(query: str, max_turns=5):
"""
执行Agent对话流程
参数:
query: 用户输入
max_turns: 最大对话轮次(防止无限循环)
"""
tool_maps = {
"rag_search": rag_search,
"calculator": calculator
}
# 初始化LLM并绑定工具
llm = ChatTongyi(model_name="qwen-plus")
tool_llm = llm.bind_tools(tools=list(tool_maps.values()))
messages = [HumanMessage(content=query)]
for turn in range(max_turns):
print(f"\n=== 第{turn+1}轮对话 ===")
response = tool_llm.invoke(messages)
messages.append(response)
if not response.tool_calls:
print("最终结果:", response.content)
return response.content
# 处理工具调用
for tool_call in response.tool_calls:
tool_name = tool_call["name"]
if tool_name not in tool_maps:
continue
tool_func = tool_maps[tool_name]
tool_output = tool_func.invoke(tool_call["args"])
messages.append(
ToolMessage(
content=tool_output,
tool_call_id=tool_call["id"],
name=tool_name
)
)
return "达到最大对话轮次限制"
4. 关键问题与解决方案
4.1 安全风险防范
原始实现中使用eval()存在严重安全隐患。改进方案:
-
输入验证:
python复制import re if not re.match(r'^[\d\s+\-*/().]+$', expression): return "错误:包含非法字符" -
使用更安全的替代方案:
ast.literal_eval(仅支持字面量表达式)- 第三方库如
simpleeval - 自定义解析器
-
LLM层面防护:
- 在系统提示词中加入安全约束
- 设置工具调用的黑名单关键词
4.2 性能优化技巧
-
向量数据库优化:
- 调整chunk_size和chunk_overlap参数
- 使用更好的embedding模型
- 定期重建索引
-
对话流程优化:
- 设置合理的max_turns值(通常3-5轮足够)
- 实现对话状态跟踪
- 添加超时机制
-
缓存策略:
- 缓存频繁查询的RAG结果
- 记忆重复计算表达式
5. 扩展应用场景
这个基础框架可以扩展到更多场景:
-
客户服务:
- 集成CRM系统查询
- 订单状态跟踪
- 自动生成服务工单
-
数据分析:
- SQL查询工具
- 可视化图表生成
- 自动报告撰写
-
办公自动化:
- 邮件处理
- 会议纪要生成
- 文档摘要
实际部署时,建议从简单场景开始,逐步增加工具复杂度。每个新工具都需要:
- 清晰的文档描述
- 完善的参数验证
- 详细的错误处理
- 充分的测试用例
6. 开发心得
在开发过程中,最大的教训是不要过度信任LLM的输出。特别是在工具调用场景下,必须:
- 对所有输入参数进行严格验证
- 实现完善的错误处理机制
- 设置合理的执行超时和重试策略
- 记录完整的执行日志用于审计
另一个实用技巧是给每个工具添加版本号,这样可以在不中断服务的情况下进行迭代更新:
python复制@tool
def rag_search_v2(query: str) -> str:
"""[v2] 改进版RAG搜索..."""
对于想要深入学习的开发者,建议从LangChain官方文档入手,然后尝试实现以下增强功能:
- 支持异步工具调用
- 实现工具组合调用(一个工具调用另一个工具)
- 添加用户权限验证
- 开发可视化调试界面