Agent 架构

# 单 Agent 循环的简化实现 —— 展示 ReAct 模式的核心逻辑

class SimpleAgent:
"""单 Agent 循环的基本结构"""

def __init__(self, model, tools, max_turns=10):
self.model = model # 大语言模型
self.tools = tools # 可用工具列表
self.max_turns = max_turns # 最大循环轮次，防止无限循环

def run(self, task: str) -> str:
"""执行任务的主循环"""
context = f"用户任务：{task}"

for turn in range(self.max_turns):
# 第一步：思考 —— 让模型决定下一步
response = self.model.think(context)

# 如果模型认为任务完成，返回最终答案
if response.is_final():
return response.content

# 第二步：行动 —— 调用模型选择的工具
tool_name = response.tool_name
tool_args = response.tool_args
tool_result = self.tools[tool_name](**tool_args)

# 第三步：将工具结果反馈给模型，进入下一轮
context += f"\n工具 {tool_name} 返回：{tool_result}"

return "达到最大轮次，任务未完成"

# 使用示例
agent = SimpleAgent(model=llm, tools={
"read_file": read_file,
"search_code": search_code,
"run_test": run_test
})
result = agent.run("修复 runoob 项目中 user.py 的类型错误")

优点

实现最简单，易于调试
适合任务边界清晰的场景
几乎所有 Agent 框架都支持

缺点

上下文窗口容易撑满
复杂任务中容易"偏航"
无法并行处理多个子任务

最佳适用场景：修复一个 bug、编写一个函数、回答一个具体问题。任务明确，复杂度适中，不需要并行或多角色协作。

如果你的任务预计需要超过 15 轮工具调用，单 Agent 循环可能不是最佳选择。考虑使用多 Agent 协作或规划执行架构来分解复杂度。

架构二：规划 + 执行（Plan & Execute）

直觉友好可审查

将"想清楚要做什么"和"实际去做"分离为两个独立阶段，提升任务的可预测性和可审计性。

规划 + 执行架构将 Agent 的工作拆分为两个明确阶段：先规划（Plan），再执行（Execute）。

在规划阶段，模型不执行任何操作，只生成一份详细的执行步骤列表。在执行阶段，系统依次完成每个步骤。这种分离让用户可以在执行前审查计划，就像 Claude Code 的 Plan Mode 一样。

两种变体

变体	行为	典型场景
静态规划	计划一次性生成，按顺序线性执行，不中途调整	流程固定、步骤明确的任务，如数据迁移脚本
动态规划	每执行一步后重新评估，根据结果调整后续计划	结果不确定的任务，如调试、探索性数据分析

动态规划更健壮，但实现复杂度更高，且每步重新规划会消耗额外的 token。

Claude Code 的 Plan Mode 就是这个架构的体现——点击"计划"后，AI 先输出一份详细计划供你审查，确认后才开始执行。这大幅提升了用户的掌控感。

实例

# Plan & Execute 架构的简化实现

class PlanExecuteAgent:
"""先规划、后执行的 Agent"""

def plan(self, task: str) -> list:
"""阶段一：生成执行计划"""
plan = self.model.generate(f"""
请将以下任务拆解为可执行的步骤列表：
任务：{task}
返回 JSON 格式的步骤列表，每步包含：
- step_id: 步骤编号
- description: 步骤描述
- tool: 需要调用的工具名
""")
return plan

def execute(self, plan: list, dynamic: bool = False) -> str:
"""阶段二：逐步执行计划"""
results = []
remaining_plan = plan.copy()

while remaining_plan:
step = remaining_plan.pop(0)
output = self.tools[step["tool"]](step["description"])
results.append({"step": step["step_id"], "output": output})

if dynamic and remaining_plan:
# 动态规划：根据当前结果重新评估后续计划
remaining_plan = self.replan(remaining_plan, results)

return self.summarize(results)

# 使用示例
agent = PlanExecuteAgent()
plan = agent.plan("为 runoob 项目添加用户认证功能")
# 人类可以先审查 plan，确认合理后再执行
result = agent.execute(plan, dynamic=True)

优点

执行前可人工审查计划
推理和执行职责分离，清晰
对长任务友好

缺点

初始计划可能不够准确
两阶段增加了延迟
静态版本难以应对意外情况

Plan & Execute 的代价是增加了推理轮次，对于简单任务反而是一种浪费。如果一个任务可以在 3 步内完成，直接用单 Agent 循环更高效。

架构三：多 Agent 协作（Multi-Agent）

生产推荐复杂任务

一个 Orchestrator（协调者）负责任务拆解和调度，多个 Subagent 各司其职，并行或串行地完成子任务，结果汇聚回 Orchestrator 做综合。

