万字长文：从0开发大模型之17种Agent架构演进

在大型语言模型（LLM）应用开发领域，Agent架构的设计正逐渐成为决定系统落地成败的关键因素。许多开发者误以为Agent的核心在于Prompt Engineering或特定框架的DSL（领域特定语言），但深入分析表明，Agent架构的本质实际上是控制流设计。一个健壮的Agent系统必须具备明确的状态建模、显式的控制流表达、局部错误截断能力以及副作用管理机制。

本文基于对all-agentic-architectures项目的深度剖析，结合agno框架的简洁实现，系统性地拆解了17种主流Agent模式的演进路径。我们将不再局限于代码实现的罗列，而是聚焦于系统升级背后的逻辑：上一代架构为何失效？新一代架构增加了何种控制能力？复杂度如何在何处失控？通过从单次生成到反思闭环，从工具交互到显式规划，再到多智能体编排与长期记忆系统的完整演化链条，本文为开发者提供了一套统一的分析框架，帮助理解如何构建可控、可解释且可恢复的智能代理系统。

统一分析框架：解构Agent架构的六个维度

为了系统化地评估每一种Agent架构，我们建立了一套统一的分析模型。这套模型不关注模型本身的“智商”，而是关注系统工程层面的稳健性。在后续对每种架构的探讨中，我们将始终围绕以下六个核心问题展开：

1. 它要解决什么问题？ 明确当前架构旨在弥补上一代架构的哪些缺陷，例如是解决知识边界问题，还是解决执行稳定性问题。
2. 它的State（状态）是什么？ 分析系统中新增了哪些关键状态字段，以及这些字段为何必须存在以支撑控制流。
3. 它的拓扑结构是什么？ 描述数据和控制流的走向，是线性链、循环回路、分叉汇聚、共享黑板，还是树状搜索结构。
4. 它的Router（路由）如何工作？ 探讨决策机制是基于固定边、条件判断、动态调度，还是包含验证回路及人工审批环节。
5. 它的失败模式是什么？ 识别架构中最脆弱的环节，例如是在状态同步、工具调用还是逻辑推理阶段容易出错。
6. 什么时候该升级到下一种？ 界定当前模式的能力边界，明确在何种业务场景下需要引入更复杂的架构。

基于Agno框架的Agent抽象表达

agno框架以其简洁的抽象能力，能够清晰地表达绝大多数Agent架构的核心逻辑。通过几个基本组件，我们可以构建出从简单到复杂的各类系统。以下代码展示了agno中最核心的抽象概念：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow.v2 import Workflow, Step, Router, Loop

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[...],
    instructions="...",
    response_model=SomePydanticModel,  # 结构化输出
)

wf = Workflow(
    name="my_flow",
    steps=[
        Step(name="plan", agent=planner_agent),
        Loop(
            name="execute_and_verify",
            steps=[executor_step, verifier_step],
            end_condition=lambda outputs: outputs[-1].content.is_done,
        ),
        Step(name="synthesize", agent=synthesizer_agent),
    ],
)
wf.run(message="...")

这段代码背后蕴含了Agent系统的最小数学结构：

• response_model定义了系统的状态空间，确保输出符合预期结构。
• Agent、tools和Step定义了状态变换函数，即如何从当前状态推导至下一状态。
• steps列表定义了确定性转移，即固定的执行顺序。
• Router和Condition定义了条件转移，允许系统根据中间结果动态调整路径。
• Loop.end_condition定义了终止条件，防止无限循环。
• wf.run()定义了可执行系统入口。

在后续的架构分析中，我们将重点描述每个架构新增的状态字段、Agent角色、路由逻辑及验证机制，而非单纯强调其“智能”程度。

Agent架构演进总览：逐步添加控制能力

Agent架构的演化并非盲目追求AGI（通用人工智能），而是为了解决一个经典工程问题：怎样让系统在更复杂的环境里，依然保持可控、可解释、可恢复。 以下是各阶段架构的核心能力演进概览：

Reflection：构建最小质量闭环

它要解决什么问题？

单次LLM生成的质量往往不稳定，存在幻觉或逻辑漏洞。Reflection（反思）架构的价值不在于让模型具备真正的“意识”，而在于它首次将生成过程拆解为两个不同职能的Pass：生成和评估。这是最小的质量闭环，通过将单步生成改为“生成→评估→修正”的三阶段控制流，显著提升了输出结果的准确性。

它的State是什么？

Reflection架构的核心状态包含三个字段：draft（初稿）、critique（评审意见）和refined_code（修正后的代码）。在agno中，我们利用Pydantic模型将这些中间思考结果显式定义为结构化输出，从而低成本地实现状态管理：

from pydantic import BaseModel, Field
from typing import List

class DraftCode(BaseModel):
    code: str = Field(description="Python code to solve the user's request.")
    explanation: str = Field(description="A brief explanation of how the code works.")

class Critique(BaseModel):
    has_errors: bool
    is_efficient: bool
    suggested_improvements: List[str]
    critique_summary: str

class RefinedCode(BaseModel):
    refined_code: str
    refinement_summary: str

这种设计使得系统不再将中间思考隐藏在非结构化的上下文中，而是将其作为显式的状态字段进行管理。

它的拓扑是什么？

Reflection的拓扑结构是纯线性的，不存在分支或循环。其执行流程如下：

1. Generator：负责生成初始代码草案。
2. Critic：负责审查草案，指出错误和改进建议。
3. Refiner：根据审查意见重写代码。

使用agno实现该线性工作流的代码如下：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow.v2 import Workflow, Step

model = OpenAIChat(id="gpt-5-mini")

generator = Agent(name="generator", model=model, response_model=DraftCode,
    instructions="You are an expert Python programmer. Write code and a brief explanation.")

critic = Agent(name="critic", model=model, response_model=Critique,
    instructions="You are a senior code reviewer. Analyze for bugs, inefficiencies and PEP8 issues.")

refiner = Agent(name="refiner", model=model, response_model=RefinedCode,
    instructions="Rewrite the code, incorporating every suggestion from the critique.")

def generator_step(step_input):
    draft = generator.run(step_input.message).content
    step_input.workflow_session_state["draft"] = draft
    return draft

def critic_step(step_input):
    draft: DraftCode = step_input.workflow_session_state["draft"]
    return critic.run(f"Review this code:\n```python\n{draft.code}\n```").content

def refiner_step(step_input):
    draft: DraftCode = step_input.workflow_session_state["draft"]
    critique: Critique = step_input.previous_step_output.content
    prompt = (f"Original code:\n```python\n{draft.code}\n```\n"
              f"Critique: {critique.model_dump_json(indent=2)}\nProduce the refined code.")
    return refiner.run(prompt).content

