AgentScope 项目学习指南

本文档面向从未接触过 AgentScope 的开发者，帮助你在最短时间内理解项目架构并上手开发。

1. 项目简介

一句话描述

AgentScope 是阿里巴巴通义实验室开源的生产级多智能体（Multi-Agent）应用开发框架，让开发者能在 5 分钟内构建出具备工具调用、记忆、规划、语音等能力的 AI Agent 应用。

核心功能列表

功能	说明
ReAct Agent	内置推理-行动循环的智能体，支持工具调用
多模型支持	支持 DashScope、OpenAI、Anthropic、Gemini、Ollama 等主流模型
工具系统（Toolkit）	注册和调用工具函数，支持代码执行、文件操作、多模态生成
MCP 协议	内置 Model Context Protocol 客户端，可接入任意 MCP 服务器
A2A 协议	Agent-to-Agent 协议，支持跨 Agent 通信与服务发现
记忆系统	短期记忆（内存/Redis/SQLite）+ 长期记忆（Mem0/ReMe）
RAG 知识库	支持多种文档格式读取和向量数据库存储检索
多 Agent 工作流	MsgHub 消息广播 + Pipeline 编排多 Agent 协作
实时语音	实时语音输入输出，支持流式 TTS
可观测性	内置 OpenTelemetry 追踪，可接入 Arize-Phoenix、Langfuse
Hook 系统	在 Agent 生命周期各阶段注入自定义逻辑
Agentic RL	与 Trinity-RFT 集成，支持强化学习微调 Agent

技术栈总览

类别	技术 / 依赖
语言	Python 3.10+
异步框架	`asyncio`（全异步设计）
模型 SDK	`dashscope`、`openai`、`anthropic`、`google-genai`、`ollama`
MCP 协议	`mcp >= 1.13`
A2A 协议	`a2a-sdk`、`nacos-sdk-python >= 3.0.0`
数据库 / 存储	`sqlalchemy`、`redis`、`pymilvus`、`qdrant-client`、`pymongo`
可观测性	`opentelemetry-api >= 1.39.0`、`opentelemetry-sdk`
音频处理	`sounddevice`、`scipy`、`websockets >= 14.0`
数据验证	`pydantic`（通过 `a2a-sdk` 引入）
Token 计数	`tiktoken`（OpenAI）、`transformers`（HuggingFace）
构建工具	`setuptools`、`pyproject.toml`

2. 目录结构说明

