1. AI Agent 开发实战:从零构建具备工具调用能力的智能体
最近在开发一个能处理公司内部查询的AI助手时,深刻体会到传统聊天机器人和真正AI Agent的区别。普通的聊天机器人只能进行简单的问答,而一个完整的AI Agent能够自主规划任务、调用工具、处理复杂场景。下面分享我在开发过程中的实战经验,特别是如何让Agent具备工具调用能力。
2. AI Agent 核心架构解析
2.1 Agent 四大核心组件
一个功能完备的AI Agent通常包含以下四个关键组件:
-
LLM(大语言模型):负责自然语言理解和生成,是Agent的"大脑"。在项目中我选择了通义千问(qwen-plus)作为基础模型,因其在中文场景表现优异。
-
记忆系统:
- 短期记忆:保存当前对话上下文
- 长期记忆:通过RAG(检索增强生成)技术访问知识库
-
规划模块:决定任务执行流程,包括是否/何时调用工具
-
工具集:Agent可以调用的外部函数,如计算器、数据库查询等
2.2 工具调用机制详解
工具调用是Agent区别于普通聊天机器人的核心能力。其工作流程如下:
- 用户输入查询
- LLM分析需求,决定是否需要调用工具
- 如需调用,返回工具名称和参数
- 执行工具函数
- 将结果返回LLM进行整合
- 生成最终回复给用户
这个过程中最关键的环节是工具的定义和绑定,下面会详细说明。
3. 实战开发:构建公司信息查询Agent
3.1 环境准备与依赖安装
首先需要安装必要的Python库:
bash复制pip install langchain langchain-community faiss-cpu dashscope
建议使用Python 3.8+环境,并准备好通义千问的API密钥。
3.2 工具函数定义规范
每个工具函数都需要遵循特定规范:
python复制from langchain_core.tools import tool
@tool
def tool_name(parameters) -> str:
"""
工具描述(LLM通过这部分理解工具用途)
参数说明:
parameter: 参数说明及示例
返回:
str: 返回结果说明及示例
"""
# 工具实现逻辑
return "结果"
关键点:
- 必须使用
@tool装饰器 - 文档字符串要详细准确(这是LLM理解工具用途的依据)
- 返回值必须是字符串
- 参数类型提示要明确
3.3 核心工具实现
3.3.1 精确计算器实现
python复制@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式,如 "2 + 2" 或 "500 * 0.8"
返回:
str: 计算结果,如 "4.0" 或 "400.0"
"""
print(f"[工具调用] 计算器正在计算: {expression}")
try:
# 安全提示:实际生产环境应替换eval为更安全的计算方式
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
重要安全提示:生产环境使用eval存在安全风险,建议替换为安全计算库如
numexpr
3.3.2 RAG知识库实现
python复制@tool
def rag_search(query: str) -> str:
"""
从公司内部知识库搜索文档,包括项目计划、预算等信息
参数:
query: 搜索关键词
返回:
str: 相关文档内容
"""
# 知识库初始化
raw_text = """
【公司内部信息】
1. 项目目标:开发智能翻译器
2. 核心技术:Transformer算法
3. 截止日期:2026-12-31
4. 经费预算:50万元
"""
# 文档处理与向量化
docs = [Document(page_content=raw_text)]
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=20
)
split_docs = text_splitter.split_documents(docs)
# 使用DashScope的嵌入模型
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
# 创建/加载FAISS向量库
RAG_PATH = "company_knowledge"
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)
)
3.4 Agent核心逻辑实现
python复制def run_agent(query: str, max_rounds=5):
# 工具映射
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()))
# 初始化消息历史
message = [HumanMessage(content=query)]
# 多轮对话循环
for i in range(max_rounds):
print(f"\n{'='*20}\n第{i+1}轮\n{query}\n{'='*20}")
# 获取LLM响应
response = tool_llm.invoke(message)
message.append(response)
print(f"需要调用{len(response.tool_calls)}个工具")
# 如果没有工具调用,返回最终结果
if not response.tool_calls:
print("最终结果:" + response.content)
return response.content
# 处理每个工具调用
for tool_call in response.tool_calls:
call_id = tool_call["id"]
func_name = tool_call["name"]
func_args = tool_call["args"]
# 安全检查
if func_name in tool_maps:
tool_func = tool_maps[func_name]
tool_output = tool_func.invoke(func_args)
print(f"工具调用:{func_name},参数:{func_args},结果:{tool_output}")
else:
tool_output = f"错误: 工具 {func_name} 不存在"
# 将工具结果加入消息历史
message.append(
ToolMessage(
content=tool_output,
tool_call_id=call_id,
name=func_name,
)
)
return "超过最大对话轮数"
4. 关键问题与优化方案
4.1 工具调用安全问题
风险点:
- 直接使用eval执行计算表达式存在代码注入风险
- 工具可能被滥用访问敏感信息
解决方案:
- 使用更安全的计算库替代eval:
python复制from numexpr import evaluate
def safe_calculator(expr: str) -> str:
try:
return str(evaluate(expr))
except:
return "计算错误"
- 增加工具调用权限控制:
python复制def check_tool_permission(tool_name, user_role):
permission_map = {
"admin": ["*"],
"staff": ["calculator"],
"guest": []
}
return tool_name in permission_map.get(user_role, [])
4.2 多轮对话控制
常见问题:
- 工具调用陷入无限循环
- 上下文过长导致性能下降
优化方案:
- 设置最大对话轮数(如代码中的max_rounds)
- 实现上下文窗口管理:
python复制def trim_messages(messages, max_tokens=4000):
total = 0
trimmed = []
for msg in reversed(messages):
msg_len = len(str(msg))
if total + msg_len > max_tokens:
break
trimmed.append(msg)
total += msg_len
return list(reversed(trimmed))
4.3 工具描述优化技巧
工具的描述文档直接影响LLM对工具的理解和使用。好的描述应包含:
- 工具的核心用途
- 每个参数的详细说明和示例
- 返回值的格式说明
- 使用场景示例
例如:
python复制@tool
def get_weather(city: str, date: str) -> str:
"""
获取指定城市在特定日期的天气信息
参数:
city: 城市名称,如"北京"、"上海"
date: 日期,格式YYYY-MM-DD,如"2024-03-15"
返回:
str: 天气信息,格式为"城市,日期:天气状况,温度范围"
例如"北京,2024-03-15:晴,8℃~15℃"
示例调用:
- 查询北京明天天气: get_weather("北京", "2024-03-16")
- 查询上海今天天气: get_weather("上海", "2024-03-15")
"""
# 实现逻辑
5. 高级应用场景扩展
5.1 动态工具加载
实现工具的热加载,无需重启Agent:
python复制class ToolManager:
def __init__(self):
self.tools = {}
def register_tool(self, tool_func):
self.tools[tool_func.name] = tool_func
def update_llm_tools(self, llm):
return llm.bind_tools(tools=list(self.tools.values()))
# 使用示例
manager = ToolManager()
manager.register_tool(calculator)
llm_with_tools = manager.update_llm_tools(llm)
5.2 工具组合调用
实现工具的链式调用,处理复杂任务:
python复制def plan_and_execute(agent, initial_query):
plan = agent(f"请将以下任务分解为多个步骤:{initial_query}")
steps = parse_steps(plan) # 解析步骤
results = []
for step in steps:
result = agent(step)
results.append(result)
final_result = agent(f"整合以下信息生成最终回复:{results}")
return final_result
5.3 性能监控与日志
添加工具调用监控:
python复制def monitored_tool(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
try:
result = func(*args, **kwargs)
status = "success"
except Exception as e:
result = str(e)
status = "failed"
end = time.time()
log_entry = {
"tool": func.__name__,
"args": args,
"kwargs": kwargs,
"status": status,
"duration": end - start,
"timestamp": datetime.now().isoformat()
}
# 写入日志系统
log_to_system(log_entry)
return result
return wrapper
# 使用装饰器
@monitored_tool
@tool
def monitored_calculator(expr: str) -> str:
# 实现逻辑
6. 实际应用案例演示
6.1 公司信息查询
python复制run_agent("公司当前有哪些项目?预算情况如何?")
执行流程:
- 识别需要查询公司信息 → 调用rag_search
- 从知识库获取项目信息
- 识别需要预算数据 → 再次调用rag_search
- 整合信息生成回复
6.2 复杂计算场景
python复制run_agent("如果项目预算增加30%,再扣除10%的管理费,最终预算是多少?")
执行流程:
- 查询当前预算 → rag_search
- 计算增加30% → calculator
- 计算扣除10% → calculator
- 生成最终结果
6.3 混合任务处理
python复制run_agent("查询项目A的截止日期,并计算距离今天还有多少天")
执行流程:
- 查询项目截止日期 → rag_search
- 获取当前日期 → 内置工具
- 计算日期差 → calculator
- 生成自然语言回复
7. 开发经验与避坑指南
7.1 工具设计最佳实践
-
单一职责原则:每个工具只做一件事
- 反例:一个工具既查询又计算
- 正例:分开为rag_search和calculator
-
输入验证:工具内部要验证参数
python复制@tool def safe_calculator(expr: str) -> str: if not re.match(r'^[\d\+\-\*\/\.\(\) ]+$', expr): return "错误:包含非法字符" # 其余逻辑 -
错误处理:提供有意义的错误信息
python复制try: result = some_operation() except SpecificError as e: return f"操作失败:{str(e)}。建议检查..."
7.2 性能优化技巧
-
工具预热:提前加载耗时资源
python复制# 在服务启动时加载 embeddings = load_embeddings() ragdb = load_vector_db() -
缓存机制:缓存频繁使用的工具结果
python复制from functools import lru_cache @lru_cache(maxsize=100) @tool def cached_search(query: str) -> str: # 实现逻辑 -
异步调用:并行执行独立工具
python复制async def parallel_tool_calls(tool_calls): tasks = [] for call in tool_calls: tool = tool_maps[call["name"]] tasks.append(tool.ainvoke(call["args"])) return await asyncio.gather(*tasks)
7.3 调试与测试建议
-
交互式调试:打印完整对话历史
python复制def debug_print(messages): for i, msg in enumerate(messages): print(f"[{i}] {type(msg).__name__}: {msg.content}") -
单元测试工具:单独测试每个工具
python复制def test_calculator(): assert calculator("2+2") == "4.0" assert calculator("10/0").startswith("计算错误") -
集成测试场景:模拟完整用户对话
python复制def test_budget_query(): result = run_agent("预算增加20%后是多少?") assert "计算结果" in result
8. 安全防护方案
8.1 输入过滤机制
python复制def sanitize_input(text: str) -> str:
# 移除危险字符
text = re.sub(r"[;\\'\"]", "", text)
# 限制长度
return text[:500]
8.2 权限控制系统
python复制class AccessControl:
def __init__(self):
self.policies = {
"calculator": ["staff", "manager"],
"rag_search": ["manager"]
}
def check_access(self, user, tool_name):
return tool_name in self.policies and \
user.role in self.policies[tool_name]
8.3 审计日志实现
python复制def log_operation(user, action, details):
entry = {
"timestamp": datetime.now(),
"user": user.id,
"action": action,
"details": details,
"ip": request.remote_addr
}
db.logs.insert_one(entry)
9. 生产环境部署建议
9.1 性能考量
-
LLM选择:根据场景选择合适大小的模型
- 简单任务:较小模型(如qwen-1.8b)
- 复杂任务:较大模型(如qwen-72b)
-
批处理:合并多个工具调用
python复制def batch_process(queries): with ThreadPoolExecutor() as executor: results = list(executor.map(run_agent, queries)) return results
9.2 监控指标
关键监控指标:
- 工具调用成功率
- 平均响应时间
- 错误类型分布
- 资源使用情况
9.3 容错设计
-
重试机制:
python复制def retry_tool(tool_func, args, max_retries=3): for i in range(max_retries): try: return tool_func(*args) except Exception as e: if i == max_retries - 1: raise time.sleep(1) -
降级方案:
python复制def fallback_response(query): return f"系统繁忙,暂��无法处理'{query}'。请稍后再试或联系管理员"
10. 未来扩展方向
- 多模态工具:支持图像、语音等工具
- 自学习机制:根据使用反馈优化工具
- 分布式执行:跨服务器调用工具
- 可视化编排:图形化工具流程设计
在实际开发中,我发现工具调用能力是区分初级和高级AI Agent的关键。通过良好的工具设计和安全防护,可以构建出真正实用的企业级智能助手。最难的部分不是编码实现,而是设计出既灵活又安全的工具架构。建议从简单场景开始,逐步扩展复杂度,同时建立完善的测试和监控体系。
