Agent 核心 ⭐

Agent 是 AgentScope 2.x 的核心抽象——一个无状态的 Reasoning-Acting 循环引擎，把模型、工具、权限、 HITL、上下文管理、中间件、状态管理与事件系统统一到同一个接口背后。本页是 AgentScope 教程的核心页，后续所有工程化能力（流式、状态会话、中间件、权限）都围绕这里的语义展开。

版本基线：本文以 Python agentscope 2.0.4（PyPI，2026-07-07）为准，要求 Python >=3.11。导入路径为 from agentscope.agent import Agent, ContextConfig, ModelConfig, ReActConfig（源： src/agentscope/agent/__init__.py）。2.x 文档站存在版本化陷阱：非版本化的 /en/ URL 会 404，本页所有官方链接均使用 versions/2.0.4/en/ 形式。

Agent 是什么

2.x 用一个统一的 Agent 类替代了 1.x 的 ReActAgent。它的职责不是"存数据"，而是驱动推理-行动循环：

接收 Msg 或恢复事件作为输入。
调用模型生成推理与可能的工具调用。
若有工具调用，先经 Permission 三态决策（allow / deny / ask），再在 Workspace 中执行。
把工具结果回灌进上下文，回到第 2 步。
直到模型不再产出工具调用，reply() 返回最终 Msg，reply_stream() 产出 ReplyEndEvent。

Agent 本身无状态——会话上下文、权限规则、压缩摘要、当前 reply_id 都显式由 AgentState （来自 agentscope.state）持有。同一个 Agent 实例在多轮 reply 之间之所以"记得"上文，是因为默认自动创建的 AgentState 在内存中累积；要跨进程或跨会话恢复，则需要把 AgentState 持久化到 RedisStorage（见本页 AgentState 一节，详细见状态与会话）。

前端类比

如果你来自前端，可以把 Agent 类比为带生命周期的 React Component：它接收输入（Msg），内部有一个 ReAct 循环（类似 useEffect 触发的副作用链），产出输出（reply / reply_stream）。但有一个关键差异——Agent 自身是无状态的，状态被显式外置到 AgentState。这更接近 Redux 的 "组件 = 纯函数 + Store"模型：Agent 是执行器，AgentState 是 Store。

AgentScope 原生语义：Agent 是事件驱动的 ReAct 循环引擎，不是 React 组件。它没有 props diff、没有 reconciliation、没有 render 阶段；reply 是一次完整的异步执行，reply_stream 是把执行过程拆成结构化事件流。中间件不是 React 的 Hooks，而是包裹整个执行链的洋葱层（类似 Koa）。把前端类比作为入口理解即可，不要把组件生命周期概念硬套到 Agent 上。

ReAct 循环

下面这张流程图说明一次 reply / reply_stream 调用如何穿过 Agent 内部。图前先说明：输入可以是 Msg、恢复事件（UserConfirmResultEvent / ExternalExecutionResultEvent）或 None（仅继续当前状态）；输出在 reply 是最终 Msg，在 reply_stream 是结构化事件流。

节点解释：

observe 注入：输入消息先进上下文。reply / reply_stream 会自动注入；observe() 则只注入不触发后续推理（见 observe）。
Model 调用：Agent 调用配置的 ChatModelBase（如 DashScopeChatModel），生成文本块与可能的工具调用块。
Permission 拦截：工具执行前必经的三态门。ask 会暂停 Agent 并产出 HITL 事件。
工具执行与回灌：被允许的工具在 Workspace 中执行，结果以 ToolResultBlock 回灌进上下文，进入下一轮推理。被拒绝的工具会生成错误结果，模型可以看到并据此调整策略。
max_iters 守卫：ReActConfig 控制最大迭代次数，超过时产出 ExceedMaxItersEvent 终止循环，防止 Agent 陷入无限工具调用。
上下文压缩：当 token 数超过 context_config.trigger_ratio × model.context_length 时自动触发压缩，把旧消息总结并可选卸载到 offloader。

构造 Agent

Agent 的全部构造参数如下（源：src/agentscope/agent/_agent.py 与官方 2.0.4 文档）：

