Agent_01_Agent核心架构

Agent 2026年1月·agent Python

创建于 2026/1/3更新于 2026/1/3

最简定义：

Agent = LLM（大脑）+ 记忆（Memory）+ 工具（Tools）+ 规划（Planning）

这四件事，每一件都可以单独拿出来优化、替换、调试。很多人觉得 Agent"不稳定"、"不可靠"，本质上是因为把这四件事混在一起写，出了问题不知道是记忆漏了、工具错了、还是规划乱了。

先看一个常见的反面做法——把所有逻辑塞进一个 prompt：

python

# 反例：一个 prompt 干所有事
system_prompt = """
你是一个数据分析助手。你需要：
1. 记住用户的偏好（之前说过喜欢柱状图而不是折线图）
2. 如果用户提到数据，用 pandas 读取 csv 文件
3. 如果用户要求画图，用 matplotlib 画图
4. 如果用户的问题不清楚，先问清楚再行动
5. 最后输出分析报告
"""

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "system", "content": system_prompt},
              {"role": "user", "content": user_input}]
)

这段代码的问题很明显：没有真正的工具调用（模型只能"告诉"你该做什么，但做不到）；记忆是假的（每次调用都是独立请求，上一轮的"偏好"下一轮就没了）；五个要求写在一个 prompt 里，模型不知道哪个先做哪个后做；输出不对的时候，你不知道是 prompt 写得不好、模型理解错了、还是该给模型加个工具。

把这四件事拆开，每件交给专门的组件处理：

text

┌─────────────────────────────────────────────────┐
│                   Agent Loop                     │
│                                                  │
│   ┌──────────┐                                   │
│   │  规划     │ ← 用户目标: "分析这份销售数据"    │
│   │ Planner  │   → 拆解: 读数据 → 清洗 → 分析    │
│   └────┬─────┘                                   │
│        │ 执行步骤                                  │
│   ┌────▼─────┐    ┌──────────┐                   │
│   │  工具     │    │  记忆     │                   │
│   │  Tools   │◄──►│ Memory   │                   │
│   │ read_csv │    │ 偏好:柱状图│                   │
│   │ analyze  │    │ 历史:Q3   │                   │
│   │ plot     │    └──────────┘                   │
│   └────┬─────┘                                   │
│        │ 结果                                     │
│   ┌────▼─────┐                                   │
│   │   LLM    │ ← 综合工具和记忆的信息，生成回复   │
│   │ (大脑)   │                                   │
│   └──────────┘                                   │
└─────────────────────────────────────────────────┘

模型还是那个模型，但给了它记忆让它能回忆、给了它工具让它能动手、给了它规划让它知道先做什么后做什么。Agent 不是"更聪明的 LLM"，而是"更有组织的 LLM"。

最简 Agent：四大组件齐全

这个例子展示了最精简但完整的 Agent 架构。每个组件独立、可替换、可调试。

python