reflection_wf = Workflow(
    name="reflection",
    session_state={"draft": None},
    steps=[
        Step(name="generator_step", executor=generator_step),
        Step(name="critic_step", executor=critic_step),
        Step(name="refiner_step", executor=refiner_step),
    ],
)

reflection_wf.run(message="Write a Python function to find the nth Fibonacci number.")

在上述代码中，Workflow.steps列表定义了确定性的执行边。上一步的输出自动作为下一步的previous_step_output，而跨非相邻步骤的数据则通过workflow_session_state进行显式传递。

Router怎么工作？

没有Router。 这是Reflection架构的关键技术边界：它不包含条件分支，也没有失败恢复机制。系统默认三步走完直接结束，无论中间结果如何。

为什么比单次生成强？

Reflection将一个模糊的任务拆解为三个更明确的子任务：尽量生成、专门找问题、只解决问题。核心经验表明，LLM作为Critic（批评者）往往比作为Generator（生成者）更稳定，因为它不需要从零创造，只需识别错误。

失败模式与升级时机

最大的问题是：它无法验证Refiner是否真的修好了Critic提到的问题。 它有Critique，但没有形成闭环验证。当需要根据中间结果继续行动，或需要接触外部世界时，就必须升级到下一代架构：要么引入工具交互（Tool Use），要么形成持续观察-行动回路（ReAct）。

Tool Use：文本世界到结构化世界的跨越

它要解决什么问题？

Reflection解决了生成“质量”的问题，但未解决“知识边界”的问题。不带工具的LLM被困在训练数据的参数知识截止日期中。Tool Use（工具使用）架构旨在让系统突破上下文与知识截止日期的封闭性，通过调用外部API或函数获取实时信息。

它的State是什么？

Tool Use的State本质上是一条事件日志：用户输入 → 模型输出 → 工具调用 → 工具返回 → 下一轮推理。在agno中，这条日志由Agent内部自动维护。一次agent.run()调用内部就完整运行了整个链路。这与Reflection需要手动维护State不同，Tool Use将状态管理托管给了框架会话上下文。

关键代码实现

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools

def get_stock_price(symbol: str) -> str:
    """Return the latest stock price for a given symbol."""
    return f"The current price of {symbol.upper()} is $172.35."

tool_agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[get_stock_price, DuckDuckGoTools()],
    instructions="Use tools to answer questions that need real-time data.",
    show_tool_calls=True,
)

tool_agent.run("What is Apple's current stock price?")

底层等价于以下概念循环：

while True:
    response = llm_with_tools.invoke(messages)
    messages.append(response)
    if not response.tool_calls:
        break
    for call in response.tool_calls:
        observation = tool_registry[call.name](**call.args)
        messages.append(ToolMessage(observation, tool_call_id=call.id))

这个while循环是agno Agent内置的核心逻辑。

新增的核心能力

Tool Use真正新增的能力不是“会调用函数”，而是：文本控制流可以跨越到结构化世界，再返回文本世界。 这个跨越点是整个Agent工程的第一道硬边界，使得LLM能够与现实世界的数据和服务进行交互。

失败模式

失败通常不来自LLM本身，而来自“边界层”：工具名幻觉、参数类型错误、工具返回格式不对、工具结果被模型错误综合。因此，Tool Use的关键难点不是Prompt，而是序列化与反序列化的稳定性。

局限性

Tool Use没有显式规定“一定要循环”。在最原始的实现里，模型可能只调一次工具就收尾。如果需要真正的观察-行动闭环，即根据工具返回结果动态决定下一步动作，就需要进入ReAct架构。

ReAct：Agent真正成形的地方

它要解决什么问题？

Tool Use的控制流尚显浅层，ReAct（Reasoning + Acting）架构旨在让工具结果不仅被消费一次，而是进入下一轮决策。从这里开始，Agent不再只是“会用工具”，而是“会根据新观察更新计划”。

State的语义变化

在ReAct中，Agent的消息序列不再只是对话历史，而是行动轨迹（Trace）：当前问题 → 上一步推理 → 工具调用 → 工具观测 → 新一轮判断。你可以把它看成隐式的工作记忆带。在agno中，这条Trace由Agent内部自动维护，一次agent.run()内部包含了用户输入、模型输出、工具调用、工具返回及二次推理，无需手写任何状态容器。

关键代码实现

agno内置的reasoning=True参数会将ReAct的“Thought / Action / Observation”显式写进Trace：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools

react_agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[DuckDuckGoTools(), YFinanceTools(stock_price=True, company_news=True)],
    instructions=[
        "You are a research assistant.",
        "Think step by step. For each step decide whether to use a tool or answer.",
        "After each tool observation, re-evaluate what you still need before answering.",
    ],
    reasoning=True,
    markdown=True,
    show_tool_calls=True,
)

react_agent.print_response(
    "Based on the latest news, should I be worried about AAPL next quarter?",
    stream=True,
)

关键不在于某一行代码，而是agno内部那条回边：只要Model回复里还有tool_calls，就回到Model再跑一次。这条tool → model的循环边，是整个Agent Architecture里最重要的回边之一，它将“单次调用”变成了“持续交互”。

拓扑本质

ReAct的最小闭环是一个持续的交互循环：

1. Thought：思考下一步做什么。
2. Action：执行工具调用。
3. Observation：获取工具返回结果。
4. Repeat：基于观察重新思考。

系统终于从“线性流程”进入“持续交互系统”。

为什么是80%任务的起点？

因为大多数任务并不需要复杂的全局规划，它们只需要：先看一眼 → 做个动作 → 根据反馈再决定下一步。这种模式已足够覆盖多轮搜索、多跳问答、网页研究及工具驱动的数据收集。

失败模式

局部贪心。 ReAct每次只基于当前Observation决策，容易走弯路、重复搜索、陷入局部最优，且无法提前安排多步任务。

升级时机

当任务需要显式的步骤顺序控制，或者需要避免局部贪心导致的效率低下时，就该升级到Planning架构。

Planning：把控制流本身变成模型输出

它要解决什么问题？

ReAct本质是在线贪心策略，对于需要顺序约束、步骤依赖和过程可追踪的任务，它开始显得力不从心。Planning（规划）架构旨在把原本隐式存在模型脑子里的执行顺序，显式拿出来，写进State。系统第一次把“控制流”对象化了。

State新增了什么？

Planning架构在State中新增了一个显式的计划列表：

class Plan(BaseModel):
    steps: List[str] = Field(description="Ordered list of tool/sub-questions.")

Workflow的session_state里还要带上中间结果：

session_state = {"plan": [], "intermediate": []}

在ReAct里，下一步做什么是临时决定的；而在Planning里，下一步做什么先被生成出来，然后才执行。

关键代码实现