agentscope/
├── src/agentscope/              # 核心库源码（主要学习区域）
│   ├── __init__.py              # 库入口：全局配置、init() 函数、模块导出
│   ├── _run_config.py           # 运行时全局配置（run_id、project、trace 开关）
│   ├── _logging.py              # 日志系统配置（基于 Python logging）
│   ├── _version.py              # 版本号定义
│   │
│   ├── agent/                   # Agent 模块（核心）
│   │   ├── _agent_base.py       # AgentBase：所有 Agent 的抽象基类
│   │   ├── _react_agent_base.py # ReActAgentBase：ReAct 算法基类（含 Hook 扩展点）
│   │   ├── _react_agent.py      # ReActAgent：完整实现的 ReAct 智能体
│   │   ├── _user_agent.py       # UserAgent：代表人类用户的 Agent
│   │   ├── _a2a_agent.py        # A2AAgent：支持 A2A 协议的 Agent
│   │   ├── _realtime_agent.py   # RealtimeAgent：实时语音 Agent
│   │   └── _user_input.py       # 用户输入抽象（终端/Studio）
│   │
│   ├── message/                 # 消息模块
│   │   ├── _message_base.py     # Msg：统一消息类（核心数据结构）
│   │   └── _message_block.py    # ContentBlock：多模态内容块定义
│   │
│   ├── model/                   # 模型模块
│   │   ├── _model_base.py       # ChatModelBase：模型抽象基类
│   │   ├── _model_response.py   # ChatResponse：模型响应封装
│   │   ├── _dashscope_model.py  # 阿里云 DashScope 模型
│   │   ├── _openai_model.py     # OpenAI 模型
│   │   ├── _anthropic_model.py  # Anthropic Claude 模型
│   │   ├── _gemini_model.py     # Google Gemini 模型
│   │   ├── _ollama_model.py     # Ollama 本地模型
│   │   └── _trinity_model.py    # Trinity-RFT 训练模型
│   │
│   ├── formatter/               # 格式化器模块
│   │   ├── _formatter_base.py   # FormatterBase：消息格式化抽象基类
│   │   ├── _dashscope_formatter.py  # DashScope 格式化器
│   │   ├── _openai_formatter.py     # OpenAI 格式化器
│   │   ├── _anthropic_formatter.py  # Anthropic 格式化器
│   │   └── ...                  # 其他模型格式化器
│   │
│   ├── tool/                    # 工具模块
│   │   ├── _toolkit.py          # Toolkit：工具注册、调用、中间件管理
│   │   ├── _response.py         # ToolResponse：工具响应封装
│   │   ├── _coding.py           # 内置工具：Python/Shell 代码执行
│   │   ├── _text_file.py        # 内置工具：文本文件读写
│   │   └── _multi_modality.py   # 内置工具：图像/音频生成
│   │
│   ├── memory/                  # 记忆模块
│   │   ├── _working_memory/     # 短期记忆（工作记忆）
│   │   │   ├── _base.py         # MemoryBase：记忆抽象基类
│   │   │   ├── _in_memory_memory.py  # InMemoryMemory：内存存储
│   │   │   ├── _redis_memory.py      # RedisMemory：Redis 存储
│   │   │   └── _sqlalchemy_memory.py # AsyncSQLAlchemyMemory：数据库存储
│   │   └── _long_term_memory/   # 长期记忆
│   │       ├── _mem0/           # Mem0 集成
│   │       └── _reme/           # ReMe 集成
│   │
│   ├── pipeline/                # 工作流编排模块
│   │   ├── _msghub.py           # MsgHub：多 Agent 消息广播中心
│   │   ├── _class.py            # SequentialPipeline / FanoutPipeline
│   │   ├── _functional.py       # sequential_pipeline / fanout_pipeline 函数
│   │   └── _chat_room.py        # ChatRoom：聊天室抽象
│   │
│   ├── mcp/                     # MCP 协议客户端模块
│   │   ├── _client_base.py      # MCPClientBase：MCP 客户端基类
│   │   ├── _http_stateless_client.py  # 无状态 HTTP 客户端
│   │   ├── _http_stateful_client.py   # 有状态 HTTP 客户端
│   │   └── _stdio_stateful_client.py  # Stdio 客户端（本地进程）
│   │
│   ├── a2a/                     # A2A 协议模块
│   │   ├── _base.py             # AgentCardResolverBase：Agent 名片解析基类
│   │   ├── _file_resolver.py    # 文件方式解析 Agent 名片
│   │   ├── _well_known_resolver.py  # .well-known 方式解析
│   │   └── _nacos_resolver.py   # Nacos 注册中心方式解析
│   │
│   ├── rag/                     # RAG 知识库模块
│   │   ├── _reader.py           # 文档读取器（PDF/Word/Excel/PPT）
│   │   ├── _store.py            # 向量数据库存储（Qdrant/Milvus/MongoDB 等）
│   │   ├── _knowledge_base.py   # KnowledgeBase：知识库管理
│   │   └── _document.py         # Document：文档数据模型
│   │
│   ├── tracing/                 # 可观测性追踪模块
│   │   ├── _setup.py            # setup_tracing：初始化 OTel 追踪
│   │   ├── _trace.py            # 追踪装饰器（trace_llm / trace_reply 等）
│   │   └── _extractor.py        # 追踪数据提取器
│   │
│   ├── plan/                    # 规划模块（任务分解）
│   │   ├── _plan_model.py       # SubTask / Plan：任务和子任务数据模型
│   │   ├── _plan_notebook.py    # PlanNotebook：管理计划并在推理时注入提示
│   │   └── _in_memory_storage.py # InMemoryPlanStorage：内存计划存储
│   │
│   ├── session/                 # 会话持久化模块（保存/恢复 Agent 状态）
│   │   ├── _session_base.py     # SessionBase：会话抽象基类（save/load_session_state）
│   │   ├── _json_session.py     # JSONSession：JSON 文件持久化
│   │   └── _redis_session.py    # RedisSession：Redis 持久化
│   │
│   ├── embedding/               # 向量嵌入模块（DashScope/OpenAI/Gemini/Ollama）
│   ├── token/                   # Token 计数模块（OpenAI/Anthropic/Gemini/HuggingFace）
│   ├── evaluate/                # 评估模块（基于 Ray 的并行评估）
│   ├── realtime/                # 实时语音模块（WebSocket 双向流）
│   ├── tuner/                   # RL 微调模块（与 Trinity-RFT 集成）
│   ├── types/                   # 公共类型定义（JSONSerializableObject 等）
│   └── module/                  # 状态模块基类（StateModule，供 Session 持久化使用）
│
├── examples/                    # 示例代码（最佳学习资源）
│   ├── agent/                   # Agent 示例（react_agent、a2a_agent、voice_agent、
│   │                            #   deep_research_agent、browser_agent、
│   │                            #   meta_planner_agent、realtime_voice_agent、a2ui_agent）
│   ├── functionality/           # 功能示例（MCP、RAG、记忆、TTS 等）
│   ├── workflows/               # 多 Agent 工作流示例
│   ├── game/                    # 游戏示例（狼人杀）
│   ├── evaluation/              # 评估示例
│   └── tuner/                   # RL 微调示例
│
├── tests/                       # 单元测试
├── docs/                        # 文档（NEWS、roadmap、changelog）
├── pyproject.toml               # 项目配置与依赖声明
└── README.md                    # 项目主页

