饮墨

子安饮墨馀三斗,留与卿儿作赋来

4 种主流 AI Agent 架构模式详解:从 ReAct 到 Multi-Agent,附可运行代码

2026 年,AI Agent 已经不是概念了——它在生产环境里跑着自动化运维、客服、数据分析。但很多团队一上来就 "调 API + 写 Prompt",没想清楚架构,结果上线后翻车:幻觉失控、循环调用烧 Token、多步任务半路断掉。本文拆解 4 种经过验证的 Agent 架构模式,每种都给出 Python 可运行示例。


痛点:为什么 "裸调 LLM" 撑不住复杂任务

直接用 LLM 做 Agent 最大的问题:

  1. 单次推理无规划——让模型 "先想后做" 全靠 Prompt 祈祷
  2. 工具调用失控——没有循环终止条件,Token 费用爆炸
  3. 多步骤状态丢失——上下文窗口不够时,前面的中间结果被截断
  4. 多模型协作没章法——谁负责规划、谁负责执行、谁做质检,全靠 if-else 硬编码

这些问题的根源是缺少架构设计。下面 4 种模式,从简到繁,按需选用。


模式 1:ReAct(Reasoning + Acting)

核心思想: 模型在每一步先 "思考"(Thought),再 "行动"(Action),最后 "观察"(Observation)结果,循环直到任务完成。

适用场景: 单步工具调用、简单问答增强、运维巡检 Agent。

import openai, json

tools = [
    {
        "type": "function",
        "function": {
            "name": "run_shell",
            "description": "在服务器上执行 shell 命令并返回输出",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "要执行的 shell 命令"}
                },
                "required": ["command"]
            }
        }
    }
]

def run_shell(command: str) -> str:
    """实际执行时替换为 subprocess.run,此处模拟"""
    import subprocess
    result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
    return result.stdout[:2000] or result.stderr[:2000]

def react_agent(query: str, max_steps: int = 5):
    messages = [
        {"role": "system", "content": "你是一个运维 Agent。用工具获取信息,逐步推理后给出最终答案。"},
        {"role": "user", "content": query}
    ]

    for step in range(max_steps):
        resp = openai.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:  # 没有工具调用 → 模型认为可以输出最终答案
            return msg.content

        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            output = run_shell(args["command"])
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": output
            })

    return "达到最大步数限制,请检查任务复杂度。"

# 示例:排查磁盘使用
# print(react_agent("检查哪个目录占用磁盘空间最大"))

关键设计: max_steps 是保险丝——防止模型陷入无限循环。生产环境建议加上 Token 消耗计数器做双重限制。


模式 2:Plan-and-Execute(规划-执行分离)

核心思想: 先用一个 "Planner" 生成完整计划(步骤列表),再由 "Executor" 逐步执行。规划和执行解耦,支持计划修正。

适用场景: 多步运维流程(部署、扩缩容)、需要审批节点的工作流。

def plan_and_execute(task: str):
    # Step 1: Planner 生成计划
    plan_prompt = f"""你是运维规划 Agent。将以下任务分解为 3-7 个具体步骤,每步一行,格式:
1. [步骤描述] — 工具: [shell命令或"无需工具"]

任务: {task}"""

    plan_resp = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": plan_prompt}]
    )
    plan = plan_resp.choices[0].message.content
    print(f"📋 计划:\n{plan}\n")

    # Step 2: Executor 逐步执行
    steps = [line.strip() for line in plan.split("\n") if line.strip() and line.strip()[0].isdigit()]
    results = []

    for i, step in enumerate(steps):
        print(f"▶ 执行步骤 {i+1}: {step}")
        exec_resp = openai.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "执行以下步骤,如需 shell 命令则调用工具。"},
                {"role": "user", "content": f"步骤: {step}\n已完成步骤的结果: {json.dumps(results[-3:], ensure_ascii=False)}"}
            ],
            tools=tools,
            tool_choice="auto"
        )
        # ... 处理工具调用(同 ReAct 模式)
        results.append({"step": step, "result": exec_resp.choices[0].message.content})

    return results

与 ReAct 的区别: ReAct 是 "走一步看一步",Plan-and-Execute 是 "先画蓝图再施工"。对于运维场景,后者更安全——可以在执行前人工审核计划。


模式 3:Tool-Use Router(工具路由模式)

核心思想: Agent 不自己执行,而是作为 "调度中心",根据意图把请求路由到不同的专业工具链。

适用场景: 统一运维入口、ChatOps、多系统集成。

