Wang Jianhua
返回

AgentScope 2.0 架构与使用指南

更新于:

Table of contents

Open Table of contents

从最小 Agent 开始

AgentScope 2.0 要求 Python 3.11+：

uv pip install agentscope

最小 Agent 由 Model、Toolkit、System Prompt 组成。官方 Quickstart 同时展示了两种输出接口：reply() 返回完整消息，reply_stream() 逐个产生事件。

import asyncio
import os

from agentscope.agent import Agent
from agentscope.credential import DashScopeCredential
from agentscope.event import EventType
from agentscope.message import UserMsg
from agentscope.model import DashScopeChatModel
from agentscope.tool import Toolkit, Bash, Read, Write, Edit

async def main() -> None:
    agent = Agent(
        name="Friday",
        system_prompt="You are a helpful assistant named Friday.",
        model=DashScopeChatModel(
            credential=DashScopeCredential(
                api_key=os.getenv("DASHSCOPE_API_KEY"),
            ),
            model="qwen-plus",
        ),
        toolkit=Toolkit(
            tools=[Bash(), Read(), Write(), Edit()],
        ),
    )

    user_msg = UserMsg(
        name="user",
        content="检查当前项目并运行测试。",
    )

    async for event in agent.reply_stream(user_msg):
        match event.type:
            case EventType.TEXT_BLOCK_DELTA:
                print(event.delta, end="", flush=True)
            case EventType.TOOL_CALL_START:
                print(f"\ntool: {event.tool_call_name}")

asyncio.run(main())

Agent 不是一个只调用一次模型的包装器。一次 reply 可以包含多轮模型调用、工具执行、权限确认和上下文压缩，直到产生最终消息或进入暂停状态。

Message 与 Event

Message 是持久化和 Agent 间通信的单位；Event 是流式 UI 与 HITL 的单位。一组 reply 事件最终可以还原成一条 AssistantMsg。

Message 由 Block 组成

from agentscope.message import (
    UserMsg,
    TextBlock,
    DataBlock,
    URLSource,
)

msg = UserMsg(
    name="user",
    content=[
        TextBlock(text="分析这张图片"),
        DataBlock(
            name="architecture",
            source=URLSource(
                url="https://example.com/architecture.png",
                media_type="image/png",
            ),
        ),
    ],
)

常用 Block 包括 TextBlock、ThinkingBlock、DataBlock、ToolCallBlock、ToolResultBlock 和 HintBlock。工具调用与结果使用同一个 ID 配对。

用 Event 重建 Message

from agentscope.event import ReplyStartEvent
from agentscope.message import AssistantMsg

reply_msg = None

async for event in agent.reply_stream(user_msg):
    if isinstance(event, ReplyStartEvent):
        reply_msg = AssistantMsg(
            name=event.name,
            content=[],
            id=event.reply_id,
        )
    elif reply_msg is not None:
        reply_msg.append_event(event)

事件按 start → delta → end 组织。文本和数据使用 block_id，工具调用与结果使用 tool_call_id，整次回复共享 reply_id。

Agent：显式状态的 ReAct 引擎

官方文档将 Agent 定义为 stateless reasoning-acting loop engine。这里的 stateless 指 Agent 不负责持久化；可恢复状态集中在显式传入的 AgentState 中。

四个公开入口

from agentscope.message import UserMsg

result = await agent.reply(
    UserMsg(
        name="user",
        content="列出当前目录中的 Python 文件。",
    ),
)

print(result.get_text_content())

reply() 与 reply_stream() 接受相同输入：新的 Msg、恢复 HITL 的结果事件，或用于继续当前状态的 None。

AgentState 是恢复点

源码：state/_state.py

class AgentState(BaseModel):
    session_id: str = Field(default_factory=_generate_id)
    summary: str | list[TextBlock | DataBlock] = ""
    context: list[Msg] = Field(default_factory=list)
    reply_id: str = Field(default_factory=_generate_id)
    cur_iter: int = 0
    permission_context: PermissionContext = Field(
        default_factory=PermissionContext,
    )
    tool_context: ToolContext = Field(default_factory=ToolContext)

Context、压缩摘要、当前 reply、权限规则和工具缓存都在 State 中。需要跨请求恢复时，序列化 agent.state，下一轮重新注入。

Toolkit：Tool、MCP 与 Skill 的统一入口

Toolkit 负责注册、生成 Schema 和分发调用。内置 Tool、Python 函数、MCP Tool 与 Skill 最终都从 Toolkit 进入 Agent。

内置工具

from agentscope.tool import (
    Toolkit,
    Bash,
    Read,
    Write,
    Edit,
    Glob,
    Grep,
)

toolkit = Toolkit(
    tools=[
        Bash(),
        Read(),
        Write(),
        Edit(),
        Glob(),
        Grep(),
    ],
)