3. 架构设计

整体架构模式与设计理念

AgentScope 采用分层 + 组合的架构模式，核心设计理念是：

全异步（async-first）：所有 Agent 调用链均基于 asyncio，支持高并发多 Agent 场景
组合优于继承：Agent 通过组合 Model、Formatter、Memory、Toolkit 等组件来获得能力，而非深层继承
模型无关性：通过 Formatter 层将 Msg 对象转换为各模型 API 所需格式，业务代码无需关心底层模型差异
可扩展 Hook 系统：在 Agent 生命周期的关键节点（reply/reasoning/acting/print/observe）注入自定义逻辑，无需修改核心代码

核心模块划分与职责

┌─────────────────────────────────────────────────────────────┐
│                        用户代码 / 示例                        │
└──────────────────────────┬──────────────────────────────────┘
                           │ 调用
┌──────────────────────────▼──────────────────────────────────┐
│                     Agent 层（agent/）                        │
│  AgentBase → ReActAgentBase → ReActAgent                     │
│  UserAgent / A2AAgent / RealtimeAgent                        │
└──┬──────────┬──────────┬──────────┬──────────┬──────────────┘
   │          │          │          │          │
   ▼          ▼          ▼          ▼          ▼
┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────────┐
│Model │  │Fmt   │  │Memory│  │Toolkit│  │RAG/Plan  │
│模型层│  │格式化│  │记忆层│  │工具层 │  │知识/规划 │
└──┬───┘  └──┬───┘  └──────┘  └──┬───┘  └──────────┘
   │         │                    │
   ▼         ▼                    ▼
┌──────────────────┐         ┌──────────┐
│  LLM API 服务    │         │ MCP 服务 │
│(DashScope/OpenAI)│         │  服务器  │
└──────────────────┘         └──────────┘

┌─────────────────────────────────────────────────────────────┐
│                   Pipeline 层（pipeline/）                    │
│  MsgHub（消息广播）/ SequentialPipeline / FanoutPipeline     │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                   基础设施层                                   │
│  Tracing（OTel）/ Logging / Session / Embedding / Token      │
└─────────────────────────────────────────────────────────────┘

模块间调用关系与数据流向

单次 Agent 调用的数据流：

用户输入 Msg
    │
    ▼
agent(msg)  →  __call__()  →  reply()
                                │
                    ┌───────────┼───────────┐
                    ▼           ▼           ▼
              memory.add()  RAG检索    长期记忆检索
                    │
                    ▼
              formatter.format(msgs)   ← 将 Msg 列表转换为模型 API 格式
                    │
                    ▼
              model(prompt, tools=...)  ← 调用 LLM API
                    │
                    ▼
              解析响应（文本 / 工具调用）
                    │
            ┌───────┴────────┐
            ▼                ▼
        纯文本响应        工具调用块（ToolUseBlock）
            │                │
            │          toolkit.call_tool_function()
            │                │
            │          工具执行结果（ToolResponse）
            │                │
            └───────┬────────┘
                    ▼
              memory.add(结果)
                    │
                    ▼
              广播给 MsgHub 订阅者
                    │
                    ▼
              返回 Msg 给调用方