当单个 Agent 面临上下文窗口不足或任务过于复杂时，多 Agent 架构提供了一个优雅的解决方案：让多个专门化的子 Agent 并行工作，由一个 Orchestrator（协调者）统筹全局。

独立上下文是核心优势

每个子 Agent 拥有独立的上下文窗口。代码审查 Agent 深度阅读 auth.py 不会影响性能分析 Agent 的判断；安全检测 Agent 产生的大量中间输出不会挤占其他 Agent 的空间。

子 Agent（Subagent）是短暂的、隔离的——完成一个任务后即销毁。Agent 团队（Agent Teams）则是多个独立 Agent 实例长时间协作、互相发消息，更像一个真实的团队。

实例

# 多 Agent 协作的简化实现

class Orchestrator:
"""编排器：负责任务拆解、分发和结果汇总"""

def __init__(self):
self.subagents = {
"code_review": Subagent(
name="代码审查",
tools=["read_file", "static_analysis"],
system_prompt="你是代码审查专家..."
),
"security": Subagent(
name="安全检测",
tools=["scan_vulnerability", "check_deps"],
system_prompt="你是安全检测专家..."
),
"performance": Subagent(
name="性能分析",
tools=["profile_code", "analyze_complexity"],
system_prompt="你是性能分析专家..."
)
}

def handle_task(self, task: str) -> dict:
# 第一步：分析任务，决定需要哪些 Subagent
needed = self.plan(task)

# 第二步：并行分发（各 Subagent 同时工作，独立上下文）
results = {}
for agent_name in needed:
sub_task = self.decompose(task, agent_name)
results[agent_name] = self.subagents[agent_name].run(sub_task)

# 第三步：汇总各 Subagent 的结果，综合输出
return self.synthesize(task, results)

# 使用示例：一次运行，三个维度并行分析
orch = Orchestrator()
report = orch.handle_task("审查 runoob 项目 PR #42")

优点

天然支持并行，速度快
子 Agent 各自独立，上下文互不干扰
可以专门化每个子 Agent 的角色

缺点

协调逻辑复杂，调试困难
多个 Agent 并行的 Token 成本更高
Orchestrator 本身可能成为瓶颈

多 Agent 协作的主要成本是编排开销。如果子任务非常简单（每个只需 1-2 步），编排开销可能超过实际工作的开销，此时单 Agent 更合适。

架构四：反思与自我修正（Reflection）

高质量输出易于集成

在 Agent 的输出环节加入质量评估，不满意则重新生成或修正，形成内部迭代循环。

反思架构为 Agent 增加了一个"质检环节"：每次生成输出后，都由一个评判者（Critic）来评估质量，如果不达标则要求修正，直到输出满足标准。

这就像开发者写完代码后自己跑一遍测试——在交付之前先自查一遍。

两种实现方式

方式	机制	优点	缺点
自我反思	同一个模型先执行再评估自己的输出	实现简单，无额外模型成本	模型可能对自己的错误"视而不见"
Critic 模型	用独立的评判模型评估执行模型的输出	更客观，能发现执行模型盲区	增加模型调用成本和延迟

"写单元测试 → 运行测试 → 观察失败 → 修复代码 → 再次运行"——这就是反思架构的经典应用。测试结果本身就是 Critic 的反馈信号。

实例

# 反思架构的简化实现

class ReflectiveAgent:
"""带有自我反思能力的 Agent"""

def __init__(self, model, tools, max_reflections=3):
self.model = model
self.tools = tools
self.max_reflections = max_reflections # 最多修正次数，防止死循环

def run(self, task: str) -> str:
# 第一步：正常执行，产生初始输出
output = self.model.generate(task)

for i in range(self.max_reflections):
# 第二步：反思 —— 评估输出质量
critique = self.model.generate(f"""
请严格评估以下输出：
原始任务：{task}
当前输出：{output}
检查：事实错误？逻辑漏洞？遗漏信息？格式问题？
如果输出完美无缺，请回复 "PASS"。
""")

if "PASS" in critique:
break # 输出通过审查

# 第三步：修正 —— 根据批评意见改进
output = self.model.generate(f"""
原始任务：{task}
上次输出：{output}
问题反馈：{critique}
请根据反馈修正输出。
""")

return output

# 使用示例
agent = ReflectiveAgent(model=llm, tools={})
code = agent.run("编写一个 Python 函数，实现 RUNOOB 字符串的 AES 加密")
# Agent 生成代码后自我检查加密实现、密钥处理，
# 发现漏洞后自动修正，确保输出安全可靠