agno的Workflow提供Loop Step，天然表达“循环执行直到Plan为空”：

from agno.workflow.v2 import Workflow, Step, Loop
from agno.tools.duckduckgo import DuckDuckGoTools

planner = Agent(name="planner", model=OpenAIChat(id="gpt-5-mini"), response_model=Plan,
    instructions="Decompose the user request into a list of atomic tool-queryable steps.")

executor = Agent(name="executor", model=OpenAIChat(id="gpt-5-mini"),
    tools=[DuckDuckGoTools()],
    instructions="Answer exactly one sub-question using tools.")

synthesizer = Agent(name="synthesizer", model=OpenAIChat(id="gpt-5-mini"),
    instructions="Combine intermediate findings into a final answer.")

def plan_step(step_input):
    plan: Plan = planner.run(step_input.message).content
    step_input.workflow_session_state["plan"] = list(plan.steps)
    step_input.workflow_session_state["intermediate"] = []
    return plan

def execute_step(step_input):
    state = step_input.workflow_session_state
    next_q = state["plan"].pop(0)
    obs = executor.run(next_q).content
    state["intermediate"].append(f"Q: {next_q}\nA: {obs}")
    return obs

def synth_step(step_input):
    state = step_input.workflow_session_state
    notes = "\n\n".join(state["intermediate"])
    return synthesizer.run(
        f"Question: {step_input.message}\nNotes:\n{notes}\nFinal answer:"
    ).content

def plan_is_empty(_outputs) -> bool:
    """Loop 终止条件：agno 的 end_condition 只接收一个参数（StepOutput 列表）。"""
    return len(planning_wf.session_state["plan"]) == 0

planning_wf = Workflow(
    name="planning",
    session_state={"plan": [], "intermediate": []},
    steps=[
        Step(name="plan", executor=plan_step),
        Loop(
            name="execute_all",
            steps=[Step(name="execute_one", executor=execute_step)],
            end_condition=plan_is_empty,
        ),
        Step(name="synthesize", executor=synth_step),
    ],
)

planning_wf.run(message="Compare the latest revenue of AAPL and MSFT and explain the gap.")

路由逻辑从“LLM在Prompt里决定下一步”变成了“一个数据结构是否还有剩余项”——这才是Planning真正的结构变化。

新增的核心能力

Planning新增的不是“更聪明的思考”，而是：把未来控制流提前Materialize（具体化）成一个数据结构。 现在可以可视化计划、检查计划、修改计划、追踪执行进度。

失败模式

它过于乐观，一旦Plan错了，后面每一步都可能是错的。可预测性增强了，适应性下降了。

升级时机

当你不再相信工具会稳定成功，或者执行过程中可能出现不可预见的错误时，就必须引入PEV（Plan-Execute-Verify）架构。

关键代码实现：Ensemble 模式的 Fan-out/Fan-in

在 agno 框架中，Workflow 原生支持通过 Parallel 步骤实现任务的扇出（Fan-out），随后利用聚合步骤完成扇入（Fan-in）。这种结构非常适合需要多视角验证的场景。

from agno.workflow.v2 import Workflow, Step, Parallel
from pydantic import BaseModel
from typing import List

class FinalRecommendation(BaseModel):
    final_recommendation: str
    confidence_score: float
    synthesis_summary: str
    identified_opportunities: List[str]
    identified_risks: List[str]

bullish = Agent(
    name="bullish", 
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="You are a growth-oriented bullish analyst. Argue why to invest."
)
value = Agent(
    name="value", 
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="You are a value analyst. Focus on margin of safety."
)
quant = Agent(
    name="quant", 
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions="You are a quantitative analyst. Focus on ratios and trends."
)

cio = Agent(
    name="cio", 
    model=OpenAIChat(id="gpt-4o-mini"),
    response_model=FinalRecommendation,
    instructions=(
        "You are the CIO. Synthesize three analyst views into one final "
        "recommendation. Do NOT hide disagreement — explicitly list risks."
    )
)

def run_one(agent):
    """工厂函数：为每个 Agent 生成独立的执行步骤"""
    def _step(si):
        # 执行 Agent 推理
        out = agent.run(si.message).content
        # 将结果写入共享的 workflow_session_state，实现并行写入
        si.workflow_session_state.setdefault("views", {})[agent.name] = out
        return out
    return _step

def aggregate_step(si):
    """聚合步骤：从共享状态读取所有观点，交由 CIO 综合"""
    views = si.workflow_session_state.get("views", {})
    # 格式化所有分析师的观点
    body = "\n\n".join(f"[{name}]\n{view}" for name, view in views.items())
    # CIO 基于完整上下文生成最终决策
    return cio.run(f"Question: {si.message}\nAnalyst views:\n{body}").content

ensemble_wf = Workflow(
    name="ensemble",
    session_state={"views": {}}, # 初始化共享状态字典
    steps=[
        Parallel(
            name="analysts",
            steps=[
                Step(name="bullish", executor=run_one(bullish)),
                Step(name="value", executor=run_one(value)),
                Step(name="quant", executor=run_one(quant)),
            ],
        ),
        Step(name="cio_synth", executor=aggregate_step),
    ],
)

ensemble_wf.run(message="Should I buy AAPL right now?")

代码解析：

• session_state 共享机制：Parallel 步骤中的各个分支通过 si.workflow_session_state 共享数据。这是实现 Fan-in 的关键，避免了复杂的返回值传递逻辑。
• 结构化输出约束：FinalRecommendation 模型强制 CIO 输出 identified_risks，确保集成过程不会简单地取平均值，而是保留冲突信息。
• 并发执行：Parallel 步骤确保三位分析师同时运行，显著降低延迟，尤其当单个 Agent 推理耗时较长时优势明显。

Ensemble 与 Multi-Agent 的本质区别

许多开发者容易混淆 Ensemble（集成） 与 Multi-Agent（多智能体协作），两者的核心差异在于任务分配逻辑：

• Multi-Agent 侧重于 分工（Division of Labor）。不同的 Agent 负责处理任务的不同子部分，例如一个负责搜索，一个负责写作，一个负责校对。它们串联起来完成一个复杂流程。
• Ensemble 侧重于 冗余（Redundancy）。不同的 Agent 分析同一个问题，提供独立的视角或解法。其目的不是拆分任务，而是通过多样性来抵消单一模型的偏差。

新增能力：通过冗余降低推理偏差

Ensemble 架构的核心价值在于 用多视角和冗余来降低单次推理的随机性与偏差。这与传统机器学习中的集成学习（Ensemble Learning）理念一致：单个模型可能在特定分布上表现不佳，但多个独立模型的误差不完全相关，通过聚合可以显著提升鲁棒性。