Tool 的运行时属性会影响调度和权限：is_concurrency_safe 决定能否并发，is_read_only 参与权限判断，is_external_tool 将执行交给外部系统。

将 Python 函数变成 Tool

from agentscope.tool import FunctionTool, Toolkit

def get_weather(
    city: str,
    unit: str = "celsius",
) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is 22°{unit[0].upper()}"

toolkit = Toolkit(
    tools=[
        FunctionTool(
            get_weather,
            is_concurrency_safe=True,
            is_read_only=True,
        ),
    ],
)

FunctionTool 从函数名、docstring 和类型标注生成 Tool Schema。需要自定义权限或流式结果时，再继承 ToolBase。

注册 MCP

from agentscope.mcp import MCPClient, HttpMCPConfig
from agentscope.tool import Toolkit

search_mcp = MCPClient(
    name="search",
    is_stateful=False,
    mcp_config=HttpMCPConfig(
        url="https://api.search.com/mcp",
    ),
    enable_tools=["web_search", "image_search"],
)

toolkit = Toolkit(mcps=[search_mcp])

MCP Tool 使用 mcp__{server_name}__{tool_name} 命名。Stateful Client 需要显式连接和关闭；Stateless HTTP Client 每次调用创建临时会话。

注册 Skill

from agentscope.tool import Toolkit

toolkit = Toolkit(
    tools=[Read(), Grep()],
    skills_or_loaders=[
        "./skills/code-review",
        "./skills/release",
    ],
)

Toolkit 只把 Skill 的名称和描述写入 System Prompt。Agent 需要时调用内置 Skill 查看工具读取完整 SKILL.md，再使用已有 Tool 执行步骤。

Context：压缩、截断与卸载

每次模型调用的输入由 System Prompt、压缩摘要和近期 Context 组成。System Prompt 还会附加 Skill 索引，并经过 on_system_prompt Middleware。

ContextConfig

from agentscope.agent import ContextConfig

agent = Agent(
    name="coder",
    system_prompt="You are a coding agent.",
    model=model,
    toolkit=toolkit,
    context_config=ContextConfig(
        trigger_ratio=0.8,
        reserve_ratio=0.1,
        tool_result_limit=3000,
    ),
)

超过 trigger_ratio × model.context_length 后，Agent 总结旧消息并保留近期消息。单个工具结果超过 tool_result_limit 时先截断。

Offloader 保留被移出的内容

from agentscope.workspace import LocalWorkspace

workspace = LocalWorkspace(
    workdir="/tmp/agent-workspace",
)
await workspace.initialize()

agent = Agent(
    name="coder",
    system_prompt="You are a coding agent.",
    model=model,
    toolkit=toolkit,
    offloader=workspace,
)

配置 Offloader 后，压缩消息写入 sessions/{session_id}/context.jsonl，完整工具结果写入 tool_result-{tool_id}.txt。Context 中只保留可按需读取的路径。

Permission 与 Middleware

Permission 决定一次 Tool Call 是否允许、拒绝或询问用户；Middleware 在 Agent 生命周期的固定位置插入横切逻辑。

PermissionContext

from agentscope.permission import (
    PermissionContext,
    PermissionMode,
)
from agentscope.state import AgentState

state = AgentState(
    permission_context=PermissionContext(
        mode=PermissionMode.EXPLORE,
    ),
)

agent = Agent(
    name="explorer",
    system_prompt="Inspect the project without changing files.",
    model=model,
    toolkit=toolkit,
    state=state,
)

常用 Mode：

Rule 优先于 Mode；Tool 的 check_permissions() 和 check_read_only() 根据真实参数执行运行时检查。

Middleware Hook

from datetime import datetime
from agentscope.middleware import MiddlewareBase

class RuntimeContextMiddleware(MiddlewareBase):
    async def on_system_prompt(
        self,
        agent,
        current_prompt,
    ):
        now = datetime.now().isoformat()
        return (
            f"{current_prompt}\n\n"
            f"## Runtime\nCurrent time: {now}"
        )

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    toolkit=toolkit,
    middlewares=[RuntimeContextMiddleware()],
)

Agent Middleware 提供 on_reply、on_reasoning、on_acting、on_model_call、on_compress_context 和 on_system_prompt。Onion Hook 包裹执行，Transformer Hook 按顺序改写 Prompt。

Workspace：执行环境与资源生命周期

Workspace 同时提供 Tool/MCP/Skill、Context Offloader 和资源生命周期。它不是单纯的工作目录。

from agentscope.agent import Agent
from agentscope.tool import Toolkit
from agentscope.workspace import LocalWorkspace

workspace = LocalWorkspace(
    workdir="./my-workspace",
    skill_paths=["./skills/code-review"],
)
await workspace.initialize()