4. 核心流程解析

4.1 应用启动流程

# src/agentscope/__init__.py

# 1. 模块加载时，创建全局配置对象（使用 ContextVar 保证线程/协程安全）
_config = _ConfigCls(
    run_id=ContextVar("run_id", default=shortuuid.uuid()),
    project=ContextVar("project", default="UnnamedProject_..."),
    ...
)

# 2. 导入所有子模块（agent、model、tool、memory 等）

# 3. 用户调用 agentscope.init() 进行可选初始化
def init(project, name, run_id, logging_path, studio_url, tracing_url):
    # 设置项目名、运行名、日志
    setup_logger(logging_level, logging_path)
    
    # 如果提供 studio_url，向 Studio 注册本次运行
    requests.post(f"{studio_url}/trpc/registerRun", json=data)
    
    # 如果提供 tracing_url，启动 OpenTelemetry 追踪
    setup_tracing(endpoint=endpoint)
    _config.trace_enabled = True

注意：agentscope.init() 是可选的，不调用也能正常使用框架，只是没有日志文件和追踪功能。

4.2 ReAct Agent 推理-行动循环

ReAct（Reasoning + Acting）是最核心的 Agent 模式，其循环逻辑在 src/agentscope/agent/_react_agent.py 中：

# 简化版核心循环（reply 方法）
async def reply(self, msg=None, ...):
    # 1. 将输入消息存入记忆
    await self.memory.add(msg)
    
    # 2. 从长期记忆检索相关信息
    await self._retrieve_from_long_term_memory(msg)
    
    # 3. 从 RAG 知识库检索相关文档
    await self._retrieve_from_knowledge(msg)
    
    # 4. 按需压缩记忆（防止 token 超限）
    await self._compress_memory_if_needed()
    
    # 5. 主循环：推理 → 行动 → 推理 → ...
    for _ in range(self.max_iters):
        # 推理：调用 LLM 生成响应
        msg_reasoning = await self._reasoning(tool_choice)
        
        # 如果没有工具调用，直接返回文本响应
        if not msg_reasoning.has_content_blocks("tool_use"):
            reply_msg = msg_reasoning
            break
        
        # 行动：执行所有工具调用
        for tool_call in msg_reasoning.get_content_blocks("tool_use"):
            structured_output = await self._acting(tool_call)
    
    return reply_msg

推理阶段（_reasoning）：

async def _reasoning(self, tool_choice=None):
    # 将记忆中的消息格式化为模型 API 所需格式
    prompt = await self.formatter.format(
        msgs=[
            Msg("system", self.sys_prompt, "system"),
            *await self.memory.get_memory(),
        ]
    )
    
    # 调用 LLM（支持流式输出）
    res = await self.model(
        prompt,
        tools=self.toolkit.get_json_schemas(),  # 传入工具描述
        tool_choice=tool_choice,
    )
    
    # 流式打印输出
    async for content_chunk in res:
        msg.content = content_chunk.content
        await self.print(msg, last=False)
    
    await self.print(msg, last=True)
    await self.memory.add(msg)  # 存入记忆
    return msg

行动阶段（_acting）：

async def _acting(self, tool_call):
    # 执行工具函数
    tool_res = await self.toolkit.call_tool_function(tool_call)
    
    # 流式处理工具结果
    async for chunk in tool_res:
        tool_res_msg.content[0]["output"] = chunk.content
        await self.print(tool_res_msg, chunk.is_last)
    
    # 工具结果存入记忆
    await self.memory.add(tool_res_msg)

4.3 多 Agent 消息广播（MsgHub）

MsgHub 是多 Agent 协作的核心机制，使用 Python 异步上下文管理器：

# src/agentscope/pipeline/_msghub.py

async with MsgHub(
    participants=[agent1, agent2, agent3],
    announcement=Msg("Host", "开始讨论", "assistant")
) as hub:
    # 进入时：
    # 1. 为每个 agent 设置订阅者列表（其他所有 agent）
    # 2. 广播 announcement 消息给所有参与者
    
    await agent1()  # agent1 回复后，自动广播给 agent2、agent3
    await agent2()  # agent2 回复后，自动广播给 agent1、agent3
    
    # 动态管理参与者
    hub.add(agent4)
    hub.delete(agent3)
    
    # 退出时：清除所有订阅关系