优点

显著提升输出质量
可以设置明确的质量标准
适合有客观评判标准的任务

缺点

多次迭代增加延迟和成本
需要设置最大迭代次数防止死循环
评判标准难以形式化时效果有限

反思循环每次迭代都是一次额外的 LLM 调用，会明显增加延迟。此外需要设置反思次数上限，否则模型可能陷入"永远不满意"的死循环。

架构五：RAG + Agent（检索增强型智能体）

知识密集任务大型知识库

在 Agent 的工具集里加入向量检索能力，让 Agent 在推理过程中动态查询外部知识库，克服上下文窗口的限制。

RAG（Retrieval-Augmented Generation，检索增强生成）本是一种让 LLM 查询外部知识库的技术。当它与 Agent 结合时，变得更加强大：Agent 可以主动决定何时检索、检索什么，而不是每次都被动地检索一次。

与普通 RAG 的关键区别

普通 RAG 是"被动、一次性"的：用户提问时固定检索一次，将结果塞入 Prompt。

RAG + Agent 则不同：Agent 自主判断在推理的哪个环节需要补充知识、需要检索什么，并可以多次查询知识库，直到获得足够的信息来完成任务。

代码库问答 Agent 被问到"为什么这里用单例模式？"时，它会主动检索项目文档、设计决策记录、相关代码文件，而不是只凭 LLM 的训练记忆来猜测。

实例

# RAG + Agent 的简化实现

class RAGAgent:
"""带有动态检索能力的 Agent"""

def __init__(self, model, vector_db, max_retrievals=5):
self.model = model
self.vector_db = vector_db # 向量数据库
self.max_retrievals = max_retrievals

def should_retrieve(self, context: str, question: str) -> bool:
"""Agent 自己判断是否需要检索更多信息"""
decision = self.model.generate(f"""
当前已知信息：{context}
当前问题：{question}
现有信息是否足以回答问题？回答 YES 或 NO。
""")
return "NO" in decision

def run(self, task: str) -> str:
context = ""
retrieval_count = 0

while retrieval_count < self.max_retrievals:
# Agent 自主判断是否需要检索
if not self.should_retrieve(context, task):
break

# Agent 自主决定检索什么
search_query = self.model.generate(f"""
任务：{task}
已有信息：{context}
为了完成任务，下一步应该检索什么信息？
""")

# 执行检索，结果追加到上下文
docs = self.vector_db.search(search_query)
context += "\n".join(docs)
retrieval_count += 1

# 综合所有信息生成最终答案
return self.model.generate(f"任务：{task}\n参考资料：{context}")

# 使用示例
agent = RAGAgent(model=llm, vector_db=runoob_docs_db)
answer = agent.run("RUNOOB 框架中如何配置数据库连接池？")
# Agent 先检索"连接池配置"，发现提到"最大连接数"
# 如果不理解，会再次检索"最大连接数最佳实践"
# 最终综合多轮检索结果给出完整回答

优点

突破上下文窗口限制
输出有据可查，减少幻觉
知识库可独立更新

缺点

检索质量影响整体效果
向量数据库的维护成本
检索延迟增加响应时间

架构六：工作流编排（Workflow / DAG）

生产首选高可靠性

把 Agent 行为固化为一张有向无环图（DAG），每个节点是一个 LLM 调用或工具调用，边表示数据依赖关系，由框架驱动执行。

这是最接近传统软件工程的一种 Agent 架构。与前面几种架构的最大区别在于：Agent 的自主决策空间被限制在单个节点内部，节点之间的流转是预先定义好的，不可更改。

纯 Agent vs DAG 工作流

特性	纯 Agent	DAG 工作流
流程控制	模型自主决定下一步	预定义的 DAG 图决定
可预测性	低，每次运行路径可能不同	高，运行路径确定
可调试性	难，依赖日志追踪	易，每个节点输入输出明确
容错性	依赖模型自行恢复	框架提供重试、断点续跑
灵活性	高，可应对意外情况	低，只能走预定义路径

DAG 的 "无环"（Acyclic）特性意味着工作流是确定性的——没有无限循环，执行路径可以完全预测，失败的节点可以单独重试。

DAG 是"低自主性、高可预测性"的极端。你需要预先设计好整个流程。这不是缺点，而是刻意的设计取舍——生产环境中有时候确定性比灵活性更重要。

实例

# DAG 工作流的简化定义（类似 LangGraph 风格）

from langgraph import StateGraph

# 定义工作流状态 —— 节点间传递的数据对象
class PipelineState:
raw_data: str = "" # 原始输入数据
cleaned_data: str = "" # 清洗后的数据
analysis_result: dict = {} # 分析结果
final_report: str = "" # 最终报告

