1. AI Agent 开发实战：从零构建具备工具调用能力的智能体
在当今AI技术快速发展的背景下，单纯的对话机器人已经不能满足复杂场景需求。一个真正的AI智能体（Agent）应该像人类助手一样，能够自主规划任务、调用工具并完成实际工作。最近我在开发一个企业级AI Agent项目时，深刻体会到了工具调用能力的重要性。这个Agent不仅能回答常规问题，还能查询公司内部数据库、进行精确计算，甚至处理复杂的多步骤任务。
传统聊天机器人最大的局限在于"纸上谈兵"——它们能说会道，但缺乏实际行动能力。而现代AI Agent通过整合大语言模型（LLM）与外部工具，实现了从"说"到"做"的跨越。下面我将分享如何用LangChain框架构建一个具备工具调用能力的AI Agent，包括核心架构设计、安全考量以及实际开发中的经验教训。
2. AI Agent 核心架构解析
2.1 Agent 四大核心组件
一个功能完备的AI Agent需要四大核心组件协同工作：
-
大语言模型(LLM)：负责自然语言理解、推理和决策。在项目中我选择了通义千问（qwen-plus）作为基础模型，因其在中文场景下的优秀表现。
-
记忆系统：
- 短期记忆：保存当前对话上下文
- 长期记忆：通过RAG（检索增强生成）技术接入企业知识库
-
规划能力：Agent需要将复杂任务拆解为可执行的子步骤。例如当用户问"预算提高46%后是多少"时，Agent需要先查询原始预算，再调用计算器进行计算。
-
工具调用：Agent与外部世界的接口。在我们的案例中实现了两个关键工具：
- RAG搜索：查询企业内部文档
- 精确计算器：处理数学运算
2.2 工具调用机制详解
工具调用是Agent能力的扩展点。在LangChain中，工具通过@tool装饰器定义：
python复制@tool
def calculator(expression: str) -> str:
"""计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果，如 "4.0" 或 "400.0"。
"""
print(f" [工具调用] 计算器正在计算: {expression}")
try:
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
工具定义有三个关键部分：
- 函数描述：LLM通过这部分理解工具用途，因此必须清晰完整
- 参数说明：包括示例，帮助模型正确调用
- 实现逻辑：最终执行实际操作的代码
重要提示：工具函数必须返回字符串，因为LLM只能处理文本格式的输入输出。
3. 实战开发：构建企业级AI Agent
3.1 环境准备与初始化
首先需要安装必要的Python包：
bash复制pip install langchain langchain-community faiss-cpu
然后初始化核心组件：
python复制from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
# 初始化大模型
llm = ChatTongyi(model_name="qwen-plus")
# 工具映射表
tool_maps = {
"rag_search": rag_search,
"calculator": calculator
}
# 将工具绑定到模型
tool_llm = llm.bind_tools(tools=list(tool_maps.values()))
3.2 RAG知识库实现
企业知识库是Agent长期记忆的核心。我们使用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)
# 向量化并存储
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
if os.path.exists(RAG_PATH):
ragdb = FAISS.load_local(RAG_PATH, embeddings)
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：每个文本块的最大长度chunk_overlap=5：块之间的重叠字符数，保证上下文连贯k=2：返回最相关的2个文档片段
3.3 多轮对话控制逻辑
Agent通过循环实现多轮对话和工具调用：
python复制def run_agent(query: str):
message = [HumanMessage(content=query)]
# 限制最多5轮对话防止死循环
for i in range(5):
response = tool_llm.invoke(message)
message.append(response)
if not response.tool_calls:
return response.content # 最终结果
for tool_call in response.tool_calls:
# 执行工具调用
tool_func = tool_maps[tool_call["name"]]
tool_output = tool_func.invoke(tool_call["args"])
# 将结果加入对话历史
message.append(
ToolMessage(
content=tool_output,
tool_call_id=tool_call["id"],
name=tool_call["name"]
)
)
这个控制流程实现了完整的"提问-决策-执行-反馈"循环。关键点包括：
- 对话轮次限制（防止无限循环）
- 工具调用结果必须封装为ToolMessage
- 每次迭代都将所有信息加入对话历史，保持上下文
4. 安全风险与防御实践
4.1 代码注入风险分析
在初版实现中，计算器工具直接使用Python的eval()函数，这带来了严重的安全隐患：
python复制@tool
def calculator(expression: str) -> str:
try:
return str(eval(expression)) # 危险！
except Exception as e:
return f"计算错误: {e}"
恶意用户可能通过精心构造的输入执行任意代码，例如：
code复制"__import__('os').system('rm -rf /')"
4.2 多层防御方案
我们实施了四重防护措施：
- 输入白名单验证：
python复制import re
def safe_eval(expr: str) -> str:
if not re.match(r'^[\d\s\+\-\*\/\.\(\)\%]+$', expr):
raise ValueError("非法字符")
# 其余验证逻辑...
- 使用ast.literal_eval替代eval：
python复制import ast
def safe_calculate(expr: str) -> str:
try:
node = ast.parse(expr, mode='eval')
if not all(isinstance(n, ast.Num) or isinstance(n, ast.BinOp) for n in ast.walk(node)):
raise ValueError("只允许基本数学运算")
return str(eval(expr, {'__builtins__': None}, {}))
except Exception as e:
return f"计算错误: {e}"
- 模型层指令防护：
在系统提示词中加入：
code复制你是一个数学计算助手，只能处理简单的算术表达式。
如果用户请求包含以下内容，必须拒绝：
- 任何非数学字符
- 系统命令
- 函数调用
- 变量赋值
- 沙箱环境执行：
考虑使用Docker容器隔离计算过程，限制资源访问。
5. 性能优化与生产实践
5.1 对话流程优化
在实际测试中，我们发现几个性能瓶颈：
-
工具调用延迟：每次RAG搜索都需要加载向量数据库
- 解决方案：实现数据库连接池，保持长连接
-
多轮对话冗余：有时Agent会不必要地重复调用工具
- 解决方案：增加工具调用缓存机制
优化后的工具调用逻辑：
python复制from functools import lru_cache
@lru_cache(maxsize=100)
def cached_rag_search(query: str) -> str:
return rag_search(query)
5.2 监控与日志
生产环境需要完善的监控体系：
python复制def log_tool_call(func):
def wrapper(*args, **kwargs):
start = time.time()
try:
result = func(*args, **kwargs)
duration = time.time() - start
log_entry = {
"tool": func.__name__,
"args": args,
"duration": duration,
"status": "success"
}
return result
except Exception as e:
log_entry["status"] = "failed"
log_entry["error"] = str(e)
raise
finally:
logging.info(json.dumps(log_entry))
return wrapper
5.3 扩展性设计
为支持未来更多工具，我们改进了工具注册机制：
python复制class ToolBox:
def __init__(self):
self.tools = {}
def register(self, name=None):
def decorator(func):
tool_name = name or func.__name__
self.tools[tool_name] = func
return func
return decorator
toolbox = ToolBox()
@toolbox.register("calc")
def safe_calculator(expr: str) -> str:
"""安全版本的计算器"""
# 实现...
这种设计允许动态添加工具，而不需要修改核心Agent代码。
6. 典型问题排查指南
在实际部署中，我们遇到了以下典型问题：
6.1 工具调用失败
症状：Agent决定调用工具，但实际未执行
排查步骤：
- 检查工具函数是否正确定义了
__doc__字符串 - 验证工具是否成功绑定到LLM（检查
tool_llm.tools） - 检查模型返回的tool_calls格式是否正确
6.2 多轮对话中断
症状：对话在2-3轮后意外结束
解决方案：
- 确保每轮对话都将完整历史传入（包括ToolMessage）
- 检查模型temperature参数（过高可能导致输出不稳定）
- 增加错误处理逻辑，避免异常终止
6.3 RAG检索效果差
优化方案：
- 调整文本分块策略（chunk_size和chunk_overlap）
- 尝试不同的embedding模型
- 添加查询重写逻辑，优化搜索关键词
python复制def query_rewrite(original_query: str) -> str:
"""优化RAG查询语句"""
rewrite_prompt = f"""
请将以下用户查询改写为更适合文档检索的形式，
保持原意但更完整明确：
原查询：{original_query}
改写后："""
return llm.invoke(rewrite_prompt).content
7. 项目演进方向
当前系统已经实现了基础的工具调用能力，下一步计划：
- 多工具协同：实现工具间的数据传递，如将RAG搜索结果自动传入计算器
- 动态工具加载：支持运行时添加/移除工具而不重启服务
- 可视化编排：提供图形界面配置工具调用流程
- 性能监控面板：实时展示各工具调用耗时、成功率等指标
一个特别实用的扩展是在计算器工具中增加单位换算功能：
python复制@tool
def enhanced_calculator(expr: str) -> str:
"""支持单位换算的科学计算器
示例：
"10km to miles" -> "6.21371 miles"
"1GB + 500MB in KB" -> "1536000.0 KB"
"""
# 实现单位转换逻辑...
通过持续迭代，我们正将这套AI Agent系统打造为企业级的智能助手平台，能够处理越来越复杂的实际业务场景。