参数	类型	作用
`name`	`str`	Agent 标识，出现在消息与日志中（必填）
`system_prompt`	`str`	基础系统提示词（必填）
`model`	`ChatModelBase`	推理用的 LLM，如 `DashScopeChatModel`（必填）
`toolkit`	`Toolkit \| None`	管理工具、MCP 客户端、Skills 与 ToolGroup；`None` 时无工具
`state`	`AgentState \| None`	持有上下文、权限上下文与会话状态；不传则自动创建
`offloader`	`Offloader \| None`	压缩上下文与大工具结果的卸载目标，需实现 `Offloader` 协议
`middlewares`	`list[MiddlewareBase] \| None`	在 reply、reasoning、acting、model call、system prompt 钩子执行
`model_config`	`ModelConfig`	重试次数与 fallback 模型
`context_config`	`ContextConfig`	上下文压缩阈值与工具结果截断长度
`react_config`	`ReActConfig`	最大迭代次数与拒绝处理策略

最小构造只需要 name、system_prompt、model。凭据通过环境变量注入，不要写死在代码里：

python

import os

from agentscope.agent import Agent
from agentscope.model import DashScopeChatModel
from agentscope.credential import DashScopeCredential

agent = Agent(
    name="my_agent",
    system_prompt="你是一个严谨的中文助手，回答前先确认事实。",
    model=DashScopeChatModel(
        credential=DashScopeCredential(
            api_key=os.environ["DASHSCOPE_API_KEY"],
        ),
        model="qwen-max",
    ),
)

DASHSCOPE_API_KEY 从进程环境读取。如果你用 OpenAI / Anthropic / Gemini，把 model 换成对应的 ChatModelBase 子类即可，Agent 不绑定具体提供商。所选模型必须支持 tool calling，否则装了 toolkit 也无法触发工具调用（见常见错误）。

reply 同步返回

reply 是 async 方法，内部消费所有事件，等 Agent 完成或暂停于外部交互时返回最终 Msg：

python

import asyncio

from agentscope.message import UserMsg

async def main() -> None:
    msg = UserMsg(name="user", content="当前目录下有哪些文件？")
    result = await agent.reply(msg)
    print(result.get_text_content())

asyncio.run(main())

多轮对话复用同一个 Agent 实例即可——默认的 AgentState 会在内存中累积上下文：

python

async def chat() -> None:
    await agent.reply(UserMsg(name="user", content="我叫苏振宇。"))
    # 第二轮：Agent 记得上一轮的内容
    result = await agent.reply(
        UserMsg(name="user", content="我刚才告诉了你什么？"),
    )
    print(result.get_text_content())  # 会提到"苏振宇"

asyncio.run(chat())

⚠️ reply 是协程，必须 await。直接调用 agent.reply(msg) 而不 await 只会返回一个 coroutine 对象，不会执行任何推理。

reply 的 inputs 参数还接受恢复事件（UserConfirmResultEvent / ExternalExecutionResultEvent）用于从 HITL 暂停态恢复，或传 None 仅基于当前状态继续。详见中断与恢复。

reply_stream 流式事件

reply_stream 是 async generator，把执行过程拆成结构化事件逐个产出，适合实时渲染文本、展示工具调用进度、捕获 HITL 暂停：

python

import asyncio

from agentscope.message import UserMsg
from agentscope.event import ReplyStartEvent, ReplyEndEvent, TextBlockDeltaEvent

async def main() -> None:
    msg = UserMsg(name="user", content="总结一下 README。")
    async for event in agent.reply_stream(msg):
        if isinstance(event, ReplyStartEvent):
            print("--- 回复开始 ---")
        elif isinstance(event, TextBlockDeltaEvent):
            print(event.delta, end="", flush=True)
        elif isinstance(event, ReplyEndEvent):
            print("\n--- 回复结束 ---")

asyncio.run(main())