在 LLM 场景下，这意味着即使某个模型因提示词敏感性或训练数据偏见产生幻觉，其他视角的 Agent 仍可能提供正确或平衡的观点，从而在聚合层被识别并修正。

潜在失败模式

尽管 Ensemble 提升了可靠性，但也引入了新的风险：

1. 成本线性增长：每次请求需调用 N 次模型，Token 消耗与延迟随 Agent 数量线性增加。
2. 共享偏见风险：如果所有 Agent 使用相同的基础模型且提示词相似，它们可能陷入相同的逻辑误区，导致“伪多样性”。
3. 聚合器瓶颈：Aggregator（如 CIO）可能强行合并不该合并的冲突观点，或者未能准确权衡不同观点的可信度。
4. 分歧掩盖：若输出格式未强制要求列出分歧，最终的“综合意见”可能平滑掉关键的风险信号。

因此，Ensemble 的关键不在于“取平均”，而在于 保留冲突信息并解释冲突。这也是为什么在 FinalRecommendation 中必须包含 identified_risks 字段的原因。

适用场景

Ensemble 架构最适合 高风险判断、事实核查、投资建议、复杂研究结论 等场景。凡是你不想把决定押在一次概率性输出上的场景，都应考虑引入集成机制。

> 📎 参考实现：13_ensemble.ipynb

13. Episodic + Semantic Memory：记忆不是简单的上下文堆砌

记忆系统并非仅仅是“外挂知识库”，而是 Agent 状态（State）的外延扩展。真正重要的不是存储了多少数据，而是系统如何区分短暂的事件与长期的事实，以及如何避免错误信息的长期固化。

它要解决什么问题？

此前的架构大多假设对话结束后系统即“失忆”。然而，用户期望 Agent 能够：

1. 记得偏好：如饮食禁忌、沟通风格。
2. 记得历史：之前讨论过的背景信息。
3. 记得事实：长期稳定不变的领域知识。

为此，我们需要两种不同类型的记忆机制：

• Episodic Memory（情景记忆）：记录“发生过什么”，通常以事件摘要形式存入向量数据库，用于检索历史交互。
• Semantic Memory（语义记忆）：记录“什么是真的”，通常以实体关系或结构化数据形式存储，用于构建长期知识图谱或事实库。

关键代码实现

agno 通过 Memory 模块管理长期用户偏好与事件，通过 AgentKnowledge 管理向量检索知识。两者结合可快速构建双记忆系统。

from agno.agent import Agent
from agno.memory.v2.memory import Memory
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.knowledge.text import TextKnowledgeBase
from agno.vectordb.lancedb import LanceDb, SearchType
from agno.embedder.openai import OpenAIEmbedder
from agno.storage.sqlite import SqliteStorage

memory = Memory(
    db=SqliteMemoryDb(table_name="user_memories", db_file="tmp/memory.db"),
    model=OpenAIChat(id="gpt-4o-mini"),
)