"""
最简 Agent 架构 —— LLM + Memory + Tools + Planner
每个组件都是独立接口，可以随时替换实现。
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any


class LLM(ABC):
    """LLM 接口 —— 所有模型实现这个契约"""

    @abstractmethod
    def generate(self, prompt: str, system: str = "", **kwargs) -> str:
        ...

    @abstractmethod
    def generate_with_tools(
        self, prompt: str, tools: list[dict], system: str = "", **kwargs
    ) -> tuple[str, list[dict]]:
        """返回 (文本, 工具调用列表)。工具调用格式: [{"name": "xxx", "args": {...}}]"""
        ...


class OpenAILLM(LLM):
    """OpenAI 模型实现"""

    def __init__(self, model: str = "gpt-4o-mini", api_key: str = ""):
        self.model = model
        self.api_key = api_key

    def generate(self, prompt: str, system: str = "", **kwargs) -> str:
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        # 实际场景：调用 openai SDK
        # response = openai.ChatCompletion.create(model=self.model, messages=messages)
        # return response.choices[0].message.content
        print(f"[LLM] 调用 {self.model}: {prompt[:80]}...")
        return "(模拟回复)"

    def generate_with_tools(
        self, prompt: str, tools: list[dict], system: str = "", **kwargs
    ) -> tuple[str, list[dict]]:
        print(f"[LLM] 调用 {self.model} (带工具): {prompt[:80]}...")
        return ("我帮你分析数据", [{"name": "read_csv", "args": {"path": "data.csv"}}])


class Memory(ABC):
    """记忆接口 —— 管读写"""

    @abstractmethod
    def add(self, key: str, value: Any) -> None:
        ...

    @abstractmethod
    def get(self, key: str) -> Any | None:
        ...

    @abstractmethod
    def get_all(self) -> dict:
        ...

    @abstractmethod
    def clear(self) -> None:
        ...


class DictMemory(Memory):
    """最简单的内存 —— 就是一个字典"""

    def __init__(self, max_size: int = 50):
        self._store: dict[str, Any] = {}
        self.max_size = max_size

    def add(self, key: str, value: Any) -> None:
        if len(self._store) >= self.max_size:
            oldest_key = next(iter(self._store))
            del self._store[oldest_key]
        self._store[key] = value

    def get(self, key: str) -> Any | None:
        return self._store.get(key)

    def get_all(self) -> dict:
        return dict(self._store)

    def clear(self) -> None:
        self._store.clear()

    def __repr__(self):
        return f"DictMemory({self._store})"


@dataclass
class ToolResult:
    """工具执行结果"""
    success: bool
    output: Any
    error: str = ""


class ToolRegistry:
    """工具注册表 —— 管着所有可用工具"""

    def __init__(self):
        self._tools: dict[str, Any] = {}

    def register(self, name: str, func: Any, description: str = "") -> None:
        self._tools[name] = {"func": func, "description": description}

    def execute(self, name: str, **kwargs) -> ToolResult:
        if name not in self._tools:
            return ToolResult(success=False, output=None, error=f"未知工具: {name}")
        try:
            result = self._tools[name]["func"](**kwargs)
            return ToolResult(success=True, output=result)
        except Exception as e:
            return ToolResult(success=False, output=None, error=str(e))

    def get_schemas(self) -> list[dict]:
        """返回所有工具的 schema，给 LLM 看"""
        return [
            {"name": name, "description": info["description"]}
            for name, info in self._tools.items()
        ]


class Planner(ABC):
    """规划接口 —— 把目标拆成步骤"""

    @abstractmethod
    def plan(self, goal: str, context: str = "") -> list[str]:
        """返回步骤列表"""
        ...


class SimplePlanner(Planner):
    """最简规划器 —— 用 LLM 生成步骤"""

    def __init__(self, llm: LLM):
        self._llm = llm

    def plan(self, goal: str, context: str = "") -> list[str]:
        system = (
            "你是一个任务规划专家。把用户的目标拆解为可执行的步骤。"
            "每步用一句话描述，返回列表格式。"
        )
        prompt = f"目标：{goal}\n上下文：{context}\n请拆解为步骤："
        result = self._llm.generate(prompt, system=system)
        return [s.strip() for s in result.split("\n") if s.strip()]


@dataclass
class AgentConfig:
    """Agent 配置"""
    max_steps: int = 10
    verbose: bool = True


class Agent:
    """
    Agent 主循环 —— 接收目标 → 规划 → 逐步执行 → 返回结果
    """

    def __init__(
        self,
        llm: LLM,
        memory: Memory,
        tools: ToolRegistry,
        planner: Planner,
        config: AgentConfig = None,
    ):
        self._llm = llm
        self._memory = memory
        self._tools = tools
        self._planner = planner
        self._config = config or AgentConfig()

    def run(self, goal: str) -> str:
        # 规划
        context = ", ".join(f"{k}={v}" for k, v in self._memory.get_all().items())
        steps = self._planner.plan(goal, context=context)

        if self._config.verbose:
            print(f"=== 规划完成，共 {len(steps)} 步 ===")
            for i, step in enumerate(steps, 1):
                print(f"  {i}. {step}")

        # 执行循环
        for step_idx, step in enumerate(steps, 1):
            if step_idx > self._config.max_steps:
                return "达到最大步数限制，任务未完成。"

            if self._config.verbose:
                print(f"\n--- 执行第 {step_idx} 步: {step} ---")

            tool_schemas = self._tools.get_schemas()
            thought, tool_calls = self._llm.generate_with_tools(
                prompt=f"当前步骤: {step}\n可用工具: {tool_schemas}",
                tools=tool_schemas,
            )

            if self._config.verbose:
                print(f"  思考: {thought}")

            for tool_call in tool_calls:
                name, args = tool_call["name"], tool_call.get("args", {})
                result = self._tools.execute(name, **args)

                if self._config.verbose:
                    status = "成功" if result.success else f"失败: {result.error}"
                    print(f"  工具 {name}: {status}")

                self._memory.add(f"step_{step_idx}_{name}", result.output)

        # 综合输出
        all_results = self._memory.get_all()
        final_prompt = (
            f"目标: {goal}\n"
            f"执行结果: {all_results}\n"
            f"请生成一份完整的回复。"
        )
        return self._llm.generate(final_prompt, system="你是数据分析助手")


def main():
    llm = OpenAILLM(model="gpt-4o-mini")
    memory = DictMemory(max_size=50)
    tools = ToolRegistry()
    planner = SimplePlanner(llm)

    def read_csv(path: str) -> dict:
        import pandas as pd
        df = pd.read_csv(path)
        return {
            "shape": df.shape,
            "columns": list(df.columns),
            "head": df.head(3).to_dict(),
            "dtypes": {col: str(dt) for col, dt in df.dtypes.items()},
        }

    def analyze(df_data: dict) -> dict:
        return {"summary": "数据正常，无异常值", "trends": "Q4 增长 15%"}

    tools.register("read_csv", read_csv, "读取 CSV 文件，返回数据结构")
    tools.register("analyze", analyze, "分析数据，返回摘要和趋势")

    memory.add("user_preference", "喜欢柱状图")

    agent = Agent(llm, memory, tools, planner, config=AgentConfig(verbose=True))
    result = agent.run("分析这份销售数据，告诉我 Q4 的表现")
    print(f"\n=== 最终结果 ===\n{result}")


if __name__ == "__main__":
    main()

几个注意点：

LLM 接口可以随时从 OpenAI 换成 Claude、本地模型，只要实现 generate 和 generatewithtools 两个方法
Memory 现在是字典，换成 Redis、SQLite、向量数据库都行
ToolRegistry 是注册表模式，新增工具只需要 tools.register(name, func, desc) 一行
Planner 现在用 LLM 规划，可以换成硬编码步骤、或者让 LLM 输出 JSON 格式的步骤列表
Agent 主循环核心逻辑不到 30 行——规划、循环执行、综合输出

对话历史记忆

上面的 DictMemory 只管 key-value，但 Agent 真正的记忆是对话历史。

python

class ChatMemory(Memory):
    """对话记忆 —— 管着完整的对话历史"""

    def __init__(self, max_tokens: int = 4000):
        self._messages: list[dict] = []
        self.max_tokens = max_tokens

    def add(self, key: str, value: Any) -> None:
        if isinstance(value, dict):
            self._messages.append(value)
        else:
            self._messages.append({"role": "user", "content": str(value)})

    def get(self, key: str) -> Any | None:
        for msg in reversed(self._messages):
            if key.lower() in msg.get("content", "").lower():
                return msg
        return None

    def get_all(self) -> dict:
        return {"messages": self._messages}

    def clear(self) -> None:
        self._messages.clear()

    def get_messages(self) -> list[dict]:
        return list(self._messages)

    def truncate_if_needed(self) -> list[dict]:
        total_chars = sum(len(msg.get("content", "")) for msg in self._messages)
        if total_chars <= self.max_tokens * 4:
            return self._messages

        trimmed = list(self._messages)
        while trimmed and total_chars > self.max_tokens * 4:
            removed = trimmed.pop(0)
            total_chars -= len(removed.get("content", ""))
        return trimmed

    def __repr__(self):
        return f"ChatMemory({len(self._messages)} messages)"


memory = ChatMemory(max_tokens=4000)
memory.add("", {"role": "user", "content": "我喜欢柱状图"})
memory.add("", {"role": "assistant", "content": "好的，记住了。以后用柱状图展示。"})
memory.add("", {"role": "user", "content": "分析这份数据"})

messages = memory.get_messages()

max_tokens 不是随便设的。你需要知道模型的上下文窗口有多大（GPT-4 是 8K 或 32K，Claude 是 200K），然后留出空间给 system prompt 和输出。设太小会丢失上下文，设太大会浪费 token 且可能让模型注意力分散。至少我目前的做法是设到窗口上限的 70% 左右，留 30% 给 system prompt 和输出。

工具的错误处理

工具执行一定会失败——文件不存在、API 超时、返回格式不对。

python

import time
from functools import wraps


class ToolError(Exception):
    """工具执行异常"""
    pass


def with_retry(max_retries: int = 3, delay: float = 1.0):
    """重试装饰器 —— 工具失败自动重试"""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_retries - 1:
                        print(f"  工具 {func.__name__} 第 {attempt + 1} 次失败: {e}，{delay}s 后重试...")
                        time.sleep(delay)
            raise ToolError(f"{func.__name__} 重试 {max_retries} 次后仍失败: {last_error}")

        return wrapper

    return decorator


@with_retry(max_retries=3, delay=2.0)
def call_api(endpoint: str) -> dict:
    import random
    if random.random() < 0.5:
        raise ConnectionError("Connection refused")
    return {"data": "success"}

工具重试是 Agent 里最实用的基础设施之一。网络调用、外部 API、文件 IO 都靠它兜底。重试次数和间隔需要根据具体场景调——外部 API 通常 3 次、间隔 2 秒就够了；本地文件操作失败通常重试也没用，可以设 1 次。

三层记忆架构

DictMemory 和 ChatMemory 都只解决了短期记忆——当前对话窗口内的信息。但 Agent 真正的记忆需求分三层：

层级	对应	存储	生命周期	检索方式
短期记忆	工作记忆	内存列表	当前会话	顺序/关键词
长期记忆	用户偏好、历史经验	文件/数据库	持久化	key-value 查询
语义记忆	"上次做过类似的事"	向量数据库	持久化	向量相似度

python

"""
三层记忆架构 —— 短期（上下文）+ 长期（持久化）+ 语义（向量检索）
"""

import json
import os
import math
from abc import ABC, abstractmethod
from typing import Any


class ShortTermMemory:
    """短期记忆 —— 当前会话的对话历史"""

    def __init__(self, max_messages: int = 50):
        self._messages: list[dict] = []
        self.max_messages = max_messages

    def add(self, role: str, content: str) -> None:
        self._messages.append({"role": role, "content": content})
        while len(self._messages) > self.max_messages:
            self._messages.pop(0)

    def get_recent(self, n: int = 10) -> list[dict]:
        return self._messages[-n:]

    def summarize(self) -> str:
        return " | ".join(
            f"{m['role']}: {m['content'][:50]}" for m in self._messages[-5:]
        )

    def to_messages(self) -> list[dict]:
        return list(self._messages)


class LongTermMemory:
    """长期记忆 —— 持久化的 key-value 存储"""

    def __init__(self, file_path: str = "memory.json"):
        self.file_path = file_path
        self._store: dict[str, Any] = self._load()

    def _load(self) -> dict:
        if os.path.exists(self.file_path):
            with open(self.file_path, "r", encoding="utf-8") as f:
                return json.load(f)
        return {}

    def _save(self) -> None:
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(self._store, f, ensure_ascii=False, indent=2)

    def set(self, key: str, value: Any) -> None:
        self._store[key] = value
        self._save()

    def get(self, key: str, default=None) -> Any:
        return self._store.get(key, default)

    def delete(self, key: str) -> None:
        self._store.pop(key, None)
        self._save()

    def get_facts(self) -> str:
        if not self._store:
            return "（无已知事实）"
        return "\n".join(f"- {k}: {v}" for k, v in self._store.items())


class SemanticMemory:
    """
    语义记忆 —— 基于向量相似度的检索

    这里用简化的 TF-IDF 风格实现（实际应该用 embedding 模型）。
    """

    def __init__(self, top_k: int = 3):
        self._entries: list[dict] = []
        self.top_k = top_k

    def store(self, text: str, metadata: dict = None) -> None:
        self._entries.append({"text": text, "metadata": metadata or {}})

    def retrieve(self, query: str) -> list[dict]:
        if not self._entries:
            return []

        query_tokens = set(self._tokenize(query))

        scored = []
        for entry in self._entries:
            entry_tokens = set(self._tokenize(entry["text"]))
            overlap = len(query_tokens & entry_tokens)
            union = len(query_tokens | entry_tokens)
            score = overlap / union if union > 0 else 0
            scored.append((score, entry))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [
            {"text": entry["text"], "metadata": entry["metadata"], "score": score}
            for score, entry in scored[: self.top_k]
        ]

    def _tokenize(self, text: str) -> list[str]:
        return text.lower().split()


class CompositeMemory:
    """组合记忆 —— 统一管理三层记忆"""

    def __init__(self):
        self.short_term = ShortTermMemory(max_messages=50)
        self.long_term = LongTermMemory(file_path="memory.json")
        self.semantic = SemanticMemory(top_k=3)

    def add_message(self, role: str, content: str, important: bool = False) -> None:
        self.short_term.add(role, content)
        if important:
            self.long_term.set(f"msg_{len(self.short_term._messages)}", content)
            self.semantic.store(content, {"role": role, "timestamp": "now"})

    def build_context(self, query: str = "") -> str:
        parts = []

        recent = self.short_term.get_recent(5)
        if recent:
            parts.append("### 最近对话")
            for msg in recent:
                parts.append(f"{msg['role']}: {msg['content']}")

        facts = self.long_term.get_facts()
        if facts:
            parts.append("### 已知事实")
            parts.append(facts)

        if query:
            related = self.semantic.retrieve(query)
            if related:
                parts.append("### 相关历史")
                for r in related:
                    parts.append(f"[相关度 {r['score']:.2f}] {r['text']}")

        return "\n\n".join(parts)


memory = CompositeMemory()

memory.add_message("user", "我喜欢用柱状图展示数据")
memory.add_message("assistant", "好的，已记住你的偏好")

memory.add_message("user", "我的数据源是 sales_2025.csv", important=True)

context = memory.build_context(query="数据分析")
print(context)

三层记忆不是"存越多越好"，每层有不同的访问频率和检索策略。短期记忆是顺序访问（最近的最重要），长期记忆是 key-value 查找（精确匹配），语义记忆是相似度检索（模糊匹配）。把它们混在一个类里，检索效率会急剧下降。实际项目中，我见过很多人把所有东西塞进一个向量库，结果每次查询都要跑 embedding，慢不说，精确查询反而不如直接查字典。

从函数签名自动生成 Tool Schema

前面工具 schema 是手动写的 description。实际项目中工具可能有几十个，手动维护容易出错。下面的代码展示如何从 Python 函数签名自动生成 JSON Schema：

python

"""
从函数签名自动生成 Tool Schema
"""

import inspect
import json
from typing import get_type_hints


def generate_tool_schema(func) -> dict:
    """
    从 Python 函数自动生成工具 schema

    提取：函数名、docstring（作为描述）、参数名、参数类型、默认值、必填项
    """
    sig = inspect.signature(func)
    hints = get_type_hints(func)

    properties = {}
    required = []

    for name, param in sig.parameters.items():
        if name == "self":
            continue

        param_type = hints.get(name, Any)
        type_mapping = {
            str: "string",
            int: "integer",
            float: "number",
            bool: "boolean",
            list: "array",
            dict: "object",
        }
        json_type = type_mapping.get(param_type, "string")

        prop = {"type": json_type}

        docstring = inspect.getdoc(func) or ""
        for line in docstring.split("\n"):
            line = line.strip()
            if line.startswith(f":param {name}:"):
                prop["description"] = line.split(":", 2)[-1].strip()

        if param.default is inspect.Parameter.empty:
            required.append(name)
        else:
            prop["default"] = param.default

        properties[name] = prop

    func_desc = (inspect.getdoc(func) or "").split("\n")[0].strip()

    return {
        "name": func.__name__,
        "description": func_desc,
        "parameters": {
            "type": "object",
            "properties": properties,
            "required": required,
        },
    }


def read_csv(path: str, delimiter: str = ",", encoding: str = "utf-8") -> dict:
    """
    读取 CSV 文件并返回数据结构

    :param path: CSV 文件路径
    :param delimiter: 分隔符，默认逗号
    :param encoding: 文件编码，默认 utf-8
    """
    return {"path": path, "rows": 100, "columns": 5}


def analyze_data(data: dict, metrics: list = None) -> dict:
    """
    分析数据并返回统计摘要

    :param data: 待分析的数据字典
    :param metrics: 要计算的指标列表
    """
    return {"mean": 0.5, "std": 0.1}


schema1 = generate_tool_schema(read_csv)
schema2 = generate_tool_schema(analyze_data)

print(json.dumps(schema1, indent=2, ensure_ascii=False))
# 输出：
# {
#   "name": "read_csv",
#   "description": "读取 CSV 文件并返回数据结构",
#   "parameters": {
#     "type": "object",
#     "properties": {
#       "path": {"type": "string", "description": "CSV 文件路径"},
#       "delimiter": {"type": "string", "description": "分隔符，默认逗号", "default": ","},
#       "encoding": {"type": "string", "description": "文件编码，默认 utf-8", "default": "utf-8"}
#     },
#     "required": ["path"]
#   }
# }

这种做法消除了手动维护 schema 和实际函数签名不一致的问题。改了函数参数，schema 自动更新。工具数量超过 10 个的时候，手动维护的 schema 一定会和实际函数不同步——我踩过这个坑。

上下文窗口管理

Agent 运行过程中，上下文（对话历史 + 工具结果）不断增长，最终会触及模型的 token 限制。三种常见策略：

python

"""
上下文窗口管理的三种策略