事件类型覆盖整个生命周期，常见的包括：ReplyStartEvent、TextBlockDeltaEvent、 ToolCallStartEvent、ToolResultEndEvent、ReplyEndEvent，以及 HITL 相关的 RequireUserConfirmEvent / RequireExternalExecutionEvent、中断相关的 UserInterruptEvent、迭代超限的 ExceedMaxItiersEvent。

要从事件流重建完整 Msg，通常以 ReplyStartEvent 为起点、用 Msg.append_event(event) 累积增量。这一模式在流式事件页有完整示例，本页不展开。

observe 注入上下文不触发推理

observe(msgs) 把消息塞进 Agent 的上下文，但不触发模型调用。它主要用在多 Agent 场景：一个 Agent 的输出需要被另一个 Agent "看到"，但不希望立即引发回复。

python

# 假设 other_agent 刚产出了 assistant_msg
await agent.observe(other_agent_msg)
# 现在 agent 的上下文里有了 other_agent_msg，但不会自动回复
# 后续调用 reply 时，模型能看到这条消息

在 Agent Team 的 Leader-Worker 协作中，observe 是把 Worker 产出回灌给 Leader 的常用手段。

中断与恢复

Agent 的中断基于 asyncio.CancelledError 实现，可以在模型推理或工具执行的任意阶段停止，上下文保持一致状态，随后可以用新消息恢复。根据 Agent 当前状态，有两种中断方式：

Agent 正在运行——取消正在 await 的 task：

python

import asyncio

from agentscope.agent import Agent
from agentscope.message import UserMsg

async def chat(agent: Agent) -> None:
    async for event in agent.reply_stream(
        UserMsg(name="user", content="..."),
    ):
        ...

async def main() -> None:
    agent = Agent(...)
    task = asyncio.create_task(chat(agent))

    # 随时取消以中断 Agent
    await asyncio.sleep(1)
    task.cancel()

asyncio.run(main())

Agent 处于暂停态（等待 HITL 确认或外部执行）——传入 UserInterruptEvent：

python

from agentscope.event import UserInterruptEvent

async def main() -> None:
    agent = Agent(...)
    # Agent 此前已因 RequireUserConfirmEvent 暂停
    async for event in agent.reply_stream(
        UserInterruptEvent(reply_id=agent.state.reply_id),
    ):
        print(event)

UserInterruptEvent 会丢弃挂起状态、为每个未完成工具调用合成一个标记为"被用户中断"的 ToolResultBlock，依次产出 ToolResultStartEvent / ToolResultTextDeltaEvent / ToolResultEndEvent 与最终 ReplyEndEvent，让 Agent 回到可接收新输入的状态。若 Agent 未处于暂停态，该事件是 no-op，立即返回。

HITL 暂停与恢复

HITL 有两种暂停场景，都通过 reply_stream 检测、通过把结果事件传回 reply / reply_stream 恢复：

RequireUserConfirmEvent：Permission 返回 ask 时触发，等待用户对工具调用做 allow / deny 决策。恢复时构造 UserConfirmResultEvent，其中每个 ConfirmResult 可以接受 suggested_rules 以便后续同类调用自动放行。
RequireExternalExecutionEvent：工具被标记为 is_external_tool=True 时触发，工具逻辑在 Agent 外部执行（人工操作或外部系统），结果包装成 ToolResultBlock 后通过 ExternalExecutionResultEvent 传回。

完整的 HITL 代码模式（接收事件 → 构造结果 → 恢复）在流式事件与状态与会话页展开。本页只需记住：Agent 暂停时 reply 会返回当前已累积的 Msg，reply_stream 会产出对应的 Require 事件，恢复时把结果事件作为 inputs 重新调用即可。

迭代超限

ReActConfig 控制最大迭代次数。当工具调用轮次超过上限时，Agent 产出 ExceedMaxItiersEvent 终止循环，防止陷入无限工具调用。这是兜底机制，正常场景下应通过更精准的 system_prompt 与工具设计避免触发。

AgentState

AgentState（来自 agentscope.state）是一个 Pydantic model，持有 Agent 恢复到原状态所需的全部信息：