agent = Agent(
    name="coder",
    system_prompt="You are a coding assistant.",
    model=model,
    toolkit=Toolkit(
        tools=await workspace.list_tools(),
        mcps=await workspace.list_mcps(),
        skills_or_loaders=await workspace.list_skills(),
    ),
    offloader=workspace,
)

AgentScope 提供 LocalWorkspace、DockerWorkspace、E2BWorkspace 和 OpenSandboxWorkspace。四者实现相同接口，但安全边界不同：Local 使用宿主机，另外三种提供独立运行环境。

Workspace Manager 用于 Agent Service。它根据 workspace_id 分配、缓存和回收 Workspace，并决定不同用户、Agent、Session 是否共享执行环境。

Agent Service：把 Agent 放进服务

Agent SDK 只负责运行一次 reply。Agent Service 增加 FastAPI、租户、Session、Storage、MessageBus、Workspace Manager、后台任务与调度。

定义与状态分离

服务端的 Agent 记录保存名称、System Prompt、Context 与 ReAct 配置；Session 记录保存模型绑定、Workspace 和可变的 AgentState。

AgentData
  name
  system_prompt
  context_config
  react_config

SessionRecord
  user_id
  agent_id
  config
  state: AgentState

同一个 Agent 可以拥有多条 Session。每轮请求读取 Session State，组装新的 Agent 执行对象，结束后再将 State 写回 Storage。

启动服务

官方示例通过 create_app() 组合基础设施：

import os
import uvicorn

from agentscope.app import create_app
from agentscope.app.message_bus import InMemoryMessageBus
from agentscope.app.storage import RedisStorage
from agentscope.app.workspace_manager import LocalWorkspaceManager

storage = RedisStorage(
    host="localhost",
    port=6379,
)

app = create_app(
    storage=storage,
    message_bus=InMemoryMessageBus(),
    workspace_manager=LocalWorkspaceManager(
        basedir=os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            "workspaces",
        ),
    ),
)

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
    )

InMemoryMessageBus 适合单进程示例。多进程部署使用 RedisMessageBus，让 Session Lock、事件回放、Inbox 和 Wakeup 跨进程共享。

Chat 与事件流分离

POST /chat
  → ChatRunRegistry.spawn
  → ChatService.run
  → Session distributed lock
  → Agent.reply_stream
  → MessageBus replay log + Pub/Sub
  → GET /sessions/{id}/stream
  → persist Msg + AgentState

POST /chat 只返回任务已启动；模型文本、工具调用、权限请求和后台结果由 Session SSE 发送。断线重连时先读取 replay log，再接入实时 Pub/Sub。

Agent Team：多 Agent 仍然是多 Session

Agent Team 建在 Agent Service 之上。Leader 和 Worker 都是独立 Session，各自拥有 State、Workspace 绑定和事件流。

Leader Session
  ├─ TeamCreate
  ├─ AgentCreate → Worker Session
  ├─ TeamSay     → recipient inbox
  └─ TeamDelete

Worker Session
  └─ TeamSay

Team 消息先作为 HintBlock 写入接收者 Inbox，再发送 Wakeup。任意服务进程都可以领取 Wakeup 并运行目标 Session，因此 Worker 不是 Leader 进程内嵌套的协程。

需要不同角色时，使用 SubAgentTemplate 固定 Prompt 和权限：

from agentscope.app import (
    SubAgentTemplate,
    create_app,
)
from agentscope.permission import (
    PermissionContext,
    PermissionMode,
)

app = create_app(
    storage=storage,
    message_bus=message_bus,
    workspace_manager=workspace_manager,
    custom_subagent_templates=[
        SubAgentTemplate(
            type="explorer",
            description="Read-only codebase exploration.",
            system_prompt_template=(
                "You are {member_name}, an explorer in "
                "team '{team_name}' led by {leader_name}. "
                "Report results with TeamSay."
            ),
            permission_context=PermissionContext(
                mode=PermissionMode.EXPLORE,
            ),
        ),
    ],
)

两条使用路径

直接使用 SDK：

Agent
  + Model
  + Toolkit
  + AgentState
  + Middleware
  + Workspace

适合脚本、单进程服务和自定义编排。调用方负责保存 State、管理 Workspace 生命周期并消费 Event。

使用 Agent Service：

FastAPI
  + Agent records
  + Sessions
  + Storage
  + MessageBus
  + Workspace Manager
  + Scheduler / Background tasks / Agent Team

适合多用户、多会话、事件流和分布式部署。两条路径使用同一个 Agent，区别在于 Loop 外围的基础设施由谁负责。

参考资料

上一篇
Agent 系统实现：Loop、工具、上下文与多 Agent
下一篇
Python 项目 AGENTS.md：开发基线与工具说明