AgentScope 代码阅读与开发指南

---

目录

1. 快速上手：5 分钟跑起来
2. 项目结构速览
3. 源码阅读路线图
4. 核心抽象与设计模式
5. 测试体系详解
6. 开发工作流
7. 常见任务速查
8. 调试技巧

---

1. 快速上手：5 分钟跑起来

1.1 环境准备

# 要求 Python ≥ 3.11
python --version

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

# 创建虚拟环境（推荐使用 uv，也可以用 venv / conda）
uv venv
source .venv/bin/activate

# 安装开发模式（含全部依赖）
uv pip install -e ".[dev]"

# 安装 pre-commit hooks
pre-commit install

1.2 验证安装

# 确认导入正常
python -c "import agentscope; print(agentscope.__version__)"

# 运行最小测试集
pytest tests/agent_basic_test.py -x -v

1.3 启动服务示例

# 需要先启动 Redis
# 然后运行完整的服务示例
python examples/agent_service/main.py

---

2. 项目结构速览

agentscope/
├── src/agentscope/                  # 核心源码
│   ├── __init__.py                  # 包入口：logger + __version__
│   ├── _version.py                  # 版本号
│   ├── _logging.py                  # 结构化日志
│   ├── _utils/                      # 内部工具 (JSON修复, 音频, mixin)
│   ├── agent/                       # ★ Agent 核心
│   │   ├── _agent.py                #   统一 Agent 类 (~96KB)
│   │   └── _config.py               #   Pydantic 配置模型
│   ├── model/                       # ★ 模型抽象
│   │   ├── _base.py                 #   ChatModelBase 基类
│   │   ├── _anthropic/              #   各提供商实现
│   │   ├── _dashscope/
│   │   ├── _gemini/
│   │   ├── _deepseek/
│   │   ├── _openai_chat.py
│   │   └── ...
│   ├── formatter/                   # ★ 消息格式化器（Msg → API 格式）
│   ├── message/                     # ★ 消息类型体系
│   │   ├── _base.py                 #   Msg, UserMsg, AssistantMsg...
│   │   └── _block.py                #   TextBlock, ToolCallBlock...
│   ├── event/                       # ★ 事件系统（30+ 事件类型）
│   ├── middleware/                   # 中间件管线
│   │   ├── _base.py
│   │   ├── _budget.py               #   Token 预算控制
│   │   ├── _longterm_memory/        #   mem0 长期记忆
│   │   ├── _tracing.py              #   OpenTelemetry 追踪
│   │   └── _tts_middleware.py       #   文本转语音
│   ├── tool/                        # ★ 工具系统
│   │   ├── _base.py                 #   ToolBase 基类
│   │   ├── _toolkit.py              #   Toolkit 工具集
│   │   ├── _adapters.py             #   MCPTool, FunctionTool
│   │   ├── _builtin.py              #   内置工具 (Bash, Edit, Read...)
│   │   └── _task.py                 #   任务管理工具
│   ├── permission/                  # 权限引擎
│   ├── workspace/                   # ★ 沙箱执行环境
│   │   ├── _base.py                 #   WorkspaceBase
│   │   ├── _local_workspace.py      #   本地子进程
│   │   ├── _docker/                 #   Docker 容器
│   │   └── _e2b.py                  #   E2B 云沙箱
│   ├── app/                         # ★ FastAPI 服务层
│   │   ├── _app.py                  #   create_app() 工厂
│   │   ├── _lifespan.py             #   异步生命周期
│   │   ├── _router/                 #   8 个 REST 路由
│   │   ├── storage/                 #   RedisStorage + ORM
│   │   ├── message_bus/             #   RedisMessageBus
│   │   ├── _service/                #   ChatService, SessionService
│   │   └── _manager/                #   运行时管理器
│   ├── credential/                  # 凭据管理 (Factory 模式)
│   ├── embedding/                   # 嵌入模型
│   ├── tts/                         # 文本转语音
│   ├── mcp/                         # MCP 客户端管理
│   ├── state/                       # Agent 状态机
│   ├── skill/                       # Skill 系统
│   ├── exception/                   # 自定义异常
│   └── types/                       # 共享类型
├── tests/                           # ★ 单元测试 (76+ 文件)
├── examples/                        # 示例
│   ├── agent_service/               #   生产服务示例
│   └── web_ui/                      #   全栈 Web UI
└── pyproject.toml                   # 构建与依赖配置

标注 ★ 的模块是理解框架的关键入口。

---

3. 源码阅读路线图

按照依赖关系，推荐以下阅读顺序：

第一阶段：理解消息与事件（基础）

message/     →    event/      →    state/
(Msg 类型)       (事件体系)        (状态机)

