Agent定义与架构——从Anthropic"Building Effective Agents"出发

日期: 2026-09-27 方向: AI系统工程 / Agent 阶段: Phase 3 - Agent架构与多Agent (Day 149-162) 标签: #Agent #Anthropic #Workflow #ReAct #BuildingEffectiveAgents

---

今日目标

为什么从 Day 149 这里讲清楚定义：用户接下来 14 天会反复使用 ReAct / Plan-Execute / LangGraph / CrewAI / AutoGen 等关键词。如果"agent"这个词被当成营销标签来用，后续讨论会全部失焦。今天必须把"什么不是 agent"也定义出来。

---

一、核心概念——严格定义

1.1 Anthropic 的定义（2024-12 原文）

"Workflows are systems where LLMs and tools are orchestrated through predefined code paths. Agents, on the other hand, are systems where LLMs dynamically direct their own processes and tool usage, maintaining control over how they accomplish tasks."

把它翻译成判定标准：

判定题：

• "用户问候 → LLM 分类 → 调相应 tool → 返回" → Workflow（Routing）
• "用户给一个研究任务 → LLM 自己反复搜索/读文档/汇总，直到自认为答完" → Agent
• "ChatGPT 网页版" → Chatbot（人在 loop 里，每轮都等用户输入）

1.2 Anthropic 的 5+1 种 Pattern

Pattern 1: Prompt chaining（链式）

顺序调用 LLM，前一个的输出作为后一个的输入。不是 agent，是 workflow。

Input ─► LLM₁ ─► Gate? ─► LLM₂ ─► LLM₃ ─► Output

金融场景：研报草稿 → 风险审查 → 合规改写 → 终稿。

Pattern 2: Routing（路由）

分类器决定走哪条分支。不是 agent。

                    ┌─► Path A (refund tool)
Input ─► Classifier ─┼─► Path B (search KB)
                    └─► Path C (escalate)

金融场景：客户问询分类（账单 / 交易 / 投诉 / 监管）。

Pattern 3: Parallelization（并行）

• Sectioning：把任务拆成独立子任务并行
• Voting：N 个 LLM 同任务投票

              ┌─► LLM (legal check) ─┐
Input ────────┼─► LLM (cost check)  ─┼─► Aggregator ─► Output
              └─► LLM (risk check)  ─┘

金融场景：合规审查 = 反洗钱 / 制裁名单 / KYC 三路并行。

Pattern 4: Orchestrator-workers（编排者-工人）

中央 LLM 动态拆分子任务，分发给 worker LLM。已经偏 agent，但 orchestrator 通常不循环。

        Orchestrator
        / | \
       v  v  v
     W1  W2  W3   ─► Synthesizer ─► Output

金融场景：投资委员会 — orchestrator 决定分析师角色，每个角色独立产出，最后汇总。

Pattern 5: Evaluator-optimizer（评估-优化）

一个 LLM 生成、另一个评估，循环到达标。

Generator ─► Evaluator ─► (pass?) ──┐
    ▲                               │
    └────── feedback ───────────────┘

金融场景：研报写作 — 写作 LLM + 风格/事实评分 LLM。

Pattern 6: Agents（自主）

LLM 自己决定调什么 tool、何时停。真正的 agent。

        ┌─────────────────┐
Input ─►│ LLM             │
        │  ▲           │  │
        │  │ observation│  │
        │  │           ▼  │
        │  ┌──────────┐  │
        │  │ Tools    │  │
        │  └──────────┘  │
        └─────────────────┘
              │
              ▼ (when done)
            Output

1.3 Anthropic 的核心建议（重点摘录）

1. "Find the simplest solution possible, and only increase complexity when needed."
2. 大多数生产用例 workflow 就够了，不要无脑上 agent
3. Agent 的 3 个必要条件：
   • 任务开放性（不能预先列步骤）
   • 可用 tool 集合明确
   • 失败成本可控或可被 human-in-the-loop 兜底
4. 框架（LangGraph/CrewAI/AutoGen）"会让简单事情更简单，但也会创造黑盒"——要先理解原理再用框架