knowledge = TextKnowledgeBase(
    path="data/facts",
    vector_db=LanceDb(
        table_name="facts",
        uri="tmp/lancedb",
        search_type=SearchType.hybrid, # 混合搜索提升准确率
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)
knowledge.load(recreate=False)

mem_agent = Agent(
    name="memorized",
    model=OpenAIChat(id="gpt-4o-mini"),
    memory=memory,                # 启用情景记忆
    enable_agentic_memory=True,   # 允许 Agent 主动写入/检索情景记忆
    enable_user_memories=True,    # 启用用户偏好记忆
    knowledge=knowledge,          # 启用语义知识库
    search_knowledge=True,        # 自动检索相关知识
    add_history_to_messages=True, # 将短期历史加入上下文
    num_history_responses=5,      # 保留最近5轮对话
    storage=SqliteStorage(table_name="sessions", db_file="tmp/sessions.db"),
    markdown=True,
)

mem_agent.print_response(
    "I'm allergic to peanuts and prefer low-carb meals. Remember that.",
    user_id="alice",
)

mem_agent.print_response(
    "Suggest a dinner plan for me.",
    user_id="alice",
)

代码解析：

• enable_agentic_memory=True：这是关键开关，它让 Agent 不仅被动检索，还能主动判断哪些信息值得存入长期记忆，实现了记忆的动态更新。
• user_id 隔离：通过 user_id 确保不同用户的记忆数据隔离，保障隐私与准确性。
• 混合存储策略：Memory 处理个性化交互历史，Knowledge 处理通用事实，两者互补而非重叠。

新增能力：状态空间的扩展

系统的 State 从单纯的“当前上下文窗口”扩展为 “图内状态 + 图外可检索历史状态”。Agent 的能力不再仅依赖有限的 Token 窗口，而取决于其从海量历史中精准提取相关信息的能力。

工作流程

记忆模块不再是附加组件，而是主控制流的一部分。当 enable_agentic_memory 开启时，Agent 会在每次响应前自动执行“检索-增强-生成-写入”的闭环链路。

潜在失败模式

1. 错误记忆污染：一旦错误信息被写入长期记忆，将在后续所有交互中持续产生负面影响，且难以察觉。
2. 召回噪声：Episodic Recall 可能召回语义相似但语境无关的历史片段，干扰当前推理。
3. 事实过时：Semantic Graph 中若存入过时事实且未建立更新机制，会导致系统性错误。
4. 抽取质量差：若记忆抽取指令不佳，可能导致记忆结构混乱，降低检索效率。

记忆让系统更强大，但也让错误变得更持久。因此，记忆的可解释性与可删除性至关重要。

何时不够用？

当你发现向量检索只能做到“找相似”，却无法进行复杂的“关系推理”（如“A 是 B 的子公司，B 收购了 C，问 A 与 C 的关系”）时，就需要引入 Graph / World-Model Memory。

> 📎 参考实现：08_episodic_with_semantic.ipynb

14. Graph / World-Model Memory：从回忆到关系推理

它要解决什么问题？

向量检索擅长回答“哪段历史最像当前问题”，但无法天然回答“实体 A 与实体 B 之间隔了几跳关系”。Graph Memory 旨在将知识从非结构化的 Chunk 组织提升到结构化的关系网络，解决多跳推理问题。

技术核心

该架构包含两个关键步骤：

1. 知识抽取：从非结构化文本中提取实体（Nodes）和关系（Relationships）。
2. 图查询生成：将自然语言问题转化为图查询语言（如 Cypher），在知识图谱中执行精确查询。

流程如下：Text -> Knowledge Graph -> Text-to-Cypher -> Query -> Answer

关键代码实现

from pydantic import BaseModel, Field
from typing import List

class Node(BaseModel):
    id: str = Field(description="Unique name or identifier for the entity.")
    type: str = Field(description="Entity type, e.g., Person, Company.")

class Relationship(BaseModel):
    source: Node
    target: Node
    type: str = Field(description="Relationship verb in ALL_CAPS, e.g., WORKS_FOR, ACQUIRED.")

class KnowledgeGraph(BaseModel):
    relationships: List[Relationship]

graph_maker = Agent(
    name="graph_maker",
    model=OpenAIChat(id="gpt-4o-mini"),
    response_model=KnowledgeGraph,
    instructions=(
        "Extract entities (nodes) and relationships from the given text. "
        "Relationship type should be an ALL_CAPS verb."
    ),
)

from agno.tools.neo4j import Neo4jTools

graph_query_agent = Agent(
    name="graph_query",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[Neo4jTools(url="bolt://localhost:7687", user="neo4j", password="...")],
    instructions=(
        "You answer questions over a Neo4j knowledge graph. "
        "First generate a Cypher query, run it, then synthesize a natural answer. "
        "If the first query returns nothing, rewrite and retry once."
    ),
)

kg = graph_maker.run("Tim Cook is the CEO of Apple. Apple acquired Beats in 2014.").content

graph_query_agent.print_response("Which companies did Apple acquire and in which year?")

代码解析：

• 结构化抽取：graph_maker 使用 Pydantic 模型强制输出标准化的节点与关系，便于后续存入图数据库。
• Text-to-Cypher：graph_query_agent 利用 LLM 的自然语言理解能力生成准确的 Cypher 查询语句，并通过 Neo4jTools 执行。
• 自我修正：指令中包含“若查询无结果则重试”，提高了图查询的鲁棒性。

新增能力：结构性推理

核心变化是从 “相似性召回” 升级到 “结构性推理”。这在企业知识问答、组织架构分析、供应链追踪等需要明确实体间关联的场景中至关重要。

潜在失败模式

1. 抽取错误导致图污染：错误的实体链接或关系类型会破坏图的连通性。
2. Schema 设计不佳：若本体论（Ontology）设计不合理，图将难以支撑复杂查询。
3. Text-to-Cypher 生成错误：LLM 可能生成语法正确但逻辑错误的查询语句。
4. 合成层误读：即使图查询结果正确，LLM 在最终总结时可能误解数据含义。

Graph Memory 的关键不仅在于 LLM，更在于整个知识建模链条的质量控制。

> 📎 参考实现：12_graph.ipynb

15. Tree-of-Thoughts (ToT)：从单路径生成到空间搜索

它要解决什么问题？

真正困难的问题往往不是“链不够长”，而是“路径会分叉，且需要回溯”。标准的 Chain-of-Thought (CoT) 是线性的，一旦中间步骤出错，全盘皆输。ToT 将推理从单路径生成升级为对候选路径空间的 搜索（Search）。

State 的定义

在 ToT 中，State 的单位不再是“一个当前答案”，而是“多条候选路径的集合”。

class ToTState(BaseModel):
    problem: str
    active_paths: List[List["PuzzleState"]] # 当前存活的推理路径
    solution: Optional[List["PuzzleState"]] = None

关键代码实现

以经典的“狼、羊、白菜”过河谜题为例，这是一个典型的需要回溯的状态空间搜索问题。在 agno 中，LLM 负责 生成候选动作，而程序化代码负责 搜索树的扩展、剪枝与验证。

架构原则：不要把搜索控制交给 LLM，交给代码。

import copy
from pydantic import BaseModel, Field
from typing import List, Optional

class PuzzleState(BaseModel):
    left_bank: frozenset = Field(default_factory=lambda: frozenset({"wolf", "goat", "cabbage"}))
    right_bank: frozenset = Field(default_factory=frozenset)
    boat_location: str = "left"
    move_description: str = "Initial state."

    class Config:
        arbitrary_types_allowed = True

    def is_valid(self) -> bool:
        """检查当前状态是否合法（狼不吃羊，羊不吃菜）"""
        dangerous = [("wolf", "goat"), ("goat", "cabbage")]
        # 船不在的那一侧是无人看管的
        unguarded = self.left_bank if self.boat_location == "right" else self.right_bank
        return not any({a, b}.issubset(unguarded) for a, b in dangerous)

    def is_goal(self) -> bool:
        return self.right_bank == frozenset({"wolf", "goat", "cabbage"})

class Proposal(BaseModel):
    moves: List[str] = Field(description="Candidate next moves.")

proposer = Agent(
    name="proposer",
    model=OpenAIChat(id="gpt-4o-mini"),
    response_model=Proposal,
    instructions="Given the current state, propose up to 3 distinct next moves.",
)

def expand(state: PuzzleState) -> List[PuzzleState]:
    """纯程序化展开：根据 proposer 的建议生成新状态，并验证合法性"""
    # 此处省略具体状态转移逻辑，实际应包含物理规则验证
    ...

def tot_solve(initial: PuzzleState, max_depth: int = 10):
    """广度优先搜索 (BFS) 求解器"""
    active_paths = [[initial]]
    for _ in range(max_depth):
        new_paths = []
        for path in active_paths:
            current_state = path[-1]
            # 获取 LLM 建议的动作
            proposal = proposer.run(str(current_state)).content
            for move in proposal.moves:
                next_state = apply_move(current_state, move) # 假设的应用动作函数
                if not next_state.is_valid():
                    continue # 剪枝：非法状态
                if next_state in path:
                    continue # 剪枝：循环检测

                new_path = path + [next_state]
                if next_state.is_goal():
                    return new_path # 找到解
                new_paths.append(new_path)
        active_paths = new_paths
        if not active_paths:
            break
    return None

代码解析：

• 职责分离：LLM (proposer) 负责创造性地提出可能的动作，Python 代码 (expand, tot_solve) 负责严格的逻辑验证、去重和路径管理。
• 剪枝策略：通过 is_valid() 和循环检测，大幅减少搜索空间，避免无效计算。

新增能力：推理即搜索

ToT 新增的核心能力是 把推理问题转写成搜索问题。这与 ReAct 或 Planning 不同：后者关注控制流设计，而 ToT 关注 搜索空间的设计与遍历。

潜在失败模式

• 组合爆炸：若分支因子过大且剪枝不力，搜索空间将指数级增长。
• 适用性局限：ToT 并非通用架构，仅适用于具有明确状态空间和验证规则的离散问题（如数学证明、编程调试、逻辑谜题）。

> 📎 参考实现：09_tree_of_thoughts.ipynb

16. Mental Loop / Simulator：行动前的内部试错

它要解决什么问题？

在机器人控制、金融交易、生产系统配置等场景中，真实世界的试错成本极高甚至不可逆。Mental Loop 旨在 将试错从真实世界迁移到模拟世界，通过反事实推理（Counterfactual Reasoning）预演行动后果。

关键代码实现

模拟器是一个普通的 Python 对象，决策 Agent 通过工具将其视为一个可快照、推进和回滚的沙箱。

import copy
import numpy as np
from pydantic import BaseModel, Field

class Portfolio(BaseModel):
    cash: float = 10000.0
    shares: int = 0
    def value(self, price: float) -> float:
        return self.cash + self.shares * price

class MarketSimulator(BaseModel):
    day: int = 0
    price: float = 100.0
    volatility: float = 0.1
    drift: float = 0.01
    portfolio: Portfolio = Field(default_factory=Portfolio)

    class Config:
        arbitrary_types_allowed = True

    def step(self, action: str, amount: float = 0.0):
        """模拟市场一步演化"""
        if action == "buy":
            n = int(amount); cost = n * self.price
            if self.portfolio.cash >= cost:
                self.portfolio.shares += n; self.portfolio.cash -= cost
        elif action == "sell":
            n = min(int(amount), self.portfolio.shares)
            self.portfolio.shares -= n; self.portfolio.cash += n * self.price

        # 模拟价格波动
        self.price *= 1 + float(np.random.normal(self.drift, self.volatility))
        self.day += 1

REAL = MarketSimulator()

def simulate_action(action: str, amount: float, horizon: int = 5) -> str:
    """在克隆的环境中预演行动"""
    sim = copy.deepcopy(REAL) # 关键：深拷贝，不影响真实环境
    sim.step(action, amount)
    for _ in range(horizon - 1):
        sim.step("hold") # 后续几天持有
    return f"Simulated value after {horizon} days: ${sim.portfolio.value(sim.price):.2f}"

def execute_action(action: str, amount: float) -> str:
    """在真实环境中执行行动"""
    REAL.step(action, amount)
    return f"Executed: {action} {amount}. Portfolio now ${REAL.portfolio.value(REAL.price):.2f}."

trader = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[simulate_action, execute_action],
    instructions=[
        "Before committing any action with execute_action, first call simulate_action.",
        "If the simulated outcome is worse than holding, do not execute.",
    ],
    show_tool_calls=True,
)