会话上下文：消息历史与压缩摘要。
权限上下文：已积累的 permission rules（包括 HITL 中用户接受的 suggested_rules）。
任务上下文：Plan 模式下的任务列表状态。
reply_id：当前回复的标识，用于中断与恢复时定位。

因为它是普通 Pydantic model，可以序列化为 JSON 存到任意后端。内置 RedisStorage（来自 agentscope.app.storage）按 (user_id, agent_id, session_id) 三元组组织状态，暴露两个热路径方法：

方法	作用
`get_session(user_id, agent_id, session_id)`	加载 `SessionRecord`，其 `.state` 即保存的 `AgentState`
`update_session_state(user_id, agent_id, session_id, state)`	回复结束后把更新后的 `AgentState` 写回 Redis

典型用法是：启动时从 RedisStorage 加载状态（找不到则新建 AgentState()），构造 Agent 时传入 state=...，回复结束后写回。完整示例与 upsert_session 的首次创建语义见状态与会话页。

python

import asyncio

from agentscope.agent import Agent
from agentscope.state import AgentState
from agentscope.app.storage import RedisStorage

async def main() -> None:
    async with RedisStorage(host="localhost", port=6379) as storage:
        record = await storage.get_session(
            user_id="u1",
            agent_id="a1",
            session_id="s1",
        )
        state = record.state if record else AgentState()

        agent = Agent(
            name="my_agent",
            system_prompt="...",
            model=...,  # 省略 model 构造
            state=state,
        )

        # ... 执行 reply ...

        await storage.update_session_state(
            user_id="u1",
            agent_id="a1",
            session_id="s1",
            state=agent.state,
        )

asyncio.run(main())

常见错误

误用 agent(msg) 调用：1.x 的 __call__ 接口在 2.x 已移除。2.x 必须用 await agent.reply(msg) 或 async for event in agent.reply_stream(msg)。
⚠️ 1.x 仅作历史对照：1.x 的 ReActAgent、agent(msg) __call__、state_dict / load_state_dict、print 接口在 2.x 全部移除，不能作为 2.x 代码使用。迁移参见 1.x 到 2.x 迁移。
忘记 await：reply 是协程，agent.reply(msg) 不加 await 只返回 coroutine 对象，不会执行任何推理，且会触发 RuntimeWarning: coroutine was never awaited。
模型不支持 tool calling：装了 toolkit 却发现 Agent 永远不调用工具，通常是所选模型不支持 tool calling。AgentScope 的 ChatModelBase 子类要求模型原生支持 function calling。
未装 toolkit 却期望工具调用：toolkit=None 时 Agent 不会暴露任何工具给模型，即使 system_prompt 里写了"请使用 Bash"也不会触发。工具必须显式装配进 Toolkit 再传入 Agent。
把 Agent 当成状态容器：Agent 自身无状态，跨进程恢复靠的是 AgentState 持久化，而不是 Agent 实例本身。直接把 Agent 对象 pickle 或序列化是错误做法。
HITL 恢复时 reply_id 不匹配：恢复事件（UserConfirmResultEvent 等）的 reply_id 必须与暂停时的 agent.state.reply_id 一致，否则无法定位挂起状态。

Agent 核心 ⭐

Agent 是什么

前端类比

ReAct 循环

构造 Agent

reply 同步返回

reply_stream 流式事件

observe 注入上下文不触发推理

中断与恢复

HITL 暂停与恢复

迭代超限

AgentState

常见错误

先修

下一步

官方参考

Agent 核心 ⭐ ​

Agent 是什么 ​

前端类比 ​

ReAct 循环 ​

构造 Agent ​

reply 同步返回 ​

reply_stream 流式事件 ​

observe 注入上下文不触发推理 ​

中断与恢复 ​

HITL 暂停与恢复 ​

迭代超限 ​

AgentState ​

常见错误 ​

先修 ​

下一步 ​

官方参考 ​

Agent 核心 ⭐

Agent 是什么

前端类比

ReAct 循环

构造 Agent

reply 同步返回

reply_stream 流式事件

observe 注入上下文不触发推理

中断与恢复

HITL 暂停与恢复

迭代超限

AgentState

常见错误

先修

下一步

官方参考