• message/_base.py — Msg 是所有消息的基类，UserMsg/AssistantMsg/SystemMsg 是其子类型。理解 Msg 的结构（role, content blocks, metadata）是一切的基础。
• message/_block.py — ContentBlock 体系：TextBlock、ThinkingBlock、ToolCallBlock、ToolResultBlock、DataBlock，支持流式增量（delta）。
• event/_event.py — 事件是 AgentScope 的”神经中枢”。重点关注 EventType 枚举和 EventBase 的继承树。每个事件对应 Agent 执行过程中的一个时间点。

第二阶段：理解模型抽象（核心接口）

model/_base.py   →   model/_openai_chat.py   →   formatter/
(ChatModelBase)      (具体实现示例)              (格式化器)

• model/_base.py — ChatModelBase 是框架最核心的抽象之一。关注 _prepare_kwargs()、_call_api()、stream() 这三个方法签名。理解基类如何统一流式和非流式两种模式。
• model/_openai_chat.py — 最简单的具体实现，作为参考模板。
• formatter/ — 每个提供商有一个 ChatFormatter 和一个 MultiAgentFormatter，负责将内部 Msg 转换为各 API 期望的格式。

第三阶段：理解 Agent 主循环（核心逻辑）

agent/_config.py   →   agent/_agent.py
(配置模型)             (Agent 类)

• agent/_config.py — ContextConfig（上下文窗口管理）、ReActConfig（循环控制）、ModelConfig（模型配置）。这些 Pydantic 模型是 Agent 行为的控制面。
• agent/_agent.py — 框架核心，约 96KB。重点关注 reply() 方法（第 ~330 行起），理解 ReAct 循环：
  1. Think — 调用模型，流式返回文本/思考/tool calls
  2. Act — 解析 tool calls，通过 _ToolCallBatch 并行执行
  3. Observe — 收集工具结果，喂回下一轮

  阅读技巧：Agent 类有 async def reply() 和内部的 _execute_tool_calls() 两个关键方法。reply() 中打出了所有事件（Event），理解事件顺序就等于理解了 Agent 的执行流程。

第四阶段：理解工具与沙箱（执行环境）

tool/_base.py   →   tool/_toolkit.py   →   tool/_builtin.py
(ToolBase)          (Toolkit 装配)          (内置工具)
                                        →   workspace/
                                            (沙箱执行)

• tool/_base.py — ToolBase 是所有工具的抽象基类。关注 execute() 的签名和 to_tool_schema()（生成 OpenAI 兼容的工具定义）。
• tool/_toolkit.py — Toolkit 将多个工具组合在一起挂载到 Agent。关注 tool_choice 逻辑：如何根据模型返回决定执行哪些工具。
• tool/_builtin.py — 内置工具（Bash, Edit, Read, Write, Glob, Grep）。注意它们如何通过 workspace 参数注入执行环境。
• workspace/_base.py — WorkspaceBase 定义了沙箱契约。LocalWorkspace 是最简单的实现，适合作为理解起点。

第五阶段：理解服务层（生产部署）

app/_app.py   →   app/_lifespan.py   →   app/_router/_chat.py
(工厂函数)          (生命周期)             (对话路由)
                                            →   app/_service/_chat_service.py
                                                (ChatService 编排)

• app/_app.py — create_app() 是服务层的入口。接收 Storage、MessageBus、WorkspaceManager 三个必选依赖，以及若干扩展点。
• app/_lifespan.py — 使用 AsyncExitStack 统一管理所有资源的启动和关闭。
• app/_service/_chat_service.py — ChatService 是编排 Agent 回复的核心服务，它调用 Agent.reply() 并将事件流式返回给前端。

---

4. 核心抽象与设计模式

4.1 策略模式 —— 模型提供商

ChatModelBase 定义接口，每个提供商（OpenAI、Anthropic、DashScope……）各自实现 _call_api() 和 _prepare_kwargs()。Agent 通过 ModelCard 查找并使用模型，完全不知道具体提供商是谁。

ChatModelBase (抽象策略)
    ├── OpenAIChatModel (具体策略)
    ├── AnthropicChatModel
    ├── DashScopeChatModel
    └── ...

4.2 工厂模式 —— Credential + App Factory

• CredentialFactory — 根据模型名称自动匹配合适的 Credential 实现。
• create_app() — FastAPI 应用工厂，接受依赖注入参数返回配置好的 App。

4.3 洋葱模型 —— 中间件管线

中间件在 Agent 的 reply() 前后形成嵌套调用链：

# 伪代码示意
def reply(msg):
    for m in middlewares:
        m.before(msg)       # 前置处理（如记忆注入）
    result = agent_impl(msg)  # 核心逻辑
    for m in reversed(middlewares):
        m.after(result)     # 后置处理（如记忆提取、TTS）
    return result