trader.print_response("Decide whether to buy AAPL today, and how much.")

代码解析：

• copy.deepcopy：确保模拟过程完全隔离，不会污染真实状态。
• 双工具模式：simulate_action 用于低成本探索，execute_action 用于高置信度执行。
• 指令约束：强制 Agent 先模拟后执行，形成内部思维闭环。

新增能力：反事实执行

系统不再直接问“下一步做什么”，而是先问 “如果我这么做，会发生什么？”。这种能力显著降低了高风险场景下的决策失误率。

潜在失败模式

• Simulation-Reality Gap（模拟-现实差距）：模拟器越不真实，Agent 越可能做出“模拟完美、现实灾难”的决定。该架构的上限往往取决于 Simulator 的保真度，而非 LLM 本身。

> 📎 参考实现：10_mental_loop.ipynb

17. Dry-Run Harness：将副作用关进闸门

安全架构的核心不是让 Agent 更保守，而是将 “是否允许执行” 做成显式的控制流。Dry-Run Harness 解决外部副作用的控制，通常与 Metacognitive（内部边界感知）配合使用。

它要解决什么问题？

当 Agent 具备发邮件、发帖、下单、改配置等能力时，核心问题是：它在执行前能否被拦截？ Dry-Run 将所有真实动作拆分为 Preview（预览） 和 Execute（执行） 两种模式。

关键代码实现

import datetime, hashlib
from typing import List

def publish_post(content: str, hashtags: List[str], dry_run: bool = True) -> str:
    """发布社交媒体帖子。若干跑模式，仅预览无副作用。"""
    ts = datetime.datetime.now().isoformat()
    full = f"{content}\n\n" + " ".join(f"#{h}" for h in hashtags)
    if dry_run:
        return f"[DRY RUN @ {ts}] Would publish:\n---\n{full}\n---"
    post_id = hashlib.md5(full.encode()).hexdigest()[:8]
    return f"[LIVE @ {ts}] Published id={post_id}"

proposer = Agent(
    name="proposer",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[publish_post],
    instructions=[
        "When asked to post, FIRST call publish_post with dry_run=True to preview.",
        "After the preview, stop and wait for human approval. Do NOT call dry_run=False yourself.",
    ],
    show_tool_calls=True,
)

def propose_step(si):
    return proposer.run(si.message).content

def approve_step(si):
    preview = si.previous_step_output
    print(f"\n--- PREVIEW ---\n{preview}\n---")
    decision = input("Approve and go live? (y/n): ").strip().lower()
    si.workflow_session_state["approved"] = (decision == "y")
    return decision

def commit_step(si):
    if not si.workflow_session_state.get("approved"):
        return "Rejected by human reviewer. Nothing was published."
    # 人类批准后，再次调用 Agent 执行真实操作
    return proposer.run(
        "Now actually publish the same post by calling publish_post with dry_run=False."
    ).content

dryrun_wf = Workflow(
    name="dryrun",
    steps=[
        Step(name="propose", executor=propose_step),
        Step(name="approve", executor=approve_step),
        Step(name="commit", executor=commit_step),
    ],
)

代码解析：

• 工具级干跑：publish_post 内置 dry_run 参数，从底层保证安全性。
• 人机协同审批：approve_step 引入人工介入，作为最终的安全闸门。
• 两阶段执行：先生成预览，经确认后再生成真实执行指令，避免 Agent 直接跳过审批。

新增能力：副作用的可审批化

Dry-Run 的核心价值在于 让副作用在控制流上变成可审批对象。它将安全问题从“模型对齐”层面下沉到“系统工程”层面，更加可靠。

潜在失败模式

1. 人工审批瓶颈：高频操作下，人工审批成为效率瓶颈。
2. 环境不一致：预览环境与真实执行环境存在细微差异，导致预览通过但执行失败。
3. 信息泄漏：过于详细的预览可能暴露敏感数据。

尽管如此，在涉及真实世界操作的生产系统中，Dry-Run 几乎是必须的架构组件。

> 📎 参考实现：14_dry_run.ipynb

18. Reflexive Metacognitive Agent：系统对自身边界的思考

它要解决什么问题？

此前的系统都在关注“如何完成任务”，而 Metacognitive Agent 首先关注 “这个任务我到底该不该做？”。这是一种更高级的控制流，因为它开始对自身能力边界进行建模。

关键代码实现

from pydantic import BaseModel, Field
from typing import Optional

class MetacognitiveAnalysis(BaseModel):
    confidence: float = Field(description="0.0~1.0, confidence in safely answering.")
    strategy: str = Field(description="'reason_directly' | 'use_tool' | 'escalate'.")
    reasoning: str
    tool_to_use: Optional[str] = None

