1. AI Agent开发实战：从零构建具备工具调用能力的智能系统
最近在开发一个能处理公司内部数据的AI助手时，发现市面上关于Agent开发的实战资料实在太少。大多数教程停留在基础对话层面，而真正能完成复杂任务的Agent系统需要更深入的技术实现。今天我就把自己踩坑总结的完整开发流程分享出来，重点讲解如何让AI具备工具调用和多轮对话能力。
这个系统最核心的价值在于：它不仅能聊天，还能主动调用计算器和数据库查询工具，通过多轮交互完成"查询预算并计算增幅"这类复合任务。下面我会从架构设计到代码实现，一步步拆解开发过程中的关键技术点。
2. AI Agent核心架构解析
2.1 四大核心组件
一个完整的AI Agent需要包含以下四个关键部分：
-
LLM核心：负责基础对话和推理能力。我们选用通义千问的qwen-plus模型，它在中文场景表现优异且API稳定。
-
记忆系统：
- 短期记忆：通过维护message列表保存对话历史
- 长期记忆：使用FAISS实现的RAG知识库，存储公司内部文档
-
规划引擎：通过多轮对话循环实现任务分解和调度
-
工具集：封装了两种关键能力：
- 精确计算（calculator工具）
- 数据查询（rag_search工具）
2.2 工具调用原理
工具调用的核心机制是"绑定-触发-反馈"循环：
- 用
@tool装饰器定义工具函数 - 通过
bind_tools()将工具与LLM绑定 - LLM根据用户问题判断是否需要调用工具
- 执行工具后将结果封装为ToolMessage返回给LLM
- LLM整合工具结果生成最终回复
这个过程中最精妙的是LangChain框架自动处理了工具描述注入、调用参数提取等复杂逻辑，开发者只需关注业务实现。
3. 完整实现步骤详解
3.1 环境准备与依赖安装
首先确保Python环境≥3.8，然后安装必要依赖：
bash复制pip install langchain-core langchain-community faiss-cpu dashscope
注意：FAISS在不同平台可能有兼容性问题，Mac用户建议用conda安装：
conda install -c conda-forge faiss-cpu
3.2 工具函数实现
3.2.1 计算器工具
python复制@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果，如 "4.0" 或 "400.0"。
"""
print(f" [工具调用] 计算器正在计算: {expression}")
try:
# 安全改进：使用ast.literal_eval替代eval
import ast
return str(ast.literal_eval(expression))
except Exception as e:
return f"计算错误: {e}"
关键改进点：
- 用
ast.literal_eval替代危险的eval函数 - 完整的参数说明和返回示例（LLM依赖这些描述来调用工具）
3.2.2 RAG搜索工具
python复制@tool
def rag_search(query: str) -> str:
"""
从公司内部数据库搜索文档，可查询项目计划、预算等信息。
参数:
query: 查询关键词，如"深蓝计划 预算"
返回:
str: 匹配的文档内容，包含项目名称、预算、截止日期等
"""
# 初始化向量数据库
if not os.path.exists("faiss_index"):
docs = [Document(page_content=raw_text)]
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=100, # 增大chunk size获取更多上下文
chunk_overlap=20
)
split_docs = text_splitter.split_documents(docs)
ragdb = FAISS.from_documents(split_docs, embeddings)
ragdb.save_local("faiss_index")
else:
ragdb = FAISS.load_local("faiss_index", embeddings)
# 执行相似度搜索
results = ragdb.similarity_search(query, k=1) # 只返回最相关的一条
return results[0].page_content if results else "未找到相关文档"
优化点：
- 动态加载索引避免重复创建
- 调整chunk大小提升搜索质量
- 限制返回结果数量避免信息过载
3.3 多轮对话引擎实现
python复制def run_agent(query: str, max_turns=5):
tool_maps = {
"rag_search": rag_search,
"calculator": calculator
}
# 初始化带工具能力的LLM
llm = ChatTongyi(model_name="qwen-plus", temperature=0.1)
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_output = tool_maps[tool_name].invoke(tool_call["args"])
messages.append(
ToolMessage(
content=tool_output,
tool_call_id=tool_call["id"],
name=tool_name
)
)
raise Exception("达到最大对话轮数仍未完成")
关键设计：
- 限制最大对话轮数防止死循环
- 严格检查工具是否存在避免非法调用
- 维护完整的消息历史确保上下文连贯
4. 实战案例演示
4.1 查询项目预算
python复制run_agent("深蓝计划的预算是多少？如果增加46%后是多少？")
执行流程：
- 第一轮：识别需要查询rag_search获取预算数据
- 第二轮：提取原始预算值（50元）
- 第三轮：调用calculator计算50*1.46
- 返回最终结果："预算为50元，增加46%后是73元"
4.2 复合任务处理
python复制run_agent("请查询深蓝计划的截止日期，并计算距离今天还有多少天")
这个案例展示了Agent如何：
- 先查询RAG获取截止日期（2026-12-31）
- 再调用计算器计算日期差值
- 整合两个工具的结果生成回复
5. 安全加固方案
5.1 风险点分析
- 代码注入：原始方案使用eval执行计算表达式
- 敏感数据泄露：RAG数据库可能包含机密信息
- 无限循环：未限制工具调用次数
5.2 防护措施
输入过滤：
python复制import re
def safe_calculator(expr: str) -> str:
if not re.match(r'^[\d+\-*/(). ]+$', expr):
return "非法表达式"
# 继续安全计算...
权限控制：
python复制class RestrictedRAG:
def __init__(self, user_role):
self.allowed_queries = {
'staff': ['预算', '截止日期'],
'manager': ['核心技术', '团队暗号']
}
self.user_role = user_role
def search(self, query):
if not any(kw in query for kw in self.allowed_queries[self.user_role]):
return "权限不足"
# 执行查询...
监控日志：
python复制def log_tool_call(func):
def wrapper(*args, **kwargs):
print(f"[审计日志] {func.__name__}被调用")
return func(*args, **kwargs)
return wrapper
@log_tool_call
@tool
def calculator(expr): ...
6. 性能优化技巧
6.1 缓存机制
python复制from functools import lru_cache
@lru_cache(maxsize=100)
def rag_search(query: str):
# 相同查询直接返回缓存结果
6.2 批量处理
python复制def batch_run_queries(queries):
# 预先加载所有必要资源
with ThreadPoolExecutor() as executor:
results = list(executor.map(run_agent, queries))
return results
6.3 异步调用
python复制async def async_run_agent(query):
tool_llm = await llm.abind_tools(tools=tool_maps.values())
response = await tool_llm.ainvoke(messages)
# 异步处理工具调用...
7. 常见问题排查
7.1 工具未被调用
可能原因：
- 工具描述不够清晰（检查docstring）
- LLM温度参数过高（建议设为0.1-0.3）
- 缺少示例参数（在描述中添加示例）
7.2 结果不准确
解决方案：
- 优化RAG的chunk_size（通常100-500）
- 增加相似度搜索的k值
- 在工具描述中明确数据格式要求
7.3 循环次数过多
处理方法：
- 设置合理的max_turns（3-5轮）
- 添加超时机制
- 在系统提示中明确限制条件
我在实际项目中发现，最影响Agent表现的因素是工具描述的准确性。建议为每个工具提供：
- 清晰的功能说明
- 完整的参数示例
- 预期的返回格式
- 常见使用场景
比如calculator工具的描述就明确指出了它适用于"需要精确计算时使用"，这能有效引导LLM在适当场景调用它。