---

二、架构图——通用 Agent Loop

┌────────────────────────────────────────────────────────────────┐
│                    Agent Loop (Anthropic style)                │
│                                                                │
│   ┌─────────┐   prompt + history + tools     ┌─────────────┐   │
│   │ Memory  │────────────────────────────────► │   LLM       │   │
│   │ (short  │                                  │ (claude     │   │
│   │  +long) │◄─── append observation ──────────│  -opus-4-7) │   │
│   └─────────┘                                  └─────────────┘   │
│        ▲                                            │              │
│        │ persist                                    │ tool_use     │
│        │                                            ▼              │
│   ┌────────────┐    execute    ┌─────────────────────────┐        │
│   │ State      │◄──────────────│  Tool Executor          │        │
│   │ (current   │               │  (search/code/api/...)  │        │
│   │  task)     │   tool_result │                         │        │
│   └────────────┘──────────────►└─────────────────────────┘        │
│                                                                    │
│   终止条件: LLM 输出 stop_reason="end_turn"  或  达到 max_iters    │
└────────────────────────────────────────────────────────────────────┘

---

三、代码——BaseAgent 接口骨架

为后续 13 天复用，先定义统一接口。

# base_agent.py
"""
Day 149 - BaseAgent interface.

This is the contract every agent in Day 150-162 must satisfy. We deliberately
do NOT depend on LangGraph / CrewAI here — those will subclass / adapt.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
import time
import uuid

# ---------- Data classes ----------
@dataclass
class Message:
    role: str               # "user" | "assistant" | "tool"
    content: Any            # str or list of blocks (Anthropic-style)
    timestamp: float = field(default_factory=time.time)

@dataclass
class ToolCall:
    id: str
    name: str
    input: dict[str, Any]

@dataclass
class ToolResult:
    tool_call_id: str
    content: Any
    is_error: bool = False

@dataclass
class AgentTrace:
    """One full agent run, for logging / eval."""
    run_id: str
    iterations: int = 0
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_results: list[ToolResult] = field(default_factory=list)
    messages: list[Message] = field(default_factory=list)
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_cost_usd: float = 0.0
    stop_reason: Optional[str] = None
    elapsed_sec: float = 0.0

# ---------- Tool abstraction ----------
@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict          # JSON Schema
    handler: Callable[[dict], Any]

    def to_anthropic(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.input_schema,
        }

# ---------- BaseAgent ----------
class BaseAgent(ABC):
    """Every agent needs: name, system prompt, tools, run() method."""

    def __init__(
        self,
        name: str,
        system_prompt: str,
        tools: list[Tool],
        model: str = "claude-opus-4-7",
        max_iterations: int = 20,
    ):
        self.name = name
        self.system_prompt = system_prompt
        self.tools = {t.name: t for t in tools}
        self.model = model
        self.max_iterations = max_iterations

    @abstractmethod
    def run(self, user_input: str) -> AgentTrace:
        """Execute the agent loop until stop or max_iterations."""
        ...

    # ------ default tool dispatcher ------
    def _execute_tool(self, call: ToolCall) -> ToolResult:
        if call.name not in self.tools:
            return ToolResult(
                tool_call_id=call.id,
                content=f"Tool '{call.name}' not found. Available: {list(self.tools)}",
                is_error=True,
            )
        try:
            result = self.tools[call.name].handler(call.input)
            return ToolResult(tool_call_id=call.id, content=str(result))
        except Exception as e:
            return ToolResult(
                tool_call_id=call.id,
                content=f"{type(e).__name__}: {e}",
                is_error=True,
            )

    @staticmethod
    def new_run_id() -> str:
        return f"run_{uuid.uuid4().hex[:8]}"

# ---------- A toy "agent" that doesn't actually loop, for unit tests ----------
class EchoAgent(BaseAgent):
    """Returns the user input verbatim. Sanity-check the interface."""
    def run(self, user_input: str) -> AgentTrace:
        t = AgentTrace(run_id=self.new_run_id())
        t.messages.append(Message(role="user", content=user_input))
        t.messages.append(Message(role="assistant", content=user_input))
        t.stop_reason = "end_turn"
        t.iterations = 1
        return t

if __name__ == "__main__":
    agent = EchoAgent(name="echo", system_prompt="", tools=[])
    trace = agent.run("hello agent")
    assert trace.messages[-1].content == "hello agent"
    print("OK base_agent contract works.")

关键设计：把 AgentTrace 作为一等公民。后面 Day 160 的 trajectory eval 会直接吃这个 trace。

---

四、金融领域应用——为什么金融业最早大规模上 agent

反例（应该用 workflow）：

• 日终 NAV 计算（步骤固定）
• 交易订单路由（速度敏感、路径明确）
• 报表生成（结构化）

---

五、Web3 集成——Agent 上链的特殊性

链上 agent 与传统 SaaS agent 的差异（贯穿后续 14 天的设计原则）：

关键设计原则（Day 161 会落地）：

1. Session keys（ERC-7715）限制 agent 权限：只能调用某些合约 / 某些函数 / 单笔金额上限 / 时间窗口
2. 支付协议（x402）让 agent 在调用付费 API 时自动用稳定币结算
3. Simulation first：链上交易前必须 eth_call 模拟
4. Human-in-the-loop：超过阈值（例如 $1k）必须人工签

---

六、生产经验与陷阱

1. "我们做了一个 agent"——其实是一个 workflow 团队为了赶 demo 把 if/else 包进 LLM prompt，自称 agent。结果根本没 loop，没 tool calling，是 prompt-chaining。先承认它是 workflow，再决定要不要升级为 agent。

2. Tool 调用无限循环 LLM 反复调同一个 tool（参数微变），不收敛。原因：tool 返回的错误信息不够具体，或 system prompt 没说"如果连续 3 次失败就返回错误"。Day 152 会专门讲。

3. Context 爆炸 每轮 tool result 都堆进 message history，10 轮后超 200k token。需要：

   • 长 result 写到 file/state，message 里只放摘要
   • 周期性 summarize 历史
   • 切到长上下文模型（claude-sonnet-4-6 1M context）

4. Cost 失控 开发时用 opus，生产忘记切 haiku。Agent loop 一次任务 50 次 LLM call × $0.05 = $2.5/任务 × 1000 用户 = $2500/天。永远在 trace 里记 token + cost。

5. Agent 框架黑盒 LangGraph 的 add_conditional_edges 出错时栈跟踪深 30 层，调试痛苦。建议：先用裸 SDK 写出 v0，再迁框架。

---

七、Cost & Latency

省钱杠杆：

• Prompt caching 可以省 90% 的 system + tool schema 成本（重复输入只算 0.1×）
• 把 reasoning 用 sonnet/haiku，最终 synthesis 用 opus（model routing）
• Tool schema 写紧凑（每个 tool 描述 < 100 token）

---

八、关键速查

5 种 pattern 决策树

任务能预先列出步骤？
├─ 能 → workflow
│    ├─ 顺序 → prompt chaining
│    ├─ 分类 → routing
│    ├─ 独立子任务 → parallelization
│    ├─ 子任务由 LLM 决定 → orchestrator-workers
│    └─ 需要"打磨" → evaluator-optimizer
└─ 不能 + 工具明确 + 失败可控 → agent

Anthropic stop_reason 枚举

---

九、面试题

Q1: 什么时候用 agent，什么时候用 workflow？

A: 任务能否预先列出步骤是分水岭。能列就 workflow（成本低、可控、可监控）。不能列且失败成本可控、tool 集合明确，才上 agent。Anthropic 原话："find the simplest solution possible"。生产中绝大多数 LLM app 是 workflow。

Q2: 如何判断一个团队的"AI agent 产品"是真 agent 还是营销话术？

A: 三个问题：① 同一个用户输入，每次 LLM 调用次数是否一样？（一样 → 不是 agent）② Tool 调用顺序是否在代码里写死？（写死 → 不是）③ 终止条件是 LLM 决定还是代码决定？（代码决定 → 不是）。

Q3: Anthropic 的 5 种 pattern 中，哪一种最像 agent，哪一种最不像？

A: 最像：orchestrator-workers + 真·agent。最不像：prompt chaining（线性、无 loop、无 tool 自主决策）。Routing/parallelization 也是 workflow。Evaluator-optimizer 介于中间，因为有反馈 loop，但路径仍由代码定义。

Q4: 给一个金融 use case，让你判断 agent vs workflow。

A: 例如"日终监管报送" → workflow（步骤明确：拉数据 → 校验 → 生成报文 → 送监管）。例如"客户来电询问'我去年的税务报告异常'" → agent（要查交易、对账、调税务规则、可能需要 escalate，路径开放）。

Q5: 为什么 Anthropic 强调"先用裸 SDK 再用框架"？

A: 框架（LangGraph 等）封装了 state machine、retry、checkpoint，但出错时栈深、调试难。裸 SDK 实现一个 ReAct loop 只需 50 行（Day 150 会写）。先理解原语，遇到框架黑盒时才能定位问题。

---

十、深入：Anthropic 5+1 pattern 的代码骨架对照

后续 Day 150-154 会逐一实现。先把每种 pattern 的"最小可读骨架"列出来，方便 mental model：

10.1 Prompt chaining（workflow）

def chain(input_text):
    a = llm.create(prompt=f"Step1: extract {input_text}")
    if "INVALID" in a: return None      # gate
    b = llm.create(prompt=f"Step2: refine {a}")
    return b

10.2 Routing（workflow）

def route(query):
    label = llm.classify(query, labels=["billing","support","sales"])
    return HANDLERS[label](query)

10.3 Parallelization (sectioning)

async def parallel(input_text):
    a, b, c = await asyncio.gather(
        llm.create(prompt=f"legal: {input_text}"),
        llm.create(prompt=f"cost: {input_text}"),
        llm.create(prompt=f"risk: {input_text}"),
    )
    return aggregator(a, b, c)

10.4 Orchestrator-workers

def orchestrate(task):
    plan = orchestrator_llm.plan(task)
    results = [worker_llm.execute(s) for s in plan.steps]
    return synthesizer_llm.combine(results)

10.5 Evaluator-optimizer

def gen_eval(task, max_iters=3):
    out = generator_llm.generate(task)
    for _ in range(max_iters):
        feedback = evaluator_llm.evaluate(out)
        if feedback.passed: return out
        out = generator_llm.regenerate(task, feedback)
    return out  # gave up

10.6 Agent (true loop)

def agent(task):
    msgs = [{"role":"user","content":task}]
    for _ in range(MAX_ITERS):
        resp = llm.create(messages=msgs, tools=TOOLS)
        msgs.append({"role":"assistant","content":resp.content})
        if resp.stop_reason == "end_turn": return resp.text
        if resp.stop_reason == "tool_use":
            results = [run(tc) for tc in resp.tool_uses]
            msgs.append({"role":"user","content":results})
    raise TimeoutError

把这 6 个骨架默写一遍，再来 Day 150 写完整版会很顺。

---

十一、生产案例：Anthropic 自己的 agents

Anthropic 在 2024-2025 公开过的几个 agent 形态：

关键模式：Anthropic 自己只在确实需要的产品上才上 agent。Claude Code 必须 agent（步骤无法预先列），Artifacts 是 workflow（步骤明确）。这是验证"先 workflow 后 agent"原则的内部一致性。

---

十二、Phase 3 阶段全图（Day 121-220 概览）

本周（149-162）是 Phase 3 中段最关键的 14 天，因为后面所有内容都建立在"我会做 agent"的基础上。

---

明日预告

Day 150: ReAct 模式 — 从零实现裸 ReAct（不用框架）

• Yao 等人 2022 ReAct 论文核心：Thought → Action → Observation 循环
• 用 Anthropic SDK + tool_use 手写完整 loop
• 跑一个金融数据查询任务，trace 每一步 thought