4.4 观察者模式 —— 事件系统

Agent 每一步都发出事件，EventBus 将事件广播给所有订阅者（前端 SSE、日志、中间件）。

4.5 惰性导入 —— 可选依赖

所有非核心依赖（gemini、ollama、redis、docker 等）在模块顶部不 import，而是在使用点导入：

# ✅ 正确：在函数内部惰性导入
def some_function():
    import google.genai  # 来自 `gemini` extra
    ...

# ❌ 错误：在模块顶部导入
import google.genai  # 会导致 import agentscope 时就报错

4.6 依赖注入 —— create_app()

app = create_app(
    storage=RedisStorage(...),       # 可替换为任意 StorageBase 实现
    message_bus=RedisMessageBus(...),# 可替换为任意 MessageBus 实现
    workspace_manager=...,
    extra_agent_middlewares=...,     # 自定义中间件
    custom_agent_cls=...,            # 自定义 Agent 类
)

---

5. 测试体系详解

5.1 测试框架

AgentScope 使用 unittest.IsolatedAsyncioTestCase（而非 pytest），所有测试是异步的。

约定	示例
文件名	{模块}_test.py
类名	{Name}Test(IsolatedAsyncioTestCase)
方法名	async def test_{场景}(self)
初始化	async def asyncSetUp(self)
清理	async def asyncTearDown(self)

5.2 Mock 模型

tests/utils.py 提供了 MockModel：

from tests.utils import MockModel

mock = MockModel()
mock.set_responses([
    ChatResponse(text="Hello", ...),
    ChatResponse(tool_calls=[...], ...),
])

agent = Agent(name="test", model=mock, ...)
result = await agent.reply("Hi")

5.3 条件跳过

import unittest

@unittest.skipIf(sys.platform == "win32", "Windows 不支持 Bash")
class BashToolTest(IsolatedAsyncioTestCase):
    ...

@unittest.skipUnless(os.getenv("ANTHROPIC_API_KEY"), "需要 API Key")
class AnthropicModelTest(IsolatedAsyncioTestCase):
    ...

5.4 运行测试

# 运行全部测试
pytest tests

# 运行单个文件
pytest tests/agent_basic_test.py -x -v

# 运行单个测试方法
pytest tests/agent_basic_test.py::AgentTest::test_reply -x -v

# 跳过慢速测试（不含 external service 依赖）
pytest tests -x -v --ignore-glob='*redis*' --ignore-glob='*docker*'

# 查看覆盖率
pytest tests --cov=agentscope --cov-report=html

提示：tests/storage_redis_test.py 使用 fakeredis，不需要真 Redis 即可运行。

---

6. 开发工作流

6.1 分支策略

# 创建特性分支
git checkout -b feat/my-feature

# 分支命名建议
feat/xxx        # 新功能
fix/xxx         # Bug 修复
refactor/xxx    # 重构
docs/xxx        # 文档
test/xxx        # 测试

6.2 提交约定

遵循 Conventional Commits：

<type>(<scope>): <description>

# 示例
feat(models): add support for Claude-3.5 Sonnet
fix(agent): resolve memory leak in ReAct loop
refactor(formatter): simplify message formatting logic

type 可用值：feat、fix、docs、style、refactor、perf、test、ci、chore。

scope 必须全小写。

6.3 提 PR 前检查清单

• [ ] pre-commit run --all-files 通过
• [ ] pytest tests 通过（或至少相关测试通过）
• [ ] 惰性导入原则：可选依赖未出现在模块顶部
• [ ] 新功能有配套单元测试
• [ ] 公开 API 有 docstring
• [ ] PR 标题符合 conventional commits 格式
• [ ] 没有多余的大批量文件改动

6.4 CI 流水线

GitHub Actions 配置在 .github/workflows/：

Workflow	触发条件	内容
unittest.yml	push + PR to main	3 OS × Python 3.11，运行全量测试
pre-commit.yml	PR to main	运行 pre-commit 检查
publish-pypi.yml	tag push	发布到 PyPI
web-ui.yml	PR to main	构建 Web UI
pr-title-check.yml	PR to main	校验 PR 标题格式

---

7. 常见任务速查

7.1 添加新的模型提供商

需要同时贡献 4 个部分（详见 CONTRIBUTING_zh.md §5）：

# 1. Credential 类
src/agentscope/credential/_my_provider.py     # 继承 CredentialBase

# 2. Model 类
src/agentscope/model/_my_provider/            # 继承 ChatModelBase
src/agentscope/model/_my_provider/_models/    # YAML 模型卡片

# 3. Formatter 类
src/agentscope/formatter/_my_provider_formatter.py  # ChatFormatter + MultiAgentFormatter

# 4. 测试
tests/model_my_provider_test.py
tests/formatter_my_provider_test.py