广播机制原理：每个 Agent 内部维护 _subscribers 字典，当 reply() 完成后，__call__ 方法自动调用 _broadcast_to_subscribers()，将回复消息发送给所有订阅者的 observe() 方法（即存入其记忆）。

4.4 MCP 工具接入流程

MCP（Model Context Protocol）允许 Agent 调用外部服务提供的工具：

from agentscope.mcp import HttpStatelessClient
from agentscope.tool import Toolkit

# 1. 创建 MCP 客户端
client = HttpStatelessClient(
    name="gaode_mcp",
    transport="streamable_http",
    url=f"https://mcp.amap.com/mcp?key={API_KEY}",
)

# 2. 方式一：将 MCP 工具注册到 Toolkit
toolkit = Toolkit()
await toolkit.register_mcp_client(client)

# 2. 方式二：获取单个工具作为可调用函数
func = await client.get_callable_function(func_name="maps_geo")
await func(address="天安门广场", city="北京")

# 3. 将 Toolkit 传给 Agent，Agent 会自动在推理时传入工具描述
agent = ReActAgent(..., toolkit=toolkit)

4.5 消息（Msg）的生命周期

Msg 是框架中最核心的数据结构，贯穿整个调用链：

# src/agentscope/message/_message_base.py

msg = Msg(
    name="Friday",           # 发送者名称
    content=[                # 内容：字符串 或 ContentBlock 列表
        TextBlock(type="text", text="你好！"),
        ImageBlock(type="image", source={"type": "url", "url": "..."}),
    ],
    role="assistant",        # 角色：user / assistant / system
    metadata={},             # 元数据：结构化输出等
)

# 常用方法
msg.get_text_content()                    # 获取纯文本内容
msg.get_content_blocks("tool_use")        # 获取特定类型的内容块
msg.has_content_blocks("tool_result")     # 检查是否包含某类型块
msg.to_dict()                             # 序列化为字典
Msg.from_dict(json_data)                  # 从字典反序列化

5. 关键设计与实现

5.1 设计模式

模式	应用场景	代码位置
模板方法模式	`AgentBase` 定义 `reply()`/`observe()` 抽象接口，子类实现具体逻辑	`agent/_agent_base.py`
策略模式	Formatter 层针对不同模型实现不同的消息格式化策略	`formatter/`
观察者模式	MsgHub 通过订阅者列表实现消息广播	`pipeline/_msghub.py`
装饰器模式	`@trace_llm`、`@trace_reply` 等追踪装饰器无侵入地添加可观测性	`tracing/_trace.py`
中间件模式（洋葱模型）	Toolkit 的 `register_middleware()` 支持工具调用前后的拦截处理	`tool/_toolkit.py`
工厂 / 组合模式	`ReActAgent` 通过构造函数组合 Model、Formatter、Memory、Toolkit	`agent/_react_agent.py`
元类（Metaclass）	`_ReActAgentMeta` 自动为 `_reasoning`/`_acting` 方法注入 Hook 包装	`agent/_agent_meta.py`

5.2 数据模型

消息内容块（ContentBlock），定义在 src/agentscope/message/_message_block.py：

# 所有内容块均为 TypedDict（类型化字典），而非类实例
TextBlock      = {"type": "text", "text": str}
ThinkingBlock  = {"type": "thinking", "thinking": str}
ToolUseBlock   = {"type": "tool_use", "id": str, "name": str, "input": dict}
ToolResultBlock = {"type": "tool_result", "id": str, "name": str, "output": ...}
ImageBlock     = {"type": "image", "source": {"type": "url"|"base64", ...}}
AudioBlock     = {"type": "audio", "source": {...}}
VideoBlock     = {"type": "video", "source": {...}}

记忆存储：短期记忆（工作记忆）支持三种后端：

InMemoryMemory：Python 列表，进程内存储，重启丢失
RedisMemory：Redis 存储，支持跨进程共享
AsyncSQLAlchemyMemory：数据库存储（SQLite/PostgreSQL 等），持久化

5.3 全局状态管理

