1. AI Agent核心架构解析
在当今AI技术快速发展的背景下,传统的聊天机器人已经无法满足复杂业务场景的需求。一个真正的AI Agent应该具备感知环境、自主决策和执行任务的能力。通过将大语言模型(LLM)与规划、记忆和工具使用能力相结合,我们可以构建出能够处理复杂任务的智能代理系统。
1.1 四大核心组件
一个完整的AI Agent系统由以下关键组件构成:
-
LLM核心:作为系统的大脑,负责语言理解、推理和决策。在示例中我们使用了通义千问(qwen-plus)模型,它具备优秀的上下文理解和逻辑推理能力。
-
记忆系统:
- 短期记忆:保存对话历史,维持上下文连贯性
- 长期记忆:通过RAG(检索增强生成)技术连接知识库,示例中使用了FAISS向量数据库存储公司内部文档
-
规划能力:Agent能够将复杂任务拆解为多个子步骤,并决定执行顺序。示例中通过多轮对话循环实现了简单的任务规划。
-
工具使用:Agent可以调用外部工具扩展能力边界。示例中实现了两个工具:
- 精确计算器:处理数学运算
- RAG搜索:查询公司内部知识库
提示:在设计工具时,函数描述文档至关重要。LLM完全依赖这些描述来决定何时以及如何使用工具,因此需要详细说明参数、返回值和示例。
1.2 工作流程设计
Agent的工作流程遵循"感知-决策-执行"循环:
- 接收用户输入(HumanMessage)
- LLM分析输入并决定是否需要调用工具
- 如果调用工具,执行相应函数并获取结果
- 将工具执行结果(ToolMessage)返回给LLM
- LLM整合信息生成最终响应
这种设计允许Agent通过工具扩展能力,同时保持LLM的核心决策作用。在实际应用中,可以根据需求扩展更多工具,如API调用、数据库查询等。
2. 工具系统实现详解
工具是AI Agent与外部世界交互的桥梁。下面我们深入分析示例中两个工具的实现细节。
2.1 计算器工具实现
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}"
关键实现要点:
- @tool装饰器:将普通Python函数转换为Agent可调用的工具
- 文档字符串:必须包含清晰的描述、参数说明和返回示例
- 错误处理:捕获所有异常并返回友好错误信息
- 类型标注:明确输入输出类型,帮助LLM正确使用工具
安全警示:直接使用eval()存在严重安全风险,可能被注入恶意代码。生产环境必须替换为更安全的实现方式。
2.2 RAG搜索工具实现
python复制@tool
def rag_search(query: str) -> str:
"""
从数据库中搜索与查询公司内部相关的文档,包括公司计划名,代号,截止日期等详细信息。
参数:
query (str): 要搜索的查询字符串。
返回:
str: 与查询相关的文档内容。
"""
# 文档预处理和向量数据库构建
raw_text = """【公司内部机密:代号"深蓝计划"】..."""
RAG_PATH = "faiss_index"
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, 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))
关键技术点:
- 文档分块:使用RecursiveCharacterTextSplitter将长文档分割为适合处理的片段
- 向量化:通过DashScopeEmbeddings生成文本嵌入向量
- 向量搜索:FAISS实现高效的相似度搜索
- 持久化:将向量数据库保存到本地避免重复构建
3. 多轮对话引擎实现
Agent的核心在于能够进行多轮对话,动态决定工具调用。下面详细解析实现机制。
3.1 对话循环设计
python复制def run_agent(query:str):
# 初始化模型和工具
tool_maps = {
"rag_search": rag_search,
"calculator": 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("="*20+"\n第"+str(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
# 处理每个工具调用
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("工具调用:" + func_name + ",参数:" + str(func_args) + ",结果:" + tool_output)
else:
tool_output = f"错误: 工具 {func_name} 不存在。"
# 将工具结果添加到消息历史
message.append(
ToolMessage(
content=tool_output,
tool_call_id=call_id,
name=func_name,
)
)
3.2 关键机制解析
- 工具绑定:通过bind_tools()方法将工具描述注入LLM的上下文
- 消息传递:维护消息历史(HumanMessage, AIMessage, ToolMessage)
- 工具调用检测:检查response.tool_calls判断是否需要调用工具
- 结果反馈:将工具执行结果封装为ToolMessage返回给LLM
- 循环控制:限制最大轮数防止无限循环
3.3 典型执行流程示例
以查询"公司的经费预算是多少,如果预算提高46%后多少"为例:
- 第一轮:LLM决定调用rag_search获取预算信息
- 第二轮:收到预算金额(50元)后,LLM决定调用calculator计算增加46%后的值
- 第三轮:整合两个工具的结果生成最终响应
4. 安全风险与防御措施
AI Agent系统面临多种安全挑战,必须采取适当的防护措施。
4.1 主要安全风险
- 代码注入:示例中calculator使用eval(),可能执行任意代码
- 提示词注入:精心设计的输入可能操纵Agent行为
- 敏感信息泄露:RAG系统可能返回不应公开的数据
- 无限循环:不当的工具调用设计可能导致死循环
4.2 安全加固方案
4.2.1 计算器安全改造
python复制import re
from ast import literal_eval
@tool
def safe_calculator(expression: str) -> str:
"""
安全版计算器,仅支持基本数学运算。
参数:
expression: 数学算式,如 "(2 + 2) * 3"。
允许运算符: + - * / % ( )
允许字符: 数字 0-9, 小数点, 空格, 上述运算符
返回:
str: 计算结果或错误信息
"""
# 白名单验证
if not re.fullmatch(r'^[\d\s+\-*/%().]+$', expression):
return "错误: 表达式包含非法字符"
try:
# 使用ast.literal_eval替代eval
return str(literal_eval(expression))
except Exception as e:
return f"计算错误: {e}"
4.2.2 其他防护措施
- 输入验证:对所有用户输入进行严格过滤
- 权限控制:限制工具可访问的资源范围
- 执行监控:记录所有工具调用及参数
- 速率限制:防止滥用和拒绝服务攻击
- 沙箱环境:在隔离环境中执行不可信代码
5. 高级应用与扩展方向
基础Agent框架可以扩展支持更复杂的应用场景。
5.1 多工具协同工作
通过设计工具间的数据流,可以实现复杂任务的自动化处理。例如:
- 从数据库查询原始数据
- 使用计算工具进行统计分析
- 调用图表生成工具可视化结果
- 通过邮件工具发送报告
5.2 动态工具加载
实现工具的热加载机制,允许运行时添加新工具而无需重启Agent:
python复制def register_tool(tool_func):
"""动态注册新工具"""
tool_maps[tool_func.name] = tool_func
llm.bind_tools(tools=list(tool_maps.values())) # 重新绑定
5.3 记忆增强策略
- 对话总结:定期压缩长对话历史
- 重要性标记:让用户标记关键信息长期保存
- 自动关联:基于语义关联相关记忆片段
在实际项目中,我们曾通过增加优先级机制解决工具冲突问题。当多个工具可用时,让LLM根据工具描述中的优先级标记选择最合适的工具。这显著提高了复杂任务的成功率。