前言

2026年,AI Agent 已经成为大模型应用最热门的落地形态。GPT-5.5 专为 Agent 时代设计,DeepSeek-V4 百万上下文让 Agent 能处理更复杂的任务——但 Agent 的核心架构其实并不神秘。

Agent = 大模型 + 工具调用 + 记忆管理 + 规划推理

拆开来看,每个模块都不复杂。本文就用纯 Python,从零手写一个可运行的 AI Agent,不依赖 LangChain、AutoGPT 等框架,让你真正理解 Agent 的工作原理。

前置知识: 了解 Python 基础,知道什么是 API 调用即可。
使用模型: 代码兼容 OpenAI 兼容接口(支持 Function Calling 的模型均可),可以用 DeepSeek、通义千问、GLM 等。


1. 核心架构:Agent 到底在做什么?

在动手之前,先建立 Agent 的整体认知。一个标准的 ReAct(Reasoning + Acting)模式 Agent 的工作循环是:

用户提问
    ↓
Agent 思考(Thought)→ 本次要做什么
    ↓
Agent 行动(Action)→ 调用某个工具(或直接回答)
    ↓
收到结果(Observation)→ 工具返回了什么
    ↓
再思考 → 再行动 → 再观察 → ...
    ↓
最终回答(Final Answer)

这就是 ReAct 模式,由 Google 在 2022 年提出,是当前大多数 Agent 的基础范式。

我们的实现将围绕这个循环展开:

┌───────────────────────────────────────────┐
│               Agent 主循环                  │
│                                           │
│  Thought → Action → Observation           │
│         ↕ (重复直到得出答案)                │
│             +                              │
│       记忆管理 (控制上下文长度)               │
└───────────────────────────────────────────┘

2. 第一步:封装 LLM 调用接口

我们先做一个统一的模型调用层,这样后续切换模型时只需改配置。

import json
import os
from typing import Optional

import requests


class LLMClient:
    """大模型调用封装,兼容 OpenAI API 格式"""

    def __init__(self, api_key: str, base_url: str, model: str):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.model = model

    def chat(
        self,
        messages: list[dict],
        tools: Optional[list[dict]] = None,
        temperature: float = 0.7,
        max_tokens: int = 4096,
    ) -> dict:
        """调用对话接口"""
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
        }
        if tools:
            payload["tools"] = tools

        resp = requests.post(
            f"{self.base_url}/chat/completions",
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
            json=payload,
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]

使用示例:

# 配置(以 DeepSeek 为例)
llm = LLMClient(
    api_key=os.environ["DASHSCOPE_API_KEY"],       # 你的 API Key
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 兼容接口地址
    model="qwen-plus",                              # 模型名称
)

response = llm.chat([
    {"role": "user", "content": "你好,请介绍一下自己"}
])
print(response["content"])

这里用 requests 而不是 openai 库,是为了零依赖、让代码更透明。


3. 第二步:工具系统——让 Agent 拥有"双手"

Agent 能做事的关键是工具(Tool)。工具就是一段代码,有名字、描述和参数,模型通过函数调用(Function Calling)来触发它。

3.1 工具定义

我们用一个统一的接口来定义工具:

from typing import Callable, Any


class Tool:
    """工具定义"""

    def __init__(
        self,
        name: str,
        description: str,
        parameters: dict,
        func: Callable[..., Any],
    ):
        self.name = name
        self.description = description
        self.parameters = parameters  # JSON Schema 格式
        self.func = func

    def to_openai_tool(self) -> dict:
        """转为 OpenAI tools 格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

    def run(self, **kwargs) -> str:
        """执行工具并返回人类可读的结果"""
        try:
            result = self.func(**kwargs)
            return str(result)
        except Exception as e:
            return f"工具执行错误: {e}"

3.2 自定义几个实用工具

import random
import datetime


def get_weather(city: str) -> str:
    """查询城市天气(模拟)"""
    weathers = ["晴 ☀️", "多云 ⛅", "小雨 🌦️", "阴天 ☁️"]
    temps = random.randint(15, 35)
    return f"{city} 当前天气:{random.choice(weathers)},温度 {temps}°C"


def calculate(expression: str) -> str:
    """执行数学计算"""
    # 安全起见,只允许数字和运算符
    allowed = set("0123456789+-*/()., ")
    if not all(c in allowed for c in expression):
        return "表达式包含非法字符"
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"


def get_current_time(format_str: str = "%Y-%m-%d %H:%M:%S") -> str:
    """获取当前时间"""
    return datetime.datetime.now().strftime(format_str)


# 注册工具
TOOLS = [
    Tool(
        name="get_weather",
        description="查询指定城市的当前天气",
        parameters={
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称,如 北京、上海、深圳",
                }
            },
            "required": ["city"],
        },
        func=get_weather,
    ),
    Tool(
        name="calculate",
        description="执行数学表达式计算,支持加减乘除和括号",
        parameters={
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式,如 (3+5)*2",
                }
            },
            "required": ["expression"],
        },
        func=calculate,
    ),
    Tool(
        name="get_current_time",
        description="获取当前日期和时间",
        parameters={
            "type": "object",
            "properties": {},
        },
        func=get_current_time,
    ),
]

# 索引工具以便快速查找
TOOL_MAP = {t.name: t for t in TOOLS}

关键点: 工具的描述(description)和参数(parameters)必须写清楚,因为模型靠这些文本来决定调用哪个工具、传什么参数。写得越清楚,模型调用越准确。


4. 第三步:Agent 主循环——思考-行动-观察

这是整个 Agent 最核心的部分。每次循环:

  1. 历史 + 新观察发给模型
  2. 模型返回:要调用工具,还是直接回答
  3. 如果要调工具 → 执行业务代码 → 把结果放回消息
  4. 循环直到模型直接回答

4.1 系统提示词

SYSTEM_PROMPT = """你是 AI Agent,你拥有调用工具的权限。