ReAct 循环每步追加 Thought + Action + Observation，
10 步之后上下文可能超过 4000 token（GPT-3.5 的限制）。
"""

from typing import Callable


class ContextWindowManager:
    """上下文管理器 —— 监控和控制 token 使用量"""

    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self._current_tokens: int = 0

    def estimate_tokens(self, text: str) -> int:
        english_chars = sum(1 for c in text if c.isascii())
        chinese_chars = len(text) - english_chars
        return int(english_chars / 4 + chinese_chars / 1.5)

    def add(self, text: str) -> bool:
        token_count = self.estimate_tokens(text)
        if self._current_tokens + token_count > self.max_tokens:
            return False
        self._current_tokens += token_count
        return True

    @property
    def usage_ratio(self) -> float:
        return self._current_tokens / self.max_tokens


class SlidingWindowStrategy:
    """
    滑动窗口 —— 只保留最近的 N 条消息

    适用场景：早期步骤的信息不再重要。
    优点：简单、token 使用量恒定。
    缺点：可能丢掉重要的早期信息。
    """

    def __init__(self, keep_recent: int = 10):
        self.keep_recent = keep_recent

    def compress(self, messages: list[dict], system_prompt: str = "") -> str:
        recent = messages[-self.keep_recent:]
        parts = [system_prompt] if system_prompt else []
        for msg in recent:
            parts.append(f"{msg['role']}: {msg['content']}")
        return "\n".join(parts)


class SummaryCompressionStrategy:
    """
    摘要压缩 —— 用 LLM 把早期步骤压缩成一段摘要

    适用场景：早期步骤有重要信息，但不需要保留完整细节。
    优点：保留关键信息、大幅减少 token。
    缺点：需要额外一次 LLM 调用；摘要可能丢失细节。
    """

    def __init__(self, llm, compress_trigger: float = 0.7):
        self.llm = llm
        self.compress_trigger = compress_trigger

    def should_compress(self, window: ContextWindowManager) -> bool:
        return window.usage_ratio > self.compress_trigger

    def compress(self, messages: list[dict], system_prompt: str = "") -> str:
        early_messages = messages[:-5]
        early_text = "\n".join(f"{m['role']}: {m['content']}" for m in early_messages)

        prompt = f"""