AGENT_SELF_MODEL = {
    "knowledge_domains": ["general health", "nutrition", "exercise"],
    "tools_available": ["symptom_checker"],
    "confidence_threshold": 0.7,
    "high_risk_topics": ["prescription dosage", "emergency medical advice"],
}

self_model_agent = Agent(
    name="self_model",
    model=OpenAIChat(id="gpt-4o-mini"),
    response_model=MetacognitiveAnalysis,
    instructions=(
        f"Your self-model is: {AGENT_SELF_MODEL}. "
        "For each user query, estimate confidence and pick a strategy: "
        "'reason_directly' if confidence >= threshold and topic is low-risk, "
        "'use_tool' if a matching tool exists, 'escalate' otherwise."
    ),
)

def symptom_checker(symptoms: str) -> str:
    return f"Reference info for: {symptoms}"

responder = Agent(model=OpenAIChat(id="gpt-4o-mini"), tools=[symptom_checker])

def meta_step(si):
    analysis: MetacognitiveAnalysis = self_model_agent.run(si.message).content
    si.workflow_session_state["analysis"] = analysis
    return analysis

def meta_router(si):
    analysis = si.workflow_session_state["analysis"]
    if analysis.strategy == "reason_directly":
        return [Step(name="answer", executor=lambda s: responder.run(s.message).content)]
    if analysis.strategy == "use_tool":
        return [Step(name="tool_answer",
                     executor=lambda s: responder.run(
                         f"Use the {analysis.tool_to_use} tool to help answer: {s.message}"
                     ).content)]
    return [Step(name="escalate",
                 executor=lambda s: "I'm not confident or allowed to answer this. Escalating to a human expert.")]

metacog_wf = Workflow(
    name="metacognitive",
    steps=[
        Step(name="self_model", executor=meta_step),
        Router(name="route_strategy", selector=meta_router),
    ],
)

代码解析：

• 自我模型（Self-Model）：明确定义 Agent 的知识域、工具集和风险阈值。
• 前置路由：在执行任何具体任务前，先由 self_model_agent 评估置信度和策略。
• 动态路由：Router 根据元认知分析结果，动态选择直接回答、调用工具或升级人工。

新增能力：边界感知

系统新增了 知道“我知道什么、不知道什么、需要什么工具、什么时候该交给人类” 的能力。

潜在失败模式

• 置信度校准失效：LLM 往往过度自信。若低估自己，系统会过度保守；若高估自己，则在高风险场景下产生危险行为。

为什么关键？

在医疗、法律、金融等领域，Agent 最强的能力不是“回答”，而是 “拒绝”。Metacognitive Agent 赋予了系统这种拒绝的勇气和依据。

> 📎 参考实现：17_reflexive_metacognitive.ipynb

19. Self-Improvement Loop：进化回路

它要解决什么问题？

单一的 Reflection（反思）仅提供一次批判机会。追求高质量输出的系统需要的是：生成 → 评估 → 修订 → 再评估 的闭环。Self-Improvement Loop 将质量优化构建为一个迭代进化的回路。

关键代码实现

关键在于终止条件的设定，agno 的 Loop.end_condition 天然支持此逻辑。

class EmailCritique(BaseModel):
    is_approved: bool
    feedback: str

class MarketingEmail(BaseModel):
    subject: str
    body: str

generator = Agent(name="email_gen", model=OpenAIChat(id="gpt-4o-mini"),
    response_model=MarketingEmail,
    instructions="Write a marketing email. Follow feedback if provided.")

critic = Agent(name="email_critic", model=OpenAIChat(id="gpt-4o-mini"),
    response_model=EmailCritique,
    instructions="Approve only if subject is compelling AND body has a clear CTA AND tone is on-brand.")

class GoldStandardMemory:
    def __init__(self):
        self.examples: List[MarketingEmail] = []

    def few_shot_block(self) -> str:
        if not self.examples:
            return "No gold examples yet."
        return "\n\n---\n\n".join(
            f"Subject: {e.subject}\nBody:\n{e.body}" for e in self.examples[-3:]
        )

    def add(self, e: MarketingEmail):
        self.examples.append(e)

GOLD = GoldStandardMemory()

def gen_step(si):
    state = si.workflow_session_state
    prior = state.get("last_email")
    critique = state.get("last_critique")

    prompt = f"Task: {si.message}\nGold examples:\n{GOLD.few_shot_block()}\n\n"
    if prior and critique:
        prompt += (f"Previous draft:\nSubject: {prior.subject}\nBody:\n{prior.body}\n"
                   f"Critic feedback: {critique.feedback}\nRewrite to address the feedback.")

    email: MarketingEmail = generator.run(prompt).content
    state["last_email"] = email
    state["revision"] = state.get("revision", 0) + 1
    return email

def critic_step(si):
    email = si.workflow_session_state["last_email"]
    verdict: EmailCritique = critic.run(
        f"Review:\nSubject: {email.subject}\nBody:\n{email.body}"
    ).content
    si.workflow_session_state["last_critique"] = verdict
    if verdict.is_approved:
        GOLD.add(email) # 只有通过审核的邮件才加入金标准库
    return verdict

def should_stop(_outputs) -> bool:
    """Loop 终止条件：critic 批准，或达到最大修订次数。"""
    state = self_improve_wf.session_state
    last = state.get("last_critique")
    if last is not None and last.is_approved:
        return True
    if state.get("revision", 0) >= 3:
        return True
    return False

self_improve_wf = Workflow(
    name="self_improve",
    session_state={"revision": 0},
    steps=[
        Loop(
            name="refine_loop",
            steps=[
                Step(name="gen", executor=gen_step),
                Step(name="critic", executor=critic_step),
            ],
            end_condition=should_stop,
        ),
    ],
)

代码解析：

• 迭代优化：Loop 结构允许 Generator 和 Critic 多次交互，直至满足标准。
• 金标准积累：GoldStandardMemory 将通过审核的结果存入上下文，作为未来生成的 Few-Shot 示例，实现 跨任务的知识沉淀。

新增能力：双层进化

1. 单次任务内迭代：通过多轮修订提升单次输出质量。
2. 跨任务积累：通过金标准库的提升，逐步提高系统的基线水平。

潜在失败模式

• Critic 标准不稳：若 Critic 过于宽松或严苛，循环可能无效或死锁。
• 收益递减：超过一定轮次后，修改带来的提升微乎其微。
• 样本污染：若低质量样本误入金标准库，将反向毒化未来生成。

Self-Improvement 不是“自动变好”，而是“在严格约束下有概率变好”。

> 📎 参考实现：15_RLHF.ipynb

20. Cellular Automata：LLM 退出主循环，智能从局部规则中涌现

它要解决什么问题？

