AI Agent架构解析：LLM+规划+记忆+工具实现

1. AI Agent 核心架构解析

AI Agent 的核心架构可以概括为"LLM+规划+记忆+工具"四大模块。这种架构设计源于对人类智能行为的模拟——我们的大脑（LLM）处理信息时，会调用记忆系统（记忆模块），制定行动计划（规划模块），并使用各种工具（工具模块）完成任务。

LLM 作为 Agent 的"大脑"，负责信息处理和决策生成。在实际应用中，我们通常选择具备较强推理能力的通用大模型作为基础，例如案例中的通义千问（qwen-plus）。这类模型在零样本学习（Zero-shot Learning）和少样本学习（Few-shot Learning）场景下表现优异，能够理解复杂指令并生成合理响应。

记忆系统分为短期记忆和长期记忆两个维度：

• 短期记忆：保存当前对话的上下文信息，通常以对话历史列表的形式维护
• 长期记忆：通过 RAG（检索增强生成）技术实现的知识库系统，案例中使用 FAISS 向量数据库存储公司内部文档

规划模块体现在 Agent 的多轮决策能力上。当收到用户查询时，Agent 会自主判断是否需要调用工具、调用哪些工具、如何处理工具返回结果等。案例中通过循环机制实现了最多5轮的决策过程。

工具模块是 Agent 与外部世界交互的接口。每个工具本质上是一个 Python 函数，但需要遵循特定规范：

1. 使用 @tool 装饰器声明
2. 函数文档字符串必须详细描述功能、参数和返回值
3. 返回值必须是字符串类型
4. 函数实现需要处理各种异常情况

2. 工具系统实现细节

2.1 工具定义规范

工具函数的定义有严格的要求，这是为了让 LLM 能够正确理解和使用工具。以案例中的计算器工具为例：

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 无法正确使用工具。

2.2 工具绑定机制

将工具与 LLM 绑定的过程实际上是为模型提供了工具的使用说明书。LangChain 的 bind_tools 方法会将工具信息编码成模型能理解的格式：

python复制tool_maps = {
    "rag_search": rag_search,
    "calculator": calculator
}
llm = ChatTongyi(model_name="qwen-plus")
tool_llm = llm.bind_tools(tools=list(tool_maps.values()))

绑定后，模型在生成响应时会额外输出 tool_calls 字段，包含以下信息：

• id：本次工具调用的唯一标识
• name：要调用的工具名称
• args：调用参数（字典格式）

3. 多轮对话控制流程

3.1 对话状态管理

Agent 通过维护 message 列表来跟踪对话状态。列表中的元素可以是：

• HumanMessage：用户输入
• AIMessage：模型响应
• ToolMessage：工具调用结果

python复制message = [HumanMessage(content=query)]
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_output = tool_maps[tool_call["name"]](**tool_call["args"])
        message.append(ToolMessage(
            content=tool_output,
            tool_call_id=tool_call["id"],
            name=tool_call["name"]
        ))

3.2 循环终止条件

为防止无限循环，必须设置合理的终止条件。案例中采用了两种策略：

1. 最大迭代次数限制（5轮）
2. 模型主动返回最终结果（无 tool_calls）

在实际应用中，还可以考虑：

• 设置超时机制
• 监控对话长度
• 检测重复工具调用

4. 安全风险与防护措施

4.1 eval 函数的安全隐患

案例中的计算器工具直接使用 eval 执行表达式，这存在严重的安全风险。恶意用户可能通过精心构造的输入执行任意代码：

python复制# 危险示例
run_agent("计算：__import__('os').system('rm -rf /')")

4.2 安全加固方案

方案一：输入白名单验证

python复制import re

def safe_calculator(expression: str) -> str:
    if not re.match(r'^[\d\s+\-*/().]+$', expression):
        return "错误：表达式包含非法字符"
    try:
        return str(eval(expression))
    except:
        return "计算错误"

方案二：使用 ast 模块解析

python复制import ast

def safe_eval(expr):
    try:
        node = ast.parse(expr, mode='eval')
        if not all(isinstance(n, (ast.Constant, ast.BinOp, ast.UnaryOp)) 
                  for n in ast.walk(node)):
            raise ValueError("不安全的表达式")
        return str(eval(compile(node, '', 'eval')))
    except Exception as e:
        return f"错误：{str(e)}"

方案三：专用数学表达式解析库

python复制from pyparsing import (Word, nums, oneOf, ParseException, 
                      operatorPrecedence, opAssoc)

def parse_expression(expr):
    integer = Word(nums)
    operand = integer
    operator = oneOf("+ - * /")
    expr_stack = operatorPrecedence(operand, [
        (operator, 2, opAssoc.LEFT),
    ])
    try:
        result = expr_stack.parseString(expr, parseAll=True)
        return str(eval(expr))
    except ParseException:
        return "表达式解析错误"

5. 生产环境优化建议

5.1 性能优化

1. 向量数据库预热：提前加载 RAG 索引，避免每次查询都重新加载
2. 工具调用缓存：对相同参数的重复计算缓存结果
3. 异步工具调用：使用 asyncio 并行执行多个工具调用

python复制import asyncio

async def async_tool_call(tool, args):
    # 实现异步工具调用
    pass

async def run_agent_async(query):
    # 异步版本的 Agent 实现
    pass

5.2 可观测性增强

1. 日志记录：详细记录每轮对话的工具调用和模型响应
2. 性能监控：跟踪每个环节的耗时
3. 异常捕获：优雅处理各种边界情况

python复制import logging
import time

logging.basicConfig(filename='agent.log', level=logging.INFO)

def log_tool_call(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            duration = time.time() - start
            logging.info(
                f"Tool {func.__name__} called with args={args}, "
                f"kwargs={kwargs}, took {duration:.2f}s"
            )
            return result
        except Exception as e:
            logging.error(f"Tool {func.__name__} failed: {str(e)}")
            raise
    return wrapper

5.3 扩展性设计

1. 动态工具注册：支持运行时添加/移除工具
2. 权限控制：不同用户可使用不同工具集
3. 工具版本管理：支持工具的多版本共存

python复制class ToolManager:
    def __init__(self):
        self._tools = {}
        
    def register(self, name, tool):
        self._tools[name] = tool
        
    def unregister(self, name):
        del self._tools[name]
        
    def get_tools(self):
        return list(self._tools.values())

6. 典型问题排查指南

6.1 工具未被调用

可能原因：

1. 工具描述不够清晰
2. 模型能力不足
3. 提示词冲突

解决方案：

• 检查工具文档字符串是否完整
• 尝试更强大的模型
• 调整系统提示词

6.2 工具调用参数错误

可能原因：

1. 参数类型不匹配
2. 参数名称不一致
3. 模型理解偏差

解决方案：

• 确保工具文档中的参数描述准确
• 使用 JSON Schema 定义参数结构
• 在工具实现中添加参数验证

6.3 循环次数过多

可能原因：

1. 工具返回结果不完整
2. 模型无法理解工具输出
3. 任务过于复杂

解决方案：

• 优化工具输出格式
• 添加中间结果处理逻辑
• 设置更严格的终止条件

在实际部署 AI Agent 时，建议先在受限环境中充分测试，逐步扩大使用范围。对于关键业务场景，应该建立完善的人工审核和干预机制。