# 定义 DAG 节点 —— 每个节点是独立的处理单元
def extract_data(state: PipelineState) -> PipelineState:
"""节点1：从 runoob 数据库中提取原始数据"""
state.raw_data = query_database("SELECT * FROM logs")
return state

def clean_data(state: PipelineState) -> PipelineState:
"""节点2：清洗数据（去重、标准化格式）"""
state.cleaned_data = preprocess(state.raw_data)
return state

def analyze_data(state: PipelineState) -> PipelineState:
"""节点3：统计分析"""
state.analysis_result = statistical_analysis(state.cleaned_data)
return state

def generate_report(state: PipelineState) -> PipelineState:
"""节点4：使用 LLM 生成报告"""
state.final_report = llm.generate(
f"基于以下分析结果生成报告：{state.analysis_result}"
)
return state

# 构建 DAG：定义节点和边（数据流向）
workflow = StateGraph(PipelineState)
workflow.add_node("extract", extract_data)
workflow.add_node("clean", clean_data)
workflow.add_node("analyze", analyze_data)
workflow.add_node("report", generate_report)

# 定义边：extract → clean → analyze → report
workflow.add_edge("extract", "clean")
workflow.add_edge("clean", "analyze")
workflow.add_edge("analyze", "report")
workflow.set_entry_point("extract")
workflow.set_finish_point("report")

# 编译并运行
app = workflow.compile()
result = app.invoke(PipelineState())
print(result.final_report)

优点

可预测、可审计、可重试
支持并行加速
工程化程度高，运维友好

缺点

流程需要预先设计，灵活性低
难以应对未预期的情况
需要学习编排框架

横向对比与如何选择

从多个维度对比六种架构，帮助你快速定位适合的选项。

架构	自主性	可预测性	并行能力	适合任务复杂度	典型实现
单 Agent 循环	高	低	无	中低	Claude Code 默认模式
规划 + 执行	中	中	部分	中高	Claude Code Plan Mode
多 Agent 协作	高	低	强	高	AutoGen, CrewAI
反思 / 自我修正	中	中	无	中	Reflexion, Self-Refine
RAG + Agent	高	中	无	中高	LangChain RAG Agent
工作流编排	低	高	强	高（固定流程）	LangGraph, Prefect

常见组合模式

实际生产系统通常会组合使用多种架构，以下是几种成熟的组合模式：

工作流编排 + 多 Agent：用 DAG 定义主流程，每个节点内部是一个独立的 Agent。例如 CI/CD 流水线中，代码检查节点是 Code Review Agent，安全扫描节点是 Security Agent。

多 Agent + RAG：多个 Subagent 共享同一个向量知识库，各自根据子任务按需检索。例如客服系统中，订单查询 Agent 和退款处理 Agent 都查询同一个知识库但检索不同的内容。

规划执行 + 反思：先规划再执行，但每个步骤执行后加入反思环节确保质量。适合对质量要求极高的任务。

如果你不确定从哪里开始，从单 Agent 循环开始。它最容易实现和调试。当你发现上下文窗口不够用时，考虑多 Agent；当你需要质量保证时，加入反思层；当流程趋于稳定时，重构为 DAG 以提升可靠性。

常见误区

误区一：架构越复杂越好

多 Agent 协作看起来很强大，但如果你的任务用单 Agent 循环在 5 步内就能完成，引入编排开销反而降低效率。原则：用能满足需求的最简架构。

误区二：反思一定能提升质量

自我反思的有效性取决于模型的自我评估能力。如果质量要求极其严格，考虑使用 Critic 模型或引入外部验证（如代码自动测试）。

误区三：DAG 工作流不需要 Agent

DAG 定义了流程骨架，但每个节点内部仍然可以是 Agent 调用。工作流编排和 Agent 能力不是互斥的，而是互补的——DAG 提供可靠性，Agent 提供灵活性。

误区四：上下文窗口大了就不需要 RAG

即使模型支持 1M token 的上下文窗口，把所有文档塞进去仍然不是最优方案。RAG 的价值不仅在于"装得下"，更在于精准检索——减少噪音、降低推理成本、提高答案准确性。

总结

六种 Agent 架构覆盖了从高度自主到高度可控的完整光谱。

选择架构时，核心考量只有两个：你需要多大的灵活性来应对意外情况，以及你需要多大的确定性来保证结果可靠。

从简单开始，在确实需要时才增加复杂度，这是 Agent 架构选型的第一原则。

116 次浏览	1次

< 首页

< 下一页