AI Agent大师之路：从零构建具备规划与记忆能力的自主智能体

1. 引言：Agent不仅仅是“调API”

2026年，AI Agent已从“玩具”走向“工具”。但市面上多数教程停留在“用LangChain调个工具”的层面，真正的自主智能体需要解决三个核心问题：

基于ReAct（Reasoning + Acting）范式的Agent

。所有代码均可运行，依赖开源库（LangChain、ChromaDB、OpenAI兼容API）。

我们设计的Agent架构如下（Mermaid图）：

pip install langchain chromadb openai tiktoken python-dotenv

我们实现三个工具：计算器、模拟搜索（可替换为真实API）、文件写入。

from langchain.tools import BaseTool
from langchain.callbacks.manager import AsyncCallbackManagerForToolRun
import subprocess
import json

class CalculatorTool(BaseTool):
    name = "calculator"
    description = "执行数学表达式计算，输入为字符串表达式，返回数值结果。"

    def _run(self, expr: str) -> str:
        try:
            # 安全计算，仅允许基本运算
            allowed = {"+", "-", "*", "/", "(", ")", " ", ".", "0","1","2","3","4","5","6","7","8","9"}
            if not set(expr).issubset(allowed):
                return "错误：表达式包含非法字符"
            result = eval(expr, {"__builtins__": None}, {})
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算错误: {str(e)}"

class SearchTool(BaseTool):
    name = "web_search"
    description = "模拟网络搜索，输入查询关键词，返回模拟结果（可替换为真实API）。"

    def _run(self, query: str) -> str:
        # 模拟数据，实际可改为调用SerpAPI或Bing
        mock_db = {
            "北京天气": "北京今日晴，气温25°C，湿度40%。",
            "AI Agent": "AI Agent是能够自主决策和执行任务的智能体，典型框架包括ReAct、AutoGPT。"
        }
        return mock_db.get(query, f"未找到关于'{query}'的信息，建议换关键词。")

class FileWriterTool(BaseTool):
    name = "file_writer"
    description = "将内容写入指定文件，输入格式为'文件路径|内容'。"

    def _run(self, input_str: str) -> str:
        try:
            path, content = input_str.split("|", 1)
            with open(path.strip(), "w", encoding="utf-8") as f:
                f.write(content)
            return f"成功写入文件: {path.strip()}"
        except Exception as e:
            return f"写入失败: {str(e)}"

使用ChromaDB存储任务嵌入，便于后续检索。

import chromadb
from chromadb.utils import embedding_functions
import uuid

class AgentMemory:
    def __init__(self, collection_name="agent_memory", persist_dir="./chroma_db"):
        self.client = chromadb.PersistentClient(path=persist_dir)
        # 使用默认的all-MiniLM-L6-v2嵌入（或替换为OpenAI嵌入）
        self.embed_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="all-MiniLM-L6-v2"
        )
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embed_fn
        )

    def add_experience(self, task: str, result: str, metadata: dict = None):
        """存储任务及其结果，用于后续检索"""
        doc_id = str(uuid.uuid4())
        self.collection.add(
            documents=[f"任务: {task}\n结果: {result}"],
            metadatas=[metadata or {}],
            ids=[doc_id]
        )
        return doc_id

    def retrieve_similar(self, query: str, k: int = 3):
        """检索与当前查询最相似的历史经验"""
        results = self.collection.query(query_texts=[query], n_results=k)
        return results['documents'][0] if results['documents'] else []

4.3 分层规划器（Hierarchical Planner）

规划器将用户目标分解为子任务，并生成依赖图。

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI  # 可使用任何兼容Chat模型