此前所有架构都默认存在一个中心 Agent 或 Orchestrator。Cellular Automata（细胞自动机） 彻底打破这一前提：有些问题不需要中心规划，而适合通过局部规则产生全局行为。

关键代码实现

class CellAgent(BaseModel):
    type: str  # 'EMPTY' | 'OBSTACLE' | 'GOAL'
    pathfinding_value: float = float("inf")

    def update(self, neighbors: List["CellAgent"]):
        if self.type == "OBSTACLE":
            return
        if self.type == "GOAL":
            self.pathfinding_value = 0
            return
        # 局部规则：取邻居最小值 + 1
        m = min((n.pathfinding_value for n in neighbors), default=float("inf"))
        self.pathfinding_value = min(self.pathfinding_value, m + 1)

def neighbors_of(snapshot, r, c):
    """获取四邻域"""
    H, W = len(snapshot), len(snapshot[0])
    out = []
    for dr, dc in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
        nr, nc = r + dr, c + dc
        if 0 <= nr < H and 0 <= nc < W:
            out.append(snapshot[nr][nc])
    return out

def run_ca(grid, steps=50):
    for _ in range(steps):
        snapshot = [[copy.deepcopy(c) for c in row] for row in grid]
        for r in range(len(grid)):
            for c in range(len(grid[0])):
                grid[r][c].update(neighbors_of(snapshot, r, c))

rule_designer = Agent(
    name="rule_designer",
    model=OpenAIChat(id="gpt-4o-mini"),
    instructions=(
        "You design local update rules for a cellular automaton. "
        "Given a high-level objective, propose a minimal Python update function "
        "that each cell will execute based only on its neighbors."
    ),
)

rule_designer.print_response(
    "Design a CA rule so that every empty cell eventually stores the shortest "
    "path length to the nearest GOAL, treating OBSTACLE as impassable."
)

代码解析：

• 去中心化：每个 CellAgent 仅根据邻居状态更新自身，无全局控制器。
• LLM 角色转变：LLM 不再参与每一步的执行，仅负责设计局部更新规则（update 函数）。
• 涌现行为：通过简单的局部规则迭代，全局上涌现出最短路径场。

新增能力：分布式涌现

这是范式的切换：从中心控制转向分布式涌现。适用于大规模并行、容错性要求高、无需精确全局规划的场景。

与前述架构的最大区别

LLM 退出了执行主回路。它最多负责规则设计、参数选择或系统解释，真正的求解过程由大量局部节点同步演化完成。

潜在失败模式

• 规则设计不当：局部规则无法收敛到期望的全局状态。
• 收敛缓慢：可能需要大量迭代步数。
• 错误涌现：可能涌现出非预期的全局结构。

> 📎 参考实现：16_cellular_automata.ipynb

21. Evaluator 不是可选项

本项目反复强调：Agent 不是只要能跑就行，必须能评估。至少需要五类 Evaluator：

1. LLM-as-a-Judge：用独立 Agent 对输出进行打分或排序。
2. 内置 Critic：直接控制循环是否继续（如 Loop.end_condition）。
3. 程序化验证：如 is_valid() / is_goal() 等硬约束代码。
4. Human-in-the-Loop：用人工审批作为最后闸门（如 Dry-Run）。
5. 演示式验证：在多场景下运行系统，观察行为一致性。

为什么 Evaluator 是核心？

一旦系统开始反思、重规划、迭代、决定是否执行或升级，你就进入了 “闭环系统”。闭环系统如果没有 Evaluator，就不知道何时停、何时改、何时拒绝。

没有 Evaluator 的 Agent，大概率只是一个会循环的 Prompt，不是一个可靠的系统。

22. 架构演化表

23. 怎么选？问你缺哪种控制能力

不要问“哪个架构最好”，而要问“我缺哪种控制能力”。

快速选型指南

• 如果你缺的是输出质量：先上 Reflection；需要多轮逼近和长期改进，上 Self-Improvement。
• 如果你缺的是与世界交互：简单任务上 Tool Use；多步动态任务上 ReAct。
• 如果你缺的是显式步骤控制：上 Planning；工具不可靠，再升级到 PEV。
• 如果你缺的是角色分工：固定分工用 Multi-Agent；动态持续调度用 Blackboard；入口分诊用 Meta-Controller；同题多视角冗余用 Ensemble。
• 如果你缺的是长期状态：记历史事件用 Episodic Memory；做关系推理用 Graph / World-Model Memory。
• 如果你缺的是求解范式：需要回溯搜索用 Tree-of-Thoughts；需要先模拟后执行用 Mental Loop；需要去中心化求解用 Cellular Automata。
• 如果你缺的是安全边界：需要副作用审批用 Dry-Run Harness；需要知道自己不能做什么用 Metacognitive Agent。

24. 最终结论：Agent 架构的演化，本质上是在设计更好的控制流

看完整个项目并用 agno 重写一遍后，我最大的感受不是“Agent 真多”，而是一个更朴素的判断：

所谓 Agent Architecture，不是模型能力表，而是控制流设计史。

它在不断回答同一组问题：

• 什么时候该停？
• 什么时候该继续？
• 什么时候该重试？
• 什么时候该换角色？
• 什么时候该查工具？
• 什么时候该调用历史？
• 什么时候该先模拟？
• 什么时候该拒绝？
• 什么时候该让人类接管？

这些问题不依赖某一个框架。无论你底层用什么 DSL，只要开始做真实的 Agent 系统，就一定会亲手长出同一组抽象：

• Workflow.steps 表达确定性边
• Router 表达条件路由
• Loop 表达循环与终止条件
• Parallel 表达并行冗余
• workflow_session_state 表达显式共享状态
• Agent(tools=...) 内置 Tool Loop
• Team(mode="route") 把一整段分诊逻辑改为一个 Team 组合

但控制流本身没有变。这恰恰说明这些问题不是某个框架发明出来的，而是真实系统演化过程中必然会长出来的。

从这个角度看，这 17 种架构并不神秘。它们只是对同一个系统问题的 17 种不同回答。

如果让我把整篇文章压缩成三句话：

1. 先别迷信“万能 Agent”，先把状态和控制流画清楚。
2. 大多数系统从 ReAct 起步，但可靠系统一定会引入验证、记忆和边界控制。
3. 真正高级的 Agent，不是更敢做事，而是更知道什么时候不该做。

一旦你看懂这条路径，后面再看到任何新的“Agent 架构名词”，你都可以先问它三个问题：

• 它新增了什么 State？
• 它新增了什么 Router？
• 它新增了什么 Evaluator？

只要这三个问题答不出来，它大概率就不是一种新架构，只是旧架构换了个名字。

参考

1. github.com/FareedKhan-dev/all-agentic-architectures
2. github.com/agno-agi/agno