AgentScope 使用 Python contextvars.ContextVar 管理全局运行配置，这是一种协程安全的上下文变量（不同 asyncio Task 之间隔离）：

# src/agentscope/_run_config.py
_config = _ConfigCls(
    run_id=ContextVar("run_id", default=...),    # 本次运行 ID
    project=ContextVar("project", default=...),  # 项目名
    trace_enabled=ContextVar("trace_enabled", default=False),
)

这意味着在同一进程中可以并发运行多个独立的 Agent 系统，各自拥有独立的配置。

5.4 错误处理与日志策略

日志：使用 Python 标准 logging 模块，日志名为 "as"：

# src/agentscope/_logging.py
logger = logging.getLogger("as")
# 格式：时间 | 级别 | 模块:函数:行号 - 消息

工具调用错误处理：Toolkit.call_tool_function() 捕获所有异常并将错误信息包装为 ToolResponse 返回给 LLM，而不是抛出异常中断流程：

except mcp.shared.exceptions.McpError as e:
    res = ToolResponse(content=[TextBlock(text=f"Error: {e}")])
except Exception as e:
    res = ToolResponse(content=[TextBlock(text=f"Error: {e}")])

用户中断处理：Agent 支持实时中断（asyncio.CancelledError），handle_interrupt() 方法处理中断后的清理逻辑，并向 LLM 补充假的工具结果以保持对话连贯性。

5.5 可观测性（Tracing）

基于 OpenTelemetry 标准，通过装饰器无侵入地追踪关键操作：

@trace_llm      # 追踪 LLM 调用（输入/输出/token 用量）
@trace_reply    # 追踪 Agent reply 调用
@trace_toolkit  # 追踪工具调用
@trace_format   # 追踪消息格式化
@trace_embedding # 追踪向量嵌入

追踪数据可发送到 AgentScope Studio 或第三方平台（Arize-Phoenix、Langfuse）。

6. 环境搭建与运行

环境依赖

Python 3.10 或更高版本（必须）
pip 或 uv（包管理器）

安装步骤

方式一：从 PyPI 安装（推荐新手）

pip install agentscope

方式二：从源码安装（推荐开发者）

# 克隆仓库
git clone -b main https://github.com/agentscope-ai/agentscope.git
cd agentscope

# 安装可编辑模式（修改源码立即生效）
pip install -e .

# 或使用 uv（更快）
uv pip install -e .

安装可选依赖

# 安装所有功能（开发推荐）
pip install -e ".[full]"

# 按需安装特定功能
pip install -e ".[a2a]"        # A2A 协议支持
pip install -e ".[realtime]"   # 实时语音支持
pip install -e ".[rag]"        # RAG 知识库支持
pip install -e ".[gemini]"     # Google Gemini 模型
pip install -e ".[ollama]"     # Ollama 本地模型
pip install -e ".[dev]"        # 开发工具（测试、文档等）

快速运行第一个示例

import os
import asyncio
from agentscope.agent import ReActAgent, UserAgent
from agentscope.model import DashScopeChatModel
from agentscope.formatter import DashScopeChatFormatter
from agentscope.memory import InMemoryMemory
from agentscope.tool import Toolkit, execute_python_code

async def main():
    # 1. 创建工具集
    toolkit = Toolkit()
    toolkit.register_tool_function(execute_python_code)

    # 2. 创建 AI Agent
    # ReActAgent 完整参数说明（均有默认值，按需传入）：
    # name            - Agent 名称（必填）
    # sys_prompt      - 系统提示词（必填）
    # model           - 聊天模型实例（必填）
    # formatter       - 消息格式化器，需与 model 匹配（必填）
    # toolkit         - 工具集，默认创建空 Toolkit
    # memory          - 短期记忆，默认 InMemoryMemory
    # long_term_memory - 长期记忆（Mem0/ReMe），默认 None
    # long_term_memory_mode - 长期记忆控制模式：
    #                   "agent_control"（工具调用）/"static_control"（自动）/"both"
    # knowledge       - RAG 知识库，默认 None
    # enable_rewrite_query - 是否在 RAG 检索前重写查询，默认 True
    # plan_notebook   - 任务规划笔记本，默认 None
    # enable_meta_tool - 是否注册 reset_equipped_tools 元工具，默认 False
    # parallel_tool_calls - 是否并行执行多个工具调用，默认 False
    # print_hint_msg  - 是否打印系统提示消息，默认 False
    # max_iters       - ReAct 循环最大迭代次数，默认 10
    # tts_model       - TTS 语音合成模型，默认 None
    # compression_config - 记忆压缩配置，默认 None（不压缩）
    agent = ReActAgent(
        name="Friday",
        sys_prompt="你是一个有帮助的助手。",
        model=DashScopeChatModel(
            model_name="qwen-max",
            api_key=os.environ["DASHSCOPE_API_KEY"],
            stream=True,
        ),
        memory=InMemoryMemory(),
        formatter=DashScopeChatFormatter(),
        toolkit=toolkit,
    )

    # 3. 创建用户 Agent（代表人类输入）
    user = UserAgent(name="user")

    # 4. 开始对话循环
    msg = None
    while True:
        msg = await agent(msg)
        msg = await user(msg)
        if msg.get_text_content() == "exit":
            break