请将以下对话历史压缩为一段简要摘要。只保留关键信息：目标、工具调用结果、重要发现。
不要保留详细的 Thought 过程。

{early_text}
"""
        summary = self.llm.generate(prompt, system="你是对话摘要专家。")

        recent = messages[-5:]
        parts = [system_prompt, f"[早期对话摘要]\n{summary}"]
        for msg in recent:
            parts.append(f"{msg['role']}: {msg['content']}")
        return "\n".join(parts)


class KeyInfoExtractionStrategy:
    """
    关键信息提取 —— 每步执行后，只保留关键信息到上下文

    适用场景：工具输出很长（比如整个 CSV 内容），但只有少量信息有用。
    优点：从源头控制 token 增长，不需要事后压缩。
    缺点：需要为每个工具定义"什么信息是关键"的逻辑。
    """

    def __init__(self):
        self._extractors: dict[str, Callable] = {}

    def register_extractor(self, tool_name: str, extractor: Callable) -> None:
        self._extractors[tool_name] = extractor

    def process_observation(self, tool_name: str, raw_output: Any) -> str:
        if tool_name in self._extractors:
            return self._extractors[tool_name](raw_output)
        return str(raw_output)[:500]


extractor = KeyInfoExtractionStrategy()

def extract_csv_info(data: dict) -> str:
    return f"文件已读取: {data['rows']} 行 × {data['columns']} 列, 字段: {', '.join(data['columns'])}"

extractor.register_extractor("read_csv", extract_csv_info)

raw_output = {"rows": 10000, "columns": ["date", "revenue", "region"], "head": {...}}
observation = extractor.process_observation("read_csv", raw_output)
print(observation)
# 输出: "文件已读取: 10000 行 × 3 列, 字段: date, revenue, region"

根据上面的代码，三种策略各有取舍：滑动窗口 token 用量固定但会丢失早期信息，适合流水线任务；摘要压缩保留关键信息但多一次 LLM 调用成本，适合长对话；关键信息提取从源头控制 token 增长，但需要为每个工具写提取逻辑，适合工具输出很长的场景。实际项目中我通常会混合使用——滑动窗口兜底，关键信息提取控制工具输出，摘要压缩只在极端情况下触发。

一些实际问题

Agent 架构真的有必要吗？什么时候直接调 LLM 就够了？

如果你的任务就是问答、翻译、总结、改写，这些是 LLM 的本职工作，加 Agent 架构只会增加复杂度。不需要外部数据、不需要调用 API、不需要读写文件的时候，纯文本处理不需要工具。任务只有一两步（比如"把这段英文翻译成中文，然后总结要点"），也不需要规划。

反过来，任务需要多步操作且中间步骤依赖外部工具或数据、需要记住之前的对话或用户偏好、同一个任务不同用户的执行路径不同、需要追踪每一步的执行结果——这些场景才需要 Agent 架构。

四大组件里，哪个最容易出问题？我个人建议重点关注记忆和工具这两个。记忆管理是最容易被低估的组件——记太多上下文爆炸，记太少 Agent 变健忘。更深层的问题是遗忘策略：删哪条记忆？FIFO 太粗暴，LRU 好一点，但最好的策略是按"重要性"标记，只删不重要的。工具失败是 Agent 最常见的问题，API 超时、文件格式不对、参数不匹配，这些都不关 LLM 的事。更深层的问题是工具描述的准确性：如果 tool schema 的 description 写得不清楚，LLM 会选错工具或者传错参数。

Planner 是最容易被过度设计的组件。简单的任务不需要复杂的规划器，很多时候一个固定的步骤列表就够了。但复杂任务没有好的规划器，Agent 就会乱来。更深层的问题是规划与执行的 gap：LLM 规划的步骤工具可能不支持，工具支持的规划可能漏掉。

Agent 架构引入的代价也很实在：一次调用等于多次 LLM 调用加多次工具调用，响应时间从秒级变成十秒级甚至分钟级；出了问题要排查是 LLM 理解错了、工具执行错了、记忆漏了还是规划乱了；每次规划、每次工具调用、每次记忆检索都可能产生 LLM 调用费用；LLM 的行为不可预测，同样的输入这次调了正确的工具，下次可能漏调。

以下情况大概率是过度设计：任务就是"用户问一个问题 → LLM 回答"但搞了一套 Agent 框架；只有 2 个工具且调用顺序固定但搞了一个 Planner；对话很短没有跨轮次的上下文需求但搞了一个向量检索的记忆系统；Agent 代码量（框架 + 组件 + 工具注册）比业务逻辑多了 5 倍以上。最经典的信号是：花了一个月搭建 Agent 框架，但实际跑的任务用一个 prompt 就能解决。

更简单的替代方案其实很多。只需要工具调用的话，OpenAI Function Calling / Claude Tool Use 就够了，SDK 原生支持；只需要记忆，把 messages 列表传回去就行；只需要固定流程，写死步骤顺序比 LLM 规划简单得多。Spring 或 Python 项目里，Service + Strategy 模式通常比 Agent 架构更轻量。

python

# 最简替代方案：不需要完整框架，只需要工具调用
import openai

client = openai.OpenAI(api_key="sk-...")

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "获取某个城市的天气",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
            },
            "required": ["city"],
        },
    },
}]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
    tools=tools,
)

if response.choices[0].message.tool_calls:
    for tool_call in response.choices[0].message.tool_calls:
        args = json.loads(tool_call.function.arguments)
        result = get_weather(**args)

注意这里用 json.loads 而不是 eval。eval 会执行任意代码，如果 LLM 返回的参数被恶意注入，后果严重。这个方案的好处是不需要自建 Agent 框架，OpenAI/Anthropic 的 SDK 已经帮你处理了"模型决定调工具 → 你执行 → 结果返回模型"的循环。

与其他设计模式的关系

Agent 架构和经典设计模式有不少重叠。ToolRegistry 本质上就是简单工厂加策略模式——每个工具是一个策略，注册表是工厂。Planner 可以在运行时切换不同的规划策略（LLM 规划、模板规划、用户指定），这是策略模式的典型应用。Agent 的主循环（规划 → 执行 → 输出）是一个模板方法，不同 Agent 的区别在于每一步的具体实现。对话过程可以用状态模式管理（初始 → 规划中 → 执行中 → 完成 → 失败），每一步执行都可以触发观察者回调用于日志、监控、用户通知。

框架中的实际应用

框架	Agent 架构	说明
LangChain	`AgentExecutor` = LLM + Tools + Memory	最经典的 Agent 框架，四大组件齐全。但代码量大，学习曲线陡
OpenAI Assistants API	内置 Agent 能力	SDK 层面封装好了工具调用和对话管理，不需要自己搭框架
CrewAI	多 Agent 协作	在单 Agent 之上加了"团队"概念，多个 Agent 各司其职
Anthropic Tool Use	Claude 原生工具调用	不需要第三方框架，Anthropic SDK 直接支持工具调用
LlamaIndex	Agent + 检索	核心能力是检索增强，但内置了 Agent 执行器

从"一个 prompt"到 Agent 架构

很多教程直接给完整的 Agent 代码，但不说怎么从简单代码一步步演进。笔者自己走过的路径是这样的：

最开始一个 prompt 搞定——response = llm.generate("分析这份销售数据，告诉我 Q4 表现")。然后发现读文件这件事 LLM 做不到，需要工具，于是手动调 pd.read_csv("sales.csv") 把结果塞给 LLM。工具从 1 个变成 10 个之后，if-else 判断用哪个工具已经没法看了，于是抽象出 ToolRegistry 来管理。再然后用户说"我喜欢柱状图"，下一轮问"换种图表"的时候 Agent 应该知道换成柱状图——这就必须加记忆了。任务变成"分析数据并生成报告"这种需要多步的事情，一步到位跑不通，于是加了 Planner 来拆解步骤。最后把这些组件组装到一起，就是 Agent 循环了。

这个过程中一个重要的体会是：不需要一开始就设计 Agent 架构。大多数好的架构是从需求里长出来的。先写最简单的代码，发现 LLM 不够用就加工具，发现记不住就加记忆，发现乱来就加规划。一步一步加，每一步都有明确的需求驱动。至少我目前的情况是这样。

使用判断标准

该用 Agent 架构的场景（至少命中两条）：

任务需要调用外部工具（API、数据库、文件、代码执行）
任务需要多轮对话，且跨轮次有上下文依赖
任务需要多步操作，且步骤之间有依赖关系
需要追踪每一步的执行结果

不该用的场景（命中任意一条就不需要）：

任务就是纯文本处理（问答、翻译、总结）
不需要外部数据或工具
任务只有一两步，步骤固定
对响应延迟要求高（< 3 秒）

写在最后

Agent 不是一个新技术，它是设计模式在 LLM 时代的应用。LLM 是大脑，记忆是数据库，工具是 API 接口，规划是策略选择。理解了这一点，就不会被各种 Agent 框架牵着走——自己就能判断哪个框架值得用、哪个不值得。

下一篇会深入 Agent 架构中最核心的执行模式——ReAct（思考-行动循环）。这是几乎所有 Agent 框架的基础执行机制，也是最容易被滥用的模式。