大数跨境
首页
>
AI Agent 的核心架构到底是什么？从 ReAct 到状态机，一次讲透
>

AI Agent 的核心架构到底是什么？从 ReAct 到状态机，一次讲透

AI Agent 的核心架构到底是什么？从 ReAct 到状态机，一次讲透 智能体AI
2026-10-17
9
导读：工作流（Workflow）还是智能体（Agent）？企业级大模型应用选型指南

Agent 不只是 LLM 加几个 Tool——真正让它跑得起来的，是一整套运行时架构。

近两年，AI Agent 热度攀升。许多开发者起步于“大模型+工具+Prompt”的简单组合，处理简单查询尚可，但面对复杂任务（如多维度销售数据分析）时，常因数据库超时、数据不全或模型循环调用而陷入困境。

此时，单纯依赖大模型生成答案已无法解决问题。工程核心在于构建一套能够持续决策、执行动作、获取反馈、处理异常并明确终止条件的运行时系统。

「这才是 AI Agent 的工程核心。Agent 不只是 LLM 加几个 Tool，它还需要一套负责执行、控制、状态管理和错误恢复的运行时架构。」

本文将从最小 Agent 入手，拆解这套架构的核心要素。

本文看点

02

Think → Act → Observe 内核循环

05

五道防线让 Agent 不失控

10

用 Python 从零搭出最小 Agent
01

DEFINITION

概念辨析：什么才算真正的 Agent？

在讨论架构前，需厘普通大模型应用、Workflow 与 Agent 的区别：

1. 普通大模型应用

流程为：用户输入 → Prompt → LLM → 返回答案。适用于文本总结、翻译等单次调用场景。若涉及多步查询与分析，单次调用显然不足。

2. Workflow（工作流）

由程序预先定义执行步骤（如：查数据→算环比→识异常→查流失→生报告）。优势在于路径清晰、易测试审计。若业务流程稳定，无需过度智能化。

3. Agent（智能体）

核心区别在于动态选择下一步行动。Agent 根据当前状态和已获信息决定后续动作（如发现华东销售下降，转而查询该区域订单取消率）。执行路径随结果动态变化，非完全预设。

三者关系：企业级应用可采用“Workflow 外层控制 + Agent 动态判断环节”的组合模式，兼顾可控性与灵活性。

4. Agent 的最小组成

从工程实现角度，Agent 包含五个基本部分：

  • LLM：负责决策，判断下一步行动。
  • Tools：负责执行，包括查询、API 调用、计算等。
  • State：负责保存状态，记录任务、历史、结果、进度及预算。
  • Loop：负责推进，连接决策、执行与反馈，支持多轮迭代。
  • Stop：负责终止，在任务完成、预算耗尽或不可恢复错误时停止。
02

CORE LOOP

Agent 内核：Think、Act、Observe 循环

Agent 的核心执行模式可概括为：Think（决策）→ Act（执行）→ Observe（观察）。

1. Think：决策下一步

模型接收任务、可用工具、已查数据、历史动作及剩余预算等上下文，决定下一步行动。例如：“暂无销售数据，先查询本月及上月各区域销售额。”注意：模型仅返回调用意图，未实际执行。

2. Act：程序执行动作

运行时解析模型返回的工具调用请求（如 JSON 格式），查找对应工具实现，校验参数后调用实际接口（数据库或 API）。执行者是程序而非模型。执行完成后，工具返回真实结果（如：华东区销售额下降 18%）。

3. Observe：反馈结果

运行时更新 Agent 状态，将工具结果提供给模型。模型基于新状态继续决策（如：检查订单取消率或客户流失率）。如此循环，直至任务完成。

4. 为何不能写成一个大函数？

随着功能增加，单体函数难以维护模型失败、工具超时、预算溢出、用户取消及断点恢复等问题。合理做法是将 Agent 视为有明确状态和转换规则的运行时系统，即引入状态机。
03

STATE

State 与状态机：超越聊天记录的持久化

聊天记录仅是 State 的一部分。Agent 还需知晓：已执行工具及结果、当前阶段、步数、Token 消耗、子任务状态及错误信息。

1. State 结构设计

简化状态结构示例：

...python

from dataclasses import dataclass, field

from typing import Any


@dataclass

class AgentState:

task_id: str

messages: list[dict[str, Any]] = field(default_factory=list)

step_count: int = 0

token_used: int = 0

status: str = "thinking"

last_error: str | None = None

实际系统还需保存工具记录、计划、检查点等。需注意上下文裁剪策略，避免 Token 无限增长，同时保留关键业务事实。

2. 状态机的价值

状态机明确定义了允许做什么及下一步状态。最小 Agent 可包含四种状态：Thinking（决策中）、Acting（执行中）、Done（完成）、Error（错误）。

状态转换规则示例：Done 状态不应继续执行工具；Error 状态不能无条件回到正常路径；Acting 状态超时需根据错误类型决定重试或终止。复杂场景可引入 WaitingApproval（待审批）、Paused（暂停）等状态。

注意：状态机不能替代权限校验。高风险操作需在执行层再次检查授权。
04

TOOL CALLING

ReAct 与 Function Calling：模型如何调用工具？

模型与工具通信主要有两种思路：

1. ReAct：约定格式

模型输出约定格式的动作信息（如 Action: query_sales），运行时解析并执行。优点是与模型无关，缺点是文本格式不稳定，需严格的解析、校验和错误处理。

2. Function Calling：结构化数据

通过向模型提供工具定义（Schema），模型生成结构化调用请求。运行时负责校验工具存在性、参数类型、权限，并执行工具。Function Calling 仅结构化意图，执行、安全校验和错误恢复仍属运行时职责。

3. 选型建议

生产系统优先使用原生 Function Calling，因其更适合参数校验和管理。ReAct 适用于不支持原生调用的场景。无论哪种方式，核心都是“获取意图→校验→执行→保存结果→继续决策”的循环。

4. 统一工具接口

为便于管理，工具应具备统一接口。Python 可使用 Protocol 定义：

...python

from typing import Any, Protocol


class Tool(Protocol):

name: str

description: str

def execute(self, arguments: dict[str, Any]) -> Any: ...
05

CONTROL

控制机制：防止 Agent 失控的五道防线

1. 限制最大执行步数 (MaxSteps)

设置单次任务最大步数（如 10 步），超过即停止。防止模型因无法推进任务而无限循环。

2. 控制 Token 预算

设置累计 Token 上限。每次调用后更新用量，达到上限即停止。需依据模型接口返回的实际用量统计，多模型环境需考虑成本差异。

3. 设置执行超时

包括单次工具/模型请求超时及任务总期限。Python 可使用 asyncio.wait_for() 等机制。注意：超时取消不一定能立即停止底层操作。

4. 检测重复动作

生成动作签名（如 tool_name:args_json），检测连续相同调用。若超过阈值，触发风险控制。注意区分数据变化导致的合理重复。

5. 限制错误自愈次数 (MaxHealAttempts)

工具返回参数错误时，允许模型修正，但需限制重试次数。区分网络超时（适合退避重试）、参数非法（适合模型修正）和权限不足（不应重试）。

「停止条件是 Agent 核心架构的一部分，而非锦上添花。」
06

ERROR HANDLING

错误处理：构建可靠的自愈机制

1. 错误分类

  • 输入/参数错误：反馈给模型修正，多次失败则停止。
  • 瞬时错误（如网络中断）：采用有限重试及指数退避。
  • 不可恢复错误（如权限不足、预算耗尽）：立即停止，避免风险扩大。

2. 自愈实现

将错误信息写入历史，让模型决定修正或确认。重新执行前必须再次校验参数。对于创建订单等敏感操作，禁止因超时而直接重试扣款。

3. 高风险动作的控制

企业 Agent 与普通聊天机器人不同，错误可能导致实际业务后果（如重复下单、越权修改）。权限检查、幂等控制、事务和审批机制不能仅靠 Prompt，需由确定性程序控制：

模型提出动作 → 运行时检查权限 → 校验参数/规则 → 检查幂等/审批 → 执行操作 → 记录审计日志

4. 完整记录

保留错误类型、工具、参数摘要、重试次数及修复结果，以便分析故障根源。
07

OBSERVABILITY

可观测性：拒绝黑盒，事件流与 Trace

良好的工程体验要求 Agent 执行过程可见。可通过定义基础事件（tool_call, tool_result, answer_delta, error, done）实现。