class HierarchicalPlanner:
    def __init__(self, llm):
        self.llm = llm
        self.decompose_prompt = PromptTemplate(
            input_variables=["goal", "history"],
            template="""你是一个任务规划专家。给定最终目标，请将其分解为3-5个可执行的子任务，并说明子任务之间的依赖顺序。
当前已有的历史经验（可能相关）：
{history}

目标：{goal}

请以JSON数组格式输出，每个任务包含：id, description, dependencies(依赖的id列表)。
只输出JSON数组，不要其他内容。"""
        )
        self.chain = LLMChain(llm=self.llm, prompt=self.decompose_prompt)

    def plan(self, goal: str, memory_hints: list):
        history_text = "\n".join(memory_hints) if memory_hints else "无相关历史。"
        response = self.chain.run(goal=goal, history=history_text)
        try:
            tasks = json.loads(response)
            return tasks  # list of dict
        except:
            # 降级方案：返回单个任务
            return [{"id": "task1", "description": goal, "dependencies": []}]

执行器对每个子任务循环执行“思考-行动-观察”，直至完成或达到最大步数。

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate

class ReActExecutor:
    def __init__(self, llm, tools, memory: AgentMemory):
        self.llm = llm
        self.tools = tools
        self.memory = memory
        # ReAct提示模板（LangChain内置，但自定义以增加记忆注入）
        self.react_prompt = PromptTemplate.from_template(
            """你是一个自主智能体，可以调用以下工具：
{tools}

工具名称: {tool_names}

你拥有一个长期记忆库，之前执行相似任务的经验如下（可能相关）：
{memory_context}

当前子任务: {task}

请使用以下格式回答：
Thought: 你关于当前子任务的思考
Action: 工具名称（必须是上述之一）
Action Input: 工具的输入参数

当你完成子任务时，使用：
Thought: 我已完成任务
Final Answer: 最终答案

开始！
{agent_scratchpad}"""
        )
        # 创建Agent
        self.agent = create_react_agent(llm, tools, self.react_prompt)

    def run_task(self, task_desc: str, max_steps=5):
        """执行单个子任务，返回结果"""
        # 检索相关记忆
        similar = self.memory.retrieve_similar(task_desc, k=2)
        memory_context = "\n".join(similar) if similar else "无相关经验。"

        # 构造agent执行器
        agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            verbose=True,
            max_iterations=max_steps,
            handle_parsing_errors=True,
            # 传递额外上下文到prompt（通过agent_kwargs）
        )
        # 由于我们自定义prompt，需要动态传入memory_context，这里简化：直接使用run并带参数
        # 更稳健方式：使用agent.run的input加上memory_context，但LangChain的create_react_agent不支持直接注入自定义变量
        # 我们改用LLMChain直接调用，但为了演示，我们重写一个简单ReAct循环
        # 为了代码清晰，此处展示标准AgentExecutor用法，但需要调整prompt模板
        # 实际生产中，建议直接继承Agent类，此处我们快速实现一个自定义ReAct循环
        return self._run_react_loop(task_desc, memory_context, max_steps)

    def _run_react_loop(self, task, memory_context, max_steps):
        """手工实现ReAct循环，完全控制输入输出"""
        scratchpad = ""
        for step in range(max_steps):
            # 填充prompt
            prompt_text = self.react_prompt.format(
                tools="\n".join([f"{t.name}: {t.description}" for t in self.tools]),
                tool_names=", ".join([t.name for t in self.tools]),
                memory_context=memory_context,
                task=task,
                agent_scratchpad=scratchpad
            )
            response = self.llm.invoke(prompt_text)
            output = response.content if hasattr(response, 'content') else str(response)
            print(f"Step {step+1} output:\n{output}")

            # 解析输出
            if "Final Answer" in output:
                # 提取最终答案
                final = output.split("Final Answer")[-1].strip()
                # 存储经验到记忆
                self.memory.add_experience(task, final)
                return final

            # 提取Action和Action Input
            import re
            action_match = re.search(r"Action: (.*?)(?:\n|$)", output)
            input_match = re.search(r"Action Input: (.*?)(?:\n|$)", output)
            if action_match and input_match:
                action_name = action_match.group(1).strip()
                action_input = input_match.group(1).strip()
                # 执行工具
                tool = next((t for t in self.tools if t.name == action_name), None)
                if tool:
                    observation = tool.run(action_input)
                else:
                    observation = f"错误：未找到工具 {action_name}"
            else:
                observation = "无法解析行动，请重试。"

            scratchpad += f"\n{output}\nObservation: {observation}\n"

        # 达到最大步数仍未完成
        final_msg = "任务未能在最大步数内完成。"
        self.memory.add_experience(task, final_msg)
        return final_msg

class MasterAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.memory = AgentMemory()
        self.planner = HierarchicalPlanner(llm)
        self.executor = ReActExecutor(llm, tools, self.memory)

    def run(self, goal: str):
        # 检索全局记忆作为规划参考
        hints = self.memory.retrieve_similar(goal, k=3)
        tasks = self.planner.plan(goal, hints)
        print(f"规划生成 {len(tasks)} 个子任务: {[t['description'] for t in tasks]}")

        results = {}
        # 按依赖顺序执行（简单拓扑排序）
        # 此处简化：顺序执行，但记录依赖
        for task in tasks:
            # 检查依赖是否完成
            deps = task.get('dependencies', [])
            if deps:
                # 简单等待依赖完成（实际可设计为异步）
                pass
            print(f"\n=== 执行子任务: {task['description']} ===")
            result = self.executor.run_task(task['description'])
            results[task['id']] = result
            # 将结果加入记忆（已在executor内添加，但可额外添加整体）
        # 最终汇总
        final_summary = f"目标 '{goal}' 已完成。各子任务结果：\n" + "\n".join([f"{tid}: {res}" for tid, res in results.items()])
        return final_summary

from langchain.llms import OpenAI

# 初始化LLM（使用本地vLLM或云API）
llm = OpenAI(
    model_name="deepseek-chat",  # 替换为你的模型
    openai_api_base="https://api.deepseek.com/v1",
    openai_api_key="your-key",
    temperature=0.3
)

tools = [CalculatorTool(), SearchTool(), FileWriterTool()]
agent = MasterAgent(llm, tools)

goal = "请帮我查询北京今天的天气，然后将天气信息写入文件'weather.txt'，并计算华氏度（摄氏*9/5+32）同时写入同一文件。"
result = agent.run(goal)
print("\n=== 最终结果 ===\n", result)

规划生成 3 个子任务: ['查询北京天气', '计算华氏度', '写入文件']
=== 执行子任务: 查询北京天气 ===
Step 1 output:
Thought: 我需要使用web_search工具查询北京天气。
Action: web_search
Action Input: 北京天气
Observation: 北京今日晴，气温25°C，湿度40%。
Step 2 output:
Thought: 我已获得天气信息。
Final Answer: 北京今日晴，气温25°C，湿度40%。
...
最终结果包含文件写入成功。

当前规划器基于LLM一次生成所有子任务，但复杂目标可能需要动态重规划。可引入

循环，每完成一个子任务评估是否调整后续计划。

我们使用简单向量相似度，但可增加

，让近期和高频经验权重更大。

实际部署时，应限制工具权限（如文件写入路径白名单），并加入人类介入（HITL）机制。

在单次任务（3个子任务，每个平均2步推理）下，总耗时约5-8秒（取决于LLM速度）。记忆检索<100ms。

本文实现的Agent已具备基础自主性，但距离“大师级”还需：

人工智能

学习如何用200+行Python代码构建具备规划与记忆能力的自主智能体。文章详解ReAct范式、任务分解、记忆检索等核心技术，提供完整代码实现，包括自定义工具集、向量记忆库和分层规划器。通过真实API环境测试多步推理能力，适合开发者快速掌握AI智能体开发核心技能。

强化学习

智能体

DeepSeek

Python

Agent

JSON

2026上云采购 | AI焕新·智启新局

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

AI Agent大师之路：从零构建具备规划与记忆能力的自主智能体

AI Agent大师之路：从零构建具备规划与记忆能力的自主智能体

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