asyncio.run(main())

关键环境变量

变量名	说明	示例
`DASHSCOPE_API_KEY`	阿里云 DashScope API 密钥	`sk-xxx`
`OPENAI_API_KEY`	OpenAI API 密钥	`sk-xxx`
`ANTHROPIC_API_KEY`	Anthropic API 密钥	`sk-ant-xxx`
`GOOGLE_API_KEY`	Google Gemini API 密钥	`AIzaXxx`
`GAODE_API_KEY`	高德地图 MCP 服务密钥（示例用）	`xxx`
`AGENTSCOPE_DISABLE_CONSOLE_OUTPUT`	设为 `true` 禁用控制台输出	`true`

7. 推荐学习路线

阶段一：入门（1-2 天）

目标：理解框架的基本概念和最简用法

读 README.md：了解项目定位和核心功能
读 src/agentscope/message/_message_base.py：理解 Msg 这个核心数据结构
读 src/agentscope/__init__.py：了解模块组织和 init() 函数
跑 examples/agent/react_agent/：运行第一个 ReAct Agent 示例
读 src/agentscope/agent/_agent_base.py（重点看 __call__、reply、observe、print 方法）

理解的核心概念：

Msg（消息）是什么，包含哪些字段
AgentBase 的生命周期：__call__ → reply → print → 广播
role 的三种取值（user/assistant/system）的含义

阶段二：进阶（3-5 天）

目标：掌握核心机制，能够定制 Agent 行为

读 src/agentscope/agent/_react_agent.py：深入理解 ReAct 循环（reply → _reasoning → _acting）
读 src/agentscope/tool/_toolkit.py：理解工具注册、调用、中间件机制
读 src/agentscope/pipeline/_msghub.py：理解多 Agent 消息广播
读 src/agentscope/memory/_working_memory/_base.py：理解记忆抽象接口
跑 examples/workflows/multiagent_conversation/：体验多 Agent 协作
跑 examples/functionality/mcp/：体验 MCP 工具接入
读 src/agentscope/formatter/_formatter_base.py：理解为什么需要 Formatter 层

理解的核心机制：

ReAct 循环如何在推理和行动之间切换
Formatter 如何将 Msg 列表转换为不同模型的 API 格式
MsgHub 的订阅者机制如何实现自动广播
Hook 系统如何在不修改核心代码的情况下扩展行为

阶段三：实践（持续）

目标：通过动手验证理解，具备独立开发能力

建议尝试的任务：

自定义工具：编写一个查询天气的工具函数，注册到 Toolkit 并让 Agent 使用

def get_weather(city: str) -> ToolResponse:
    """查询指定城市的天气"""
    # 实现天气查询逻辑
    return ToolResponse(content=[TextBlock(type="text", text=f"{city}今天晴天")])

toolkit.register_tool_function(get_weather)

自定义 Hook：注册一个 post_reply Hook，记录每次 Agent 回复的 token 用量

def log_reply_hook(agent, kwargs, output):
    print(f"Agent {agent.name} 回复完成")

ReActAgent.register_class_hook("post_reply", "log_reply", log_reply_hook)

多 Agent 辩论：参考 examples/workflows/multiagent_debate/，实现一个三个 Agent 轮流辩论的场景
接入 RAG：参考 examples/functionality/rag/，为 Agent 添加本地文档知识库
实现自定义 Agent：继承 ReActAgentBase，重写 _reasoning 和 _acting 方法，实现一个有特殊行为的 Agent