1. 事件流解耦

使用异步生成器逐步输出事件，前端通过 SSE 或 WebSocket 接收。用户端展示进度，开发者端记录详细日志（耗时、错误码、Token 用量），并注意敏感数据脱敏。

2. 执行轨迹 (Trace)

串联模型调用、工具执行、状态转换和结果，形成完整 Trace。借此分析任务成功率、平均时长、单位成本及错误恢复效果，评估是否适合生产环境。
08

PLANNING

任务规划：Plan-and-Execute 与 DAG

对于多步骤任务，ReAct 的动态执行可能增加延迟和成本。引入 Plan-and-Execute 模式：先生成计划，再执行。

1. DAG 与并行执行

使用有向无环图 (DAG) 描述任务依赖。独立子任务（如查销售额、查流失率）可并行执行，汇总任务需等待前置完成，提升效率。

2. 计划验证

模型生成计划后，运行时需验证：ID 唯一性、依赖存在性、无循环依赖、工具合法性及预算合规性。模型负责提出，程序负责验证。

3. 混合模式

结合 ReAct 与 Planning：生成初步计划 → 执行子任务 → 观察结果 → 判断计划有效性 → 有效则继续，失效则重新规划。兼顾效率与灵活性。
09

PERSISTENCE

持久化：从一次性程序到可恢复长期任务

长周期任务需应对服务重启或人工审批等待。内存状态不足，需引入 Checkpoint（检查点）。

1. Checkpoint 机制

关键操作后保存状态快照，包括任务 ID、消息历史、已完成工具、计划、预算及时间戳。可使用 JSON 文件或数据库存储，并通过 Store 接口抽象实现。

2. 恢复执行的复杂性

工具调用可能产生外部副作用（如创建订单）。若进程在保存状态前崩溃，恢复时直接重放会导致重复操作。解决方案：

  • 使用幂等键：业务系统识别同一操作。
  • 状态核对：恢复前检查外部系统真实状态。
  • 事务机制与人工确认：针对高风险操作。
10

IMPLEMENTATION

实战：用 Python 组织最小 Agent 代码

整合前述机制，构建模块化 Python Agent。

1. 项目结构

分为 agent/（执行控制）、tools/（工具定义）、storage/（持久化）、tests/（测试）。模块化便于职责分离。

2. 统一工具接口与注册表

定义 Tool Protocol，实现 CalculatorTool 等具体工具。建立 ToolRegistry 字典，通过名称查找工具实现。注意：注册不等于授权，执行前需校验权限。

3. 状态与预算控制

定义 AgentState 数据类保存进度。定义 AgentBudget 检查步数和 Token 上限。

4. 执行循环

封装 LLMClient 适配层，统一不同模型的响应格式。主循环逻辑：检查预算 → 请求模型 → 判断类型（最终答案/工具调用） → 执行工具/返回结果 → 更新状态 → 下一轮。

...python

def run_agent(state, llm, registry, budget):

while True:

budget.check(state)

response = llm.complete(messages=state.messages, tools=...)

if response["type"] == "final":

state.status = "done"

return response["content"]

# 执行工具调用逻辑...

5. 持久化与测试

实现 save_state/load_state 方法。编写单元测试覆盖：普通问答、正常工具调用、参数错误、超时、重复调用、预算溢出、状态恢复、非法工具及重复副作用等场景。测试不仅关注最终答案，更要验证中间执行过程的正确性。
∞

THE END

结语：企业级 Agent 的核心竞争力

大模型仅是 Agent 系统的一部分。真正决定其能否稳定工作的，是状态机、统一工具接口、预算控制、错误处理、事件流、Checkpoint 及幂等机制等工程设施。

企业应用关注的是：数据准确性、权限合规、故障处理、证据可追溯及适时停止。架构应随业务复杂度演进：简单问答用模型，固定流程用 Workflow，动态判断用 Agent，复杂并行引入规划。

大模型决定了系统能做出何种判断，而运行时架构决定了这些判断能否转化为可靠、可控、可恢复的实际行动。这是从 Demo 走向生产系统的关键跨越。


END

【声明】内容源于网络
0
0
智能体AI
1234
内容 521
粉丝 0
智能体AI 1234
总阅读23.3k
粉丝0
内容521