TOOL_REGISTRY = {
    "k8s_ops": {
        "description": "Kubernetes 集群操作:Pod 状态、日志、扩缩容",
        "handler": lambda params: run_shell(f"kubectl {params['kubectl_args']}"),
    },
    "db_query": {
        "description": "数据库慢查询分析",
        "handler": lambda params: run_shell(
            f"psql -h {params['host']} -U monitor -c \"SELECT * FROM pg_stat_activity WHERE state='active' ORDER BY duration DESC LIMIT 5;\""
        ),
    },
    "alert_silence": {
        "description": "静默 Alertmanager 告警",
        "handler": lambda params: run_shell(
            f"amtool silence add alertname={params['alert_name']} --duration={params.get('duration','2h')} --comment='Agent auto-silence'"
        ),
    },
}

def router_agent(user_input: str):
    # 让 LLM 选择工具和参数
    routing_prompt = f"""根据用户请求,选择最合适的工具并提取参数。
可用工具: {json.dumps({k: v['description'] for k, v in TOOL_REGISTRY.items()}, ensure_ascii=False)}

用户: {user_input}

返回 JSON: {{"tool": "工具名", "params": {{...}}}}"""

    resp = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": routing_prompt}],
        response_format={"type": "json_object"}
    )

    route = json.loads(resp.choices[0].message.content)
    tool = TOOL_REGISTRY.get(route["tool"])
    if not tool:
        return f"未知工具: {route['tool']}"

    return tool["handler"](route["params"])

优势: 新增能力只需往 TOOL_REGISTRY 里注册,Agent 逻辑不用改。这就是 OpenClaw 等平台的 Skills 机制的底层思路。


模式 4:Multi-Agent(多智能体协作)

核心思想: 多个专职 Agent 各司其职,通过消息传递协作。典型角色:Planner、Researcher、Executor、Reviewer。

适用场景: 复杂故障排查、安全审计、需要 "四眼原则" 的场景。

from dataclasses import dataclass
from typing import Literal

@dataclass
class AgentMessage:
    sender: str
    content: str
    msg_type: Literal["task", "result", "review", "approved", "rejected"]

def specialist_agent(name: str, system_prompt: str, task: str, context: list[str] = None) -> str:
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": f"任务: {task}\n\n上下文:\n" + "\n".join(context or [])}
    ]
    resp = openai.chat.completions.create(model="gpt-4o", messages=messages)
    return resp.choices[0].message.content

def multi_agent_diagnose(issue: str):
    # Agent 1: Diagnostician — 分析可能原因
    diagnosis = specialist_agent(
        "Diagnostician",
        "你是故障诊断专家。分析问题的可能原因,列出排查步骤。",
        issue
    )
    print(f"🔍 诊断结果:\n{diagnosis}\n")

    # Agent 2: Executor — 执行排查命令
    exec_result = specialist_agent(
        "Executor",
        "你是运维执行者。根据诊断建议执行排查,返回关键数据。",
        f"执行以下诊断步骤: {diagnosis}",
        context=[f"原始问题: {issue}"]
    )
    print(f"⚙️ 执行结果:\n{exec_result}\n")

    # Agent 3: Reviewer — 审核结论
    review = specialist_agent(
        "Reviewer",
        "你是运维审核员。检查诊断和执行结果,确认根因是否找到,给出最终建议。",
        f"审核以下排查过程",
        context=[f"问题: {issue}", f"诊断: {diagnosis}", f"执行: {exec_result}"]
    )
    print(f"✅ 审核结论:\n{review}")

    return review

生产建议: Multi-Agent 的通信开销大,建议用消息队列(Redis Stream / NATS)解耦,每个 Agent 独立部署,支持横向扩展。


避坑指南

坑 1:Agent 陷入死循环

ReAct 模式下模型可能反复调用同一工具。解法:max_steps + 检测连续相同工具调用(超过 2 次强制终止)。

坑 2:Plan-and-Execute 的计划过于理想

Planner 生成的步骤可能不可执行(比如假设了不存在的命令)。解法: 每步执行后检查返回码,失败时回传 Planner 重新规划(replan 机制)。

坑 3:Multi-Agent 的 Token 成本失控

3 个 Agent 协作一次,Token 消耗是单 Agent 的 5-8 倍。解法: 内层 Agent 用轻量模型(GPT-4o-mini / Claude Haiku),只有 Reviewer 用强模型。


总结:怎么选?

模式 复杂度 适用场景 Token 成本
ReAct 单步工具调用、简单巡检
Plan-and-Execute ⭐⭐ 多步流程、需要审批
Tool-Use Router ⭐⭐ ChatOps、统一运维入口 低-中
Multi-Agent ⭐⭐⭐ 复杂诊断、安全审计

经验法则: 能用 ReAct 搞定的别上 Multi-Agent。架构的目的是解决问题,不是炫技。先跑通最简方案,再按需升级。

您还没有登录,请登录后发表评论。