8. 常见问题与注意事项

容易踩的坑

1. 忘记 await

所有 Agent 调用都是异步的，必须在 async 函数中使用 await：

# ❌ 错误
msg = agent(msg)

# ✅ 正确
msg = await agent(msg)

2. 混淆 content 的两种格式

Msg.content 可以是字符串或 ContentBlock 列表，使用时注意区分：

# 字符串格式（简单文本）
msg = Msg("user", "你好", "user")

# 块格式（多模态或工具调用）
msg = Msg("user", [TextBlock(type="text", text="你好")], "user")

# 安全获取文本内容（两种格式都兼容）
text = msg.get_text_content()

3. Formatter 与 Model 必须匹配

不同模型的 API 格式不同，必须使用对应的 Formatter：

# ✅ 正确配对
model = DashScopeChatModel(...)
formatter = DashScopeChatFormatter()

# ❌ 错误：Formatter 和 Model 不匹配
model = DashScopeChatModel(...)
formatter = OpenAIChatFormatter()  # 会导致 API 请求格式错误

4. MCP 有状态客户端需要先连接

# 有状态客户端（StdIOStatefulClient / HttpStatefulClient）需要先 connect
client = StdIOStatefulClient(...)
await client.connect()  # 不能省略！
await toolkit.register_mcp_client(client)

5. Hook 是类级别共享的

register_class_hook 注册的 Hook 对所有实例生效，测试时注意清理：

# 测试后记得清理
ReActAgent.clear_class_hooks()

代码中的特殊约定

文件命名：私有模块以单下划线开头（_agent_base.py），公开 API 通过 __init__.py 导出
异步生成器：工具函数可以返回 AsyncGenerator[ToolResponse, None] 实现流式输出
_MemoryMark：记忆中的消息可以打标记（HINT/COMPRESSED），用于过滤和管理
ToolResponse 的 is_last 字段：流式工具响应中，is_last=True 表示最后一个 chunk
metadata 字段：Msg.metadata 用于传递结构化输出（如 Pydantic 模型验证后的数据）

值得注意的 TODO / 技术债务

以下是源码中标注的 TODO，反映了当前的技术债务：

tool/_toolkit.py：A2A 相关的 Card Resolver 依赖需要从 a2a 依赖中拆分出来
```
# TODO: split the card resolvers from the a2a dependency
```
agent/_react_agent.py：记忆压缩时多模态内容块的处理方式尚未确定
```
# TODO: What if the compressed messages include multimodal blocks?
```
agent/_react_agent.py：最大迭代次数耗尽时，结构化输出的强制生成逻辑待完善
```
# TODO: handle the structured output here, maybe force calling the finish_function here
```
agent/_agent_base.py：终端中多模态内容块（图片/视频/音频）的显示方式待设计
```
# TODO: We should consider how to handle the multimodal blocks in the terminal
```
pyproject.toml：pypdf 版本被临时固定（<=6.5.0），因为更新版本存在解析问题

附录：快速参考

常用导入

# 核心类
from agentscope.agent import ReActAgent, UserAgent, AgentBase
from agentscope.message import Msg, TextBlock, ToolUseBlock, ImageBlock
from agentscope.model import DashScopeChatModel, OpenAIChatModel
from agentscope.formatter import DashScopeChatFormatter, OpenAIChatFormatter
from agentscope.memory import InMemoryMemory, RedisMemory
from agentscope.tool import Toolkit, execute_python_code, execute_shell_command
from agentscope.pipeline import MsgHub, sequential_pipeline
from agentscope.mcp import HttpStatelessClient, StdIOStatefulClient

# 初始化（可选）
import agentscope
agentscope.init(project="MyProject", logging_level="DEBUG")

支持的模型一览

模型类	对应服务	推荐 Formatter
`DashScopeChatModel`	阿里云通义	`DashScopeChatFormatter`
`OpenAIChatModel`	OpenAI / 兼容接口	`OpenAIChatFormatter`
`AnthropicChatModel`	Anthropic Claude	`AnthropicChatFormatter`
`GeminiChatModel`	Google Gemini	`GeminiChatFormatter`
`OllamaChatModel`	Ollama 本地模型	`OllamaChatFormatter`