7.2 添加新的 Workspace 后端

需要两个类：

# 1. Workspace 类（继承 WorkspaceBase）
src/agentscope/workspace/_my_workspace.py

# 2. Workspace Manager 类（继承 WorkspaceManagerBase）
src/agentscope/app/workspace_manager/_my_manager.py

7.3 添加新的内置工具

# src/agentscope/tool/_builtin.py 中新增
@tool_registry.register("my_tool")
class MyTool(ToolBase):
    def __init__(self, workspace: WorkspaceBase, ...):
        ...
    async def execute(self, ...) -> ToolResponse:
        ...
    @classmethod
    def to_tool_schema(cls) -> dict:
        ...

7.4 添加新的中间件

# src/agentscope/middleware/_my_middleware.py
from ._base import MiddlewareBase

class MyMiddleware(MiddlewareBase):
    async def on_reply_start(self, ctx):
        # Agent 开始回复前执行
        ...

    async def on_reply_end(self, ctx):
        # Agent 回复结束后执行
        ...

7.5 添加新的 Skill

在 examples/ 或独立仓库中贡献，参考 skill/_base.py 和已有的 LocalSkillLoader。

7.6 依赖管理速查

依赖类型	声明位置	导入方式
核心依赖	pyproject.toml → [project.dependencies]	模块顶部直接 import
可选依赖	pyproject.toml → [project.optional-dependencies]	使用点惰性导入
开发依赖	pyproject.toml → [project.optional-dependencies] → dev	仅开发环境需要

---

8. 调试技巧

8.1 本地调试 Agent

import asyncio
from agentscope.agent import Agent
from agentscope.model import OpenAIChatModel
from tests.utils import MockModel

# 使用 MockModel 避免调用真实 API
mock = MockModel()
mock.set_responses([
    ChatResponse(text="I'll use the bash tool to check."),
    ChatResponse(tool_calls=[{"name": "bash", "args": {"cmd": "ls -la"}}]),
])

agent = Agent(
    name="debug_agent",
    model=mock,
    system_prompt="You are a helpful assistant.",
    tools=[...],
)
result = asyncio.run(agent.reply("List files"))
print(result)

8.2 查看事件流

from agentscope.event import EventBus

bus = EventBus()

@bus.subscribe
def on_event(event):
    print(f"[{event.type}] {event.__class__.__name__}")

# Agent 创建时传入这个 bus
agent = Agent(..., event_bus=bus)

8.3 使用 OpenTelemetry 追踪

from agentscope.middleware import TracingMiddleware

middleware = TracingMiddleware(
    service_name="my-agent",
    otlp_endpoint="http://localhost:4317",
)
agent = Agent(
    ...,
    middlewares=[middleware],
)

8.4 日志级别

import agentscope

# 设置日志级别
agentscope.setup_logger(level="DEBUG")

8.5 文件定位速查表

你想找什么	文件路径
Agent 主循环	src/agentscope/agent/_agent.py → reply() 方法
模型调用逻辑	src/agentscope/model/_base.py → ChatModelBase._call_api()
事件定义	src/agentscope/event/_event.py
消息类型定义	src/agentscope/message/_base.py
内容块类型	src/agentscope/message/_block.py
工具基类	src/agentscope/tool/_base.py
MCP 适配器	src/agentscope/tool/_adapters.py
权限引擎	src/agentscope/permission/_engine.py
沙箱基类	src/agentscope/workspace/_base.py
本地沙箱	src/agentscope/workspace/_local_workspace.py
中间件基类	src/agentscope/middleware/_base.py
服务工厂	src/agentscope/app/_app.py
对话服务	src/agentscope/app/_service/_chat_service.py
存储抽象	src/agentscope/app/storage/_base.py
Redis 实现	src/agentscope/app/storage/_redis_storage.py
消息总线	src/agentscope/app/message_bus/_base.py
凭据工厂	src/agentscope/credential/（__init__.py + 各实现）
Mock 测试工具	tests/utils.py
Agent 基本测试	tests/agent_basic_test.py
生产服务示例	examples/agent_service/main.py

---

附录：依赖关系拓扑

                  Agent (agent/_agent.py)
                 /     |      |       
                /      |      |        
         ChatModel   Toolkit  Middleware  Permission
         (model/)   (tool/)  (middleware/) (permission/)
            |           |         |
        Formatter   Workspace   EventBus
       (formatter/) (workspace/) (event/)
            |           |
         Msg          Offloader
       (message/)   (workspace/)

     --- 服务层 ---
         create_app()
            |
    ┌───────┼───────────┐
    |       |           |
 Storage  MessageBus  ChatService
(app/storage) (app/message_bus) (app/_service)

箭头方向表示依赖关系：A → B 表示 A 依赖 B。

---