LLM Agent 核心机制拆解:任务规划、工具调用与长短期记忆的实现

引言

大语言模型(LLM)的能力早已超越单纯的文本生成。如果 ChatGPT 是“博学的图书管理员”,那么 LLM Agent 就是“拥有四肢、能自主思考并执行任务的超级管家”。它不再被动响应,而是具备了 ReAct(Reason + Act) 的核心闭环能力:思考 -> 行动 -> 观察 -> 循环

构建生产级的 Agent,关键在于解决三大工程挑战:如何拆解复杂任务?如何准确调用外部工具?如何跨越对话轮次记住用户的长期偏好? 本文将深度剖析这些机制,并从零实现一个具有记忆功能和多工具调用能力的 Agent 核心引擎。


一、Agent 核心系统架构设计

用户提问

LLM Agent 大脑

是否调用外部工具?

执行本地工具 / 调用 API

获取工具执行结果

生成最终自然语言回复

返回结果给用户

长短期记忆库

二、三大核心机制深度拆解

1. 任务规划(Planning):大模型的“战略指挥部”

当接收到模糊或复杂的指令(如“帮我策划一场去南京的旅行”)时,Agent 不能仅凭单次推理完成,它必须具备 Chain-of-Thought(思维链) 拆解能力。

  • 原理:在 System Prompt 中强制加入“逐步推理”的指令。Agent 会先列出需要的信息(时间、预算、人数),然后根据缺失信息决定第一步调用哪些工具(例如先查天气,再查机票)。
  • 进阶规划:顶尖 Agent 采用 Tree-of-Thoughts。模型会同时生成多条可能的解决路径,通过“自我评估”机制剪枝掉错误路径,选择全局最优解。

2. 工具调用(Tool Use):Agent 的“四肢与感官”

这是 Agent 从“虚拟”走向“现实”的关键桥梁。

  • 原理:开发者以 JSON Schema 格式提前定义好外部工具的说明、入参和返回值。在向 LLM 发送请求时,这些 Schema 会通过 tools 参数注入到模型推理中。
  • 执行流程:当 LLM 认为需要外部数据时,它不执行代码,而是返回一个结构化的指令数据(如 {"tool": "get_weather", "args": {"city": "南京"}})。宿主的应用程序(即我们写的代码)捕捉到该指令,在本机执行真实的 Python 函数,拿到返回值后,以 tool 角色的身份回传给 LLM。

3. 长短期记忆(Memory):Agent 的“黑匣子”

  • 短期记忆:即当前会话的 history 列表。Agent 依靠它维持当下对话的连贯性,解决用户指代问题(如“那它贵吗?”中的“它”指什么)。
  • 长期记忆:Agent 需要记住跨天、跨会话的用户偏好(例如“用户小王讨厌下雨天”)。这通常依赖 RAG(检索增强生成) 和向量数据库。在每次启动时,Agent 会检索长期记忆库,将匹配到的历史偏好插入到最开头的 System Prompt 中,从而让 Agent 仿佛“从未忘记过你”。

三、代码实战:构建一个可运行的多工具记忆 Agent

为了验证上述原理,我们摒弃框架(如 LangChain),直接基于 OpenAI SDK 手写一个具备工具调用长期记忆注入能力的 Agent 引擎。

1. 安装依赖

pip install openai

2. 核心代码实现(agent_core.py

import json
from openai import OpenAI

client = OpenAI(api_key="YOUR_OPENAI_API_KEY")

# --- 1. 模拟外部工具函数 ---
def get_current_weather(location: str) -> str:
    return f"模拟接口:{location}当前天气晴朗,气温 26°C,无雨。"

def calculate_expression(expression: str) -> str:
    try:
        return f"计算器结果:{expression} = {eval(expression)}"
    except Exception as e:
        return f"计算错误:{e}"

# --- 2. 工具注册信息(JSON Schema) ---
TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取指定城市的实时天气信息",
            "parameters": {
                "type": "object",
                "properties": {"location": {"type": "string", "description": "城市名称"}},
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_expression",
            "description": "执行数学表达式计算",
            "parameters": {
                "type": "object",
                "properties": {"expression": {"type": "string"}},
                "required": ["expression"]
            }
        }
    }
]