请遵循 ReAct 模式工作:

1. 仔细分析用户的请求
2. 如果需要调用工具,按以下格式输出你的思考过程:
   Thought: 分析当前情况,判断下一步需要做什么
   Action: 调用合适的工具
   (系统将执行工具并返回结果)

3. 如果已有足够信息,直接回答用户

可用工具:
{tools_description}

请开始。"""

4.2 Agent 主类

class Agent:
    """ReAct 模式 AI Agent"""

    def __init__(self, llm: LLMClient, tools: list[Tool], max_iterations: int = 10):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.tool_definitions = [t.to_openai_tool() for t in tools]
        self.max_iterations = max_iterations
        self.messages = []

    def _build_tools_description(self) -> str:
        """为系统提示词生成工具描述"""
        descs = []
        for t in self.tools.values():
            params = t.parameters.get("properties", {})
            param_str = ", ".join(
                f"{name}({info.get('type', '未知')})" for name, info in params.items()
            )
            descs.append(f"- {t.name}({param_str}): {t.description}")
        return "\n".join(descs)

    def run(self, user_input: str) -> str:
        """运行 Agent,处理用户输入并返回最终结果"""
        # 初始化对话
        self.messages = [
            {
                "role": "system",
                "content": SYSTEM_PROMPT.format(
                    tools_description=self._build_tools_description()
                ),
            },
            {"role": "user", "content": user_input},
        ]

        for iteration in range(self.max_iterations):
            print(f"\n--- 迭代 {iteration + 1} ---")

            # 调用模型
            response = self.llm.chat(
                messages=self.messages,
                tools=self.tool_definitions,
            )

            # 检查是否有工具调用
            if "tool_calls" in response and response["tool_calls"]:
                # 模型决定调用工具
                self.messages.append(response)

                for tool_call in response["tool_calls"]:
                    tool_name = tool_call["function"]["name"]
                    try:
                        arguments = json.loads(tool_call["function"]["arguments"])
                    except json.JSONDecodeError:
                        arguments = {}

                    print(f"  → 调用工具: {tool_name}({arguments})")

                    # 执行工具
                    if tool_name in self.tools:
                        result = self.tools[tool_name].run(**arguments)
                    else:
                        result = f"错误:未找到工具 {tool_name}"

                    print(f"  ← 工具结果: {result[:100]}...")

                    # 把工具结果放回对话
                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call["id"],
                        "content": result,
                    })
            else:
                # 模型直接回答,就是最终结果
                final_answer = response.get("content", "")
                print(f"  ← 最终回答: {final_answer[:100]}...")
                return final_answer

        return "已达最大迭代次数,无法完成请求。"

4.3 试跑一下

# 初始化 Agent
agent = Agent(llm=llm, tools=TOOLS)

# 测试 1:多工具协同
result = agent.run("北京现在几度?帮我算一下从今天到月底还有几天?")
print(f"\n最终结果:\n{result}")

运行这个例子,Agent 会:

  1. 调用 get_weather 查询北京天气
  2. 调用 get_current_time 获取今天日期
  3. 调用 calculate 计算剩余天数
  4. 综合结果完整回答

这正是 多步推理 + 多工具协作——Agent 最核心的能力。


5. 第四步:记忆管理

Agent 跑几轮后,messages 会变得很长。对于上下文窗口有限的模型(以及节省 tokens 成本),我们需要记忆管理策略。

这里实现最简单的滑动窗口 + 摘要策略:

class MemoryManager:
    """记忆管理器:控制上下文长度"""

    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens

    def trim_messages(
        self,
        messages: list[dict],
        system_prompt: str,
        reserve_last: int = 4,
    ) -> list[dict]:
        """裁剪消息列表,保留最近 N 条"""
        # 简化版:保留 system + 最近 reserve_last 条消息
        system_msg = next((m for m in messages if m["role"] == "system"), None)
        recent = messages[-reserve_last:] if len(messages) > reserve_last else messages

        trimmed = []
        if system_msg:
            trimmed.append(system_msg)

        # 如果裁剪后丢失了关键信息,添加一条摘要
        if len(messages) > reserve_last + 1:
            trimmed.append({
                "role": "system",
                "content": f"[系统摘要] 前面讨论了 {len(messages) - reserve_last - 1} 轮对话,"
                           f"已省略。请基于最近的信息继续。"
            })

        trimmed.extend(recent)
        return trimmed

在 Agent run() 方法的每次迭代后加入:

self.messages = MemoryManager(max_tokens=8000).trim_messages(
    self.messages, system_prompt
)

实际生产场景中,更复杂的做法包括:

  • LLM 摘要:定期让模型对历史对话做摘要,替换掉原始内容
  • 向量检索:把关键信息存入向量数据库,需要时召回
  • 结构化记忆:把工具调用结果按类型存入不同槽位

6. 完整示例:一个有多轮记忆的对话

agent = Agent(llm=llm, tools=TOOLS)

# 第一轮
print("=== 第一轮 ===")
r1 = agent.run("我的名字叫张三,我是北京的程序员")
print(f"Agent: {r1}\n")

# 第二轮 —— Agent 应该记得之前的信息
print("=== 第二轮 ===")
r2 = agent.run("我叫什么名字?我在哪个城市?")
print(f"Agent: {r2}\n")

# 第三轮 —— 查询本地天气
print("=== 第三轮 ===")
r3 = agent.run("帮我看看我这里的天气怎么样?")
print(f"Agent: {r3}\n")

由于我们把记忆放入了 messages 对话历史中,Agent 会在上下文中看到之前的对话,从而回答:"你叫张三,在北京……“然后调用 `get_weather(city=“北京”)”。


7. 进阶:让 Agent 更聪明

7.1 错误重试

当工具调用失败时,不直接终止,而是让模型自己尝试修复:

def run_with_retry(self, user_input: str, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            return self.run(user_input)
        except Exception as e:
            self.messages.append({
                "role": "user",
                "content": f"上一步出错了:{e},请换一种方式重试。"
            })
    return "重试多次仍失败"

7.2 动态工具注册

让 Agent 在运行过程中自己创建新工具(动态生成 Python 函数):

def register_dynamic_tool(self, name: str, code: str, description: str):
    """动态注册一个工具"""
    local_scope = {}
    exec(code, {"__builtins__": {}}, local_scope)
    func = local_scope.get(name)
    if not func or not callable(func):
        raise ValueError("工具代码必须定义一个同名函数")

    tool = Tool(
        name=name,
        description=description,
        parameters={
            "type": "object",
            "properties": {},
        },
        func=func,
    )
    self.tools[name] = tool
    self.tool_definitions.append(tool.to_openai_tool())

这样 Agent 就可以先让模型写一段代码,然后注册为工具反复使用——这是一个简单的 Code as Tool 模式。


8. 总结与展望

我们从头实现的 Agent 虽然只有不到 200 行核心代码,但已经包含了现代 Agent 的三大核心机制:

模块 实现 说明
工具调用 Tool 类 + Function Calling 模型决定调什么工具、传什么参数
推理规划 ReAct 循环 思考→行动→观察→再思考
记忆管理 滑动窗口 + 摘要 控制上下文长度,保留关键信息

这个简陋的 Agent 可以直接跑起来,做天气查询、数学计算、日期计算等任务。如果你想让它更强大:

  • 接入 MCP (Model Context Protocol) 协议,统一管理外部工具
  • 加入 子 Agent 路由,一个 Agent 搞不定就派子 Agent
  • 使用 结构化输出 替代自由文本的 thought 输出,更稳定

最核心的认知是: Agent 不是一个黑盒子,它的本质就是 循环(Loop)+ 工具(Tool)+ 记忆(Memory)。框架让你搭建得更快,但自己手写一遍,才能真正理解每个环节的设计取舍。

如果你想看某个方向的深入实现(比如 MCP 集成、子 Agent 调度、Graph 编排),欢迎评论区留言。


本文配套代码仓库: github.com/your-name/agent-from-scratch(示例地址,欢迎 fork 实战)

全部代码在 Python 3.12+ 下测试通过,使用的依赖仅 requests,零框架依赖。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