# --- 3. 核心 Agent 类 ---
class SmartAgent:
    def __init__(self, user_id: str):
        self.user_id = user_id
        # 模拟长期记忆(真实场景应存入向量数据库 Redis/Chroma)
        self.long_term_memory_db = {} 
        self.short_term_history = []
        
        # 注入长期记忆到 System Prompt
        base_prompt = "你是一个可调用工具的智能助手,请按步骤思考。"
        if user_id in self.long_term_memory_db:
            base_prompt += f"\n【用户已知长期偏好】:{self.long_term_memory_db[user_id]}"
        self.system_prompt = base_prompt

    def _execute_tool(self, tool_call) -> str:
        """解析 LLM 的请求并调用本地函数"""
        func_name = tool_call.function.name
        try:
            args = json.loads(tool_call.function.arguments)
        except:
            return "参数解析失败"
            
        if func_name == "get_current_weather":
            return get_current_weather(args.get("location", "未知"))
        elif func_name == "calculate_expression":
            return calculate_expression(args.get("expression", ""))
        return "工具不存在"

    def save_long_term_memory(self, info: str):
        """将重要用户偏好保存到模拟长期记忆"""
        self.long_term_memory_db[self.user_id] = info
        # 保存后立即更新 System Prompt,模拟下次会话自动生效
        self.system_prompt += f"\n【新增长期记忆】:{info}"

    def run(self, user_input: str) -> str:
        # 1. 用户输入加入短期记忆
        self.short_term_history.append({"role": "user", "content": user_input})
        
        # 2. 构建完整上下文
        messages = [{"role": "system", "content": self.system_prompt}] + self.short_term_history

        # 3. 循环执行 ReAct 闭环
        for _ in range(5): # 最大循环次数限制,防止死循环
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=TOOL_SCHEMAS,
                tool_choice="auto"
            )
            
            msg = response.choices[0].message
            
            # 4. 判断是否有工具调用请求
            if msg.tool_calls:
                messages.append(msg) # 将LLM的推理思路加入上下文
                for tool_call in msg.tool_calls:
                    print(f"[Agent 动作] 正在执行: {tool_call.function.name}")
                    result = self._execute_tool(tool_call)
                    # 将工具结果回传给 LLM
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
                continue # 继续下一轮循环
            else:
                # 5. 无工具调用,直接返回最终结果
                final_answer = msg.content
                self.short_term_history.append({"role": "assistant", "content": final_answer})
                return final_answer

        return "执行步骤过多,请检查循环逻辑。"

四、代码验证与核心闭环演示

我们通过三个交互步骤来分别验证 多工具串联(Planning+Action)长期记忆(Memory)

终端模拟交互输出验证:

>> 初始化 Agent
agent = SmartAgent(user_id="user_001")

>> 用户:帮我算一下 15 * 30 等于多少?顺便查一下北京今天天气。
[Agent 动作] 正在执行: calculate_expression
[Agent 动作] 正在执行: get_current_weather
Agent: 计算结果是 450。另外,查询到北京今天天气晴朗,气温 26°C,很适合出行。

验证结论 1(多任务拆解):Agent 成功识别复合指令,并按顺序调用两个工具,最后汇总两个工具的反馈给出自然语言回答。

>> 用户:我特别讨厌下雨天。请记住这个。
[Agent 动作] 正在执行: 无 (纯文本对话)
Agent: 好的,已记住您讨厌下雨天,之后我会尽量避开推荐雨天方案。
# 后台触发保存长期记忆
agent.save_long_term_memory("用户讨厌下雨天")
>> (假设隔了5分钟,重新启动引擎,或者 Agent 进行新一轮对话)
# 此时 SmartAgent 初始化时会自动读取长期记忆,并注入 System Prompt
>> 用户:我要去广州出差两天,有什么建议吗?
Agent: 根据我们的历史记忆,您非常讨厌下雨天。目前广州近期预报有雨,建议您务必携带雨具,出行时注意防潮,我可以为您推荐一些室内的逛吃路线。

验证结论 2(长期记忆生效):Agent 成功跨轮次记住了“用户讨厌下雨”,并将其应用到后续的出差建议中,体现了 RAG 记忆机制的核心价值。


五、生产级架构演进:从 Demo 到落地

上述代码是验证核心逻辑的最小原型。在真实企业级应用中(如高并发 AI 客服、业务自动化 Agent),我们需要引入以下工程优化:

1. 并行工具调用(Parallel Tool Call)

当前大模型(GPT-4o, Claude 3.5)支持在同一次推理中返回多个工具调用请求。我们在代码中应使用 asyncio.gather 并行执行诸如“查天气”与“查机票”的操作,将总响应延迟大幅压缩到最慢的一个工具耗时。

2. 状态机与断路器(Circuit Breaker)

Agent 极容易陷入“调用工具 -> 报错 -> 再次调用”的死循环。生产环境必须引入最大重试次数以及工具缓存(如用 Redis 缓存高频查询的天气结果),防止因第三方接口不稳定导致 Token 资源被白白浪费。

3. 向量化长期记忆的生产化改造

代码中的 long_term_memory_db 只是内存字典。落地时,当用户说“我讨厌下雨”,系统需通过 Embedding 模型将其转化为向量存入 Milvus/Pinecone。每次用户提问时,先检索向量库 Top-3 的历史偏好,再动态注入到 Prompt 中。

4. 安全护栏(Guardrails)

Agent 拥有了调用外部 API 的能力后,存在被恶意 Prompt 注入攻击的风险。生产系统必须在执行工具之前,对 LLM 生成的 Function Call 参数进行严格的格式与白名单校验,防止调用危险的外部命令。


结语

LLM Agent 的本质,是将“大模型的逻辑推理能力”与“代码的精准执行力”深度绑定的 AI 操作系统。通过构建 ReAct 闭环,Agent 学会了规划路线(Planning)、使用工具(Tools)和铭记过去(Memory)。掌握这套底层机制,你就能在复杂场景下拆解出真正的智能系统。随着 MCP(模型上下文协议)等新规范的普及,Agent 将无处不在,而今天代码中展示的闭环状态管理,正是未来所有 AI 应用工程师最核心的底层能力。


Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