QwenPaw 架构与设计思想分析报告

一、项目概览

QwenPaw（前身为 CoPaw，v1.0.0 融入 Qwen 开源生态后更名）是一个个人 AI 助理平台，Apache-2.0 许可，Python 3.10+。核心定位是本地优先、多端触达、技能驱动、多智能体协作。

技术栈：AgentScope（多智能体框架）、FastAPI（HTTP）、Vue 3（Console）、Tauri（桌面壳）、Playwright（浏览器自动化）、MCP（Model Context Protocol）。

二、源码目录结构

qwenpaw/
├── agents/                  # Agent 核心
│   ├── react_agent.py       # QwenPawAgent 主类
│   ├── tool_guard_mixin.py  # 工具安全拦截
│   ├── coding_mode_mixin.py # Coding Mode 增强
│   ├── hooks/               # BootstrapHook 等
│   ├── memory/              # BaseMemoryManager + 3 个后端
│   ├── context/             # BaseContextManager + LightContextManager
│   ├── skill_system/        # 技能系统（pool + workspace）
│   ├── tools/               # 21 个内置工具
│   └── ...
├── providers/               # 8 个 provider 实现 + 抽象基类
│   ├── provider.py          # Provider(ProviderInfo, ABC)
│   ├── provider_manager.py  # ProviderManager 单例
│   ├── openai_provider.py   # 含 OpenAI/OpenCode/Kilo 三个类
│   ├── anthropic_provider.py / gemini_provider.py
│   ├── ollama_provider.py / lmstudio_provider.py
│   ├── openrouter_provider.py
│   ├── multimodal_prober.py / model_capability_cache.py
│   ├── capability_baseline.py / rate_limiter.py
│   ├── retry_chat_model.py / oauth/
├── security/
│   ├── tool_guard/          # 工具调用拦截
│   │   ├── engine.py / models.py
│   │   ├── execution_level.py  # ToolExecutionLevel (STRICT/SMART/AUTO/OFF)
│   │   ├── approval.py / i18n.py / utils.py
│   │   └── guardians/       # rule_guardian / file_guardian / shell_evasion_guardian
│   └── skill_scanner/       # 技能安装前安全扫描
├── app/
│   ├── _app.py              # FastAPI lifespan（两阶段启动）
│   ├── multi_agent_manager.py / workspace/
│   ├── agent_context.py     # 多 agent 路由
│   ├── mcp/                 # stateful_client / manager / watcher
│   ├── channels/            # 18 个渠道 + 基础设施（详见第 4 节）
│   ├── runner/              # AgentRunner + DynamicMultiAgentRunner
│   ├── routers/ / approvals/
├── plugins/                 # PluginType 枚举 / loader / registry
├── plan/                    # hints.py / schemas.py / broadcast.py
├── config/ / constant.py / exceptions.py

三、分层架构

按"从外到内"看下来一共六层：

┌──────────────────────────────────────────────────┐
│   Channels Layer（18 个内置渠道）                │
├──────────────────────────────────────────────────┤
│   Console Layer（Web UI / Tauri Desktop）         │
├──────────────────────────────────────────────────┤
│   Agent Layer（QwenPawAgent + Mixins）            │
├──────────────────────────────────────────────────┤
│   Security Layer（ToolGuard + SkillScanner）      │
├──────────────────────────────────────────────────┤
│   Provider Layer（Provider(ProviderInfo, ABC)）   │
├──────────────────────────────────────────────────┤
│   Memory / Context Layer（3 后端 + LightContext） │
└──────────────────────────────────────────────────┘

四、Channels Layer

app/channels/registry.py 的 _BUILTIN_SPECS 字典里整 18 个内置渠道：

类别	渠道
即时通讯	imessage、discord、dingtalk、feishu、qq、telegram、mattermost、mqtt、matrix、wecom、wechat、onebot
智能终端	console、voice、sip
自家生态	xiaoyi、yuanbao
智能终端	console、voice、sip
自家生态	xiaoyi、yuanbao

每个渠道为独立子目录，含 channel.py 加配套（utils.py、auth.py、constants.py、cards/ 等），不与渠道基础设施混层。

核心契约（base.py:BaseChannel(ABC)）：

BaseChannel.channel: str — 渠道标识
_is_native_payload(payload) / _consume_one_request(merged) — 原生格式适配
merge_native_items / merge_requests — 批量合并
MessageRenderer + RenderStyle — 渠道适配渲染
get_access_control_store() — 渠道级访问控制

ChannelManager（manager.py）：_process_batch 按原生 / 通用分流批处理，调用 BaseChannel._consume_one_request；通过 UnifiedQueueManager 控制每渠道队列上限（_CHANNEL_QUEUE_MAXSIZE = 1000）。

五、Agent Layer

类层级（agents/react_agent.py:85）：

class QwenPawAgent(CodingModeMixin, ToolGuardMixin, ReActAgent):
    # MRO: QwenPawAgent → CodingModeMixin → ToolGuardMixin → ReActAgent
    # Each `_acting` override must call `super()._acting(...)` to keep
    # the full chain active.

super()._acting() / super()._reasoning() 调用是 MRO 链正常工作的必要条件。QwenPawAgent._acting 加 plan gate，ToolGuardMixin._acting 加安全拦截，两层叠合。

5.1 工具注册三层机制（`react_agent.py:237-416`）

A) 硬编码内置工具（21 个）（行 289-317）：

tool_functions = {
    "execute_shell_command", "read_file", "write_file", "edit_file",
    "grep_search", "glob_search", "browser_use", "desktop_screenshot",
    "view_image", "view_video", "send_file_to_user", "get_current_time",
    "set_user_timezone", "get_token_usage", "delegate_external_agent",
    "list_agents", "chat_with_agent", "submit_to_agent", "check_agent_task",
    "spawn_subagent", "run_tool_batch",
    # 仅在 make-skill 技能启用时注册
    **({"materialize_skill": materialize_skill} if "make-skill" in effective_skills else {}),
}

B) 插件工具（__all__ 自动发现）（行 322-336）：通过 getattr(tools_module, "__all__", []) 扫描；插件工具若未在 config 中显式启用则跳过（安全设计，区别于硬编码工具的"默认启用"）。

C) 后台任务管理工具（自动注册）（行 376-403）：当任一已启用工具的 async_execution=True 时，自动注册 view_task / wait_task / cancel_task 三个 AgentScope Toolkit 内置任务管理工具。当前仅 execute_shell_command 和 delegate_external_agent 两个工具支持 async_execution。

D) Coding Mode 工具：lsp、ast_search（仅在 coding_mode.enabled=True 且 CLI / 语言服务器可达时注册）。

5.2 Hook 生命周期（`react_agent.py:488-529`）

# BootstrapHook (pre_reasoning)
self.register_instance_hook("pre_reasoning", "bootstrap_hook", ...)

# ContextManager 生命周期（4 个阶段，pre_reply 剪枝 tool_result）
self.register_instance_hook("pre_reasoning", "context_pre_reasoning", ...)
self.register_instance_hook("pre_reply",   "context_pre_reply",      ...)  # 工具结果剪枝
self.register_instance_hook("post_acting", "context_post_acting",     ...)  # 协议占位
self.register_instance_hook("post_reply",  "context_post_reply",      ...)

5.3 媒体容错三层机制（`react_agent.py:1022-1245`）

1) 主动层（调用前）：

should_strip = not get_active_model_supports_multimodal() or self._model_rejects_media()
若支持 formatter 请求时规范化：_set_formatter_media_strip(True)（通过 formatter._qwenpaw_force_strip_media 标志）
否则：_proactive_strip_media_blocks() → _strip_media_blocks_from_memory()

2) 被动层（失败时）：

super()._reasoning() 抛异常 → _is_bad_request_or_media_error() 判断（status_code=400 或关键词 image/audio/video/vision/multimodal/image_url）→ 切换至请求时 strip 重试，再失败则降级到 memory 块剥离

3) 学习层：

失败后 get_capability_cache().learn(model_key, "rejects_media", True) 写入能力缓存

4) media block 类型：_MEDIA_BLOCK_TYPES = {"image", "audio", "video", "file"}

formatter 标志位与 memory 块剥离是两种并存策略，由 _uses_request_time_media_normalization() 决定走哪条。

5.4 Plan 门控（`plan/hints.py`）

plan_notebook 维护 4 个标志位：

Flag	触发	作用
`_plan_tool_gate`	`/plan` 进入	初始门——仅 `create_plan` 可执行
`_plan_awaiting_user_confirm`	`_acting()` 预锁	仅 `create_plan` / `revise_current_plan` / `finish_plan` 可执行
`_plan_text_only_after_mutation`	plan 变更后由 `_acting()` 设置	下一轮 `_reasoning` 强制 `tool_choice="none"`
`_plan_just_mutated`	同上	用于 broadcast 流程

_filter_plan_tools()（react_agent.py:998-1020）在工具执行之前抢锁，防 asyncio.gather 并发竞态。

5.5 Auto-Continue

_auto_continue_if_text_only()（行 895-965）：当模型返回纯文本、且 running.auto_continue_on_text_only=True 时，注入中英双语并重跑最多 _AUTO_CONTINUE_MAX_EXTRA=2 次推理，直到出现 tool_use 或触顶。

5.6 MCP 客户端恢复

_recover_mcp_client()（行 621-636）：

_reconnect_mcp_client(client)  →  成功? 返回原 client
                                      ↓ 失败
_rebuild_mcp_client(client)    →  None? 返回 None
                                      ↓ 重建
_reconnect_mcp_client(rebuilt) →  成功? 返回
                                      ↓ 失败
返回 None

重建依赖 client._qwenpaw_rebuild_info 元数据（transport、name、command、args、env、url、headers）。

5.7 多 Agent 协作

Workspace 是 Agent 容器（app/workspace/workspace.py），MultiAgentManager 管理多个 Workspace。路由优先级（app/agent_context.py:45-129）：

1. 显式 agent_id 参数 2. request.state.agent_id（agent-scoped router 注入） 3. X-Agent-Id 请求头 4. config 中 agents.active_agent 兜底（默认 "default"）

MultiAgentManager.get_agent(agent_id) 懒加载——锁仅在 dict 检查时短暂持有，workspace 启动期间释放锁以支持并发初始化，多个并发请求通过 asyncio.Event 协调。

六、Security Layer

6.1 Tool Guard 执行级别

security/tool_guard/execution_level.py 定义 4 个 ToolExecutionLevel：

class ToolExecutionLevel(str, Enum):
    STRICT = "strict"   # 所有工具需审批
    SMART  = "smart"    # INFO/LOW 自动放行，MEDIUM+ 需审批（推荐）
    AUTO   = "auto"     # 仅显式 guarded_tools 需审批（向后兼容）
    OFF    = "off"      # 完全关闭

6.2 三 Guardian 子模块

security/tool_guard/guardians/ 下三个具体 guardian：

RuleBasedToolGuardian — 规则匹配（三个环境变量优先级叠加：QWENPAW_TOOL_GUARD_TOOLS / QWENPAW_TOOL_GUARD_DENIED_TOOLS / QWENPAW_TOOL_GUARD_AUTO_DENIED_RULES）
FilePathToolGuardian — 路径白名单 / 黑名单（含 _DEFAULT_DENY_DIRS 保护 secret 目录）
ShellEvasionGuardian — Shell 混淆检测

硬编码密钥检测、数据外泄检测在 security/skill_scanner/（技能安装前扫描），与 ToolGuard 职责分离。

6.3 ToolGuardMixin 行为

agents/tool_guard_mixin.py：覆盖 _acting（拦截） + _reasoning（预处理）。决策通过 _init_tool_guard() 懒初始化 ToolGuardEngine 和 ApprovalService：

self._tool_guard_engine = get_guard_engine()
self._tool_guard_approval_service = get_approval_service()

6.4 多语言支持

_TOOL_GUARD_I18N 字典包含 4 种语言：en / zh / ru / ja（tool_guard_mixin.py:46）。

6.5 默认开启逻辑

_guard_enabled() 优先级：QWENPAW_TOOL_GUARD_ENABLED 环境变量 > config.json > True（默认开启）。

七、Provider Layer

7.1 数据模型

providers/provider.py 定义：

class ModelInfo(BaseModel):
    id, name
    supports_multimodal / supports_image / supports_video  # 三模态独立标记
    probe_source   # 'documentation' / 'probed'
    is_free
    max_tokens     # 默认 8192
    max_input_length  # 默认 128 * 1024
    generate_kwargs   # per-model 覆盖

class ExtendedModelInfo(ModelInfo):
    provider, input_modalities, output_modalities

class ProviderInfo(BaseModel):  # 含 generate_kwargs
    ...

class Provider(ProviderInfo, ABC):  # 第 195 行
    @abstractmethod get_chat_model_instance(...)
    @abstractmethod list_models(...)
    @abstractmethod check_connection(...)

7.2 参数合并

get_effective_generate_kwargs(model_id)（第 336 行）：

return self._deep_merge(
    self.generate_kwargs,        # provider 级作 base
    model.generate_kwargs or {}, # model 级覆盖
)

7.3 Provider 实现

8 个实现：openai_provider.py 含 3 个类（OpenAIProvider / OpenCodeProvider / KiloProvider），加 anthropic_provider.py、gemini_provider.py、ollama_provider.py、lmstudio_provider.py、openrouter_provider.py 各 1 个。

7.4 配套模块

multimodal_prober.py — ProbeResult 数据类
model_capability_cache.py — get_capability_cache()，含 learn() 写入失败能力
capability_baseline.py — 文档级能力基线
rate_limiter.py — 速率控制
retry_chat_model.py — 重试包装
openai_chat_model_compat.py — OpenAI 兼容适配
oauth/ — OAuth 子模块

八、Memory 与 Context Layer

agents/memory/__init__.py 导出 3 个后端：

from .agent_md_manager import AgentMdManager            # 基于文件系统
from .base_memory_manager import BaseMemoryManager      # 抽象基类
from .reme_light_memory_manager import ReMeLightMemoryManager
from .adbpg_memory_manager import ADBPGMemoryManager    # 注册 "adbpg" 后端

生命周期（base_memory_manager.py）：

__init__ → start() → summarize()/memory_search() → close()

主动记忆（Proactive）：通过 agents/memory/proactive.py 提供，按需懒导入避免 proactive → react_agent → agents.memory 循环。

8.1 LightContextManager

agents/context/light_context_manager.py：

@context_registry.register("light")
class LightContextManager(BaseContextManager):
    """Tool-result pruning via _prune_tool_result()
       Context-size checking via _check_context()
       Message compaction via _compact_context()
       Agent context retrieval via get_agent_context()"""

Token 估算：使用 EstimatedTokenCounter（utils/estimate_token_counter.py），通过 get_token_counter() 获取；token 格式化 _fmt_tokens(n) 在 ≥1k 时显示为 82.3k。

主动记忆跳过：_AUTOMATION_MEMORY_SKIP_SOURCES = {"cron", "heartbeat"}——非用户自动触发的请求不写入长期记忆。

完整的压缩机制由 _compact_context() + _check_context() + _prune_tool_result() 三段构成，区分 INITIAL_USER_MESSAGE_EN/ZH 与 UPDATE_USER_MESSAGE_EN/ZH 两套压缩 prompt。

九、Plugins 与 Skills

9.1 PluginType 枚举

plugins/architecture.py:12-35 定义 PluginType(str, Enum) 6 个值：

TOOL      = "tool"       # 注册 LLM 可调用的工具函数
PROVIDER  = "provider"   # 注册自定义 LLM 端点
HOOK      = "hook"       # 应用启动 / 关闭钩子
COMMAND   = "command"    # 注册 /slash 控制命令
FRONTEND  = "frontend"   # 注册前端 JS bundle
GENERAL   = "general"    # 不匹配其他类别的兜底

PluginEntryPoints 是 Pydantic BaseModel，含 frontend / backend 入口字符串。

9.2 Skills 系统

agents/skill_system/ 7 个模块：

models.py — SkillInfo、SkillConflictError、BuiltinSkillIdentity、BuiltinSkillVariant
store.py — get_workspace_skills_dir / get_skill_pool_dirs / read_skill_manifest / safe_skill_dir
pool_service.py — SkillPoolService（全局技能池）
workspace_service.py — SkillService（工作区级）
registry.py — resolve_effective_skills / apply_skill_config_env_overrides / ensure_skills_initialized 等
hub.py — 远端市场接入
__init__.py — 统一出口

关键设计：

resolve_effective_skills(workspace_dir, channel_name) — 按 workspace + channel 解析有效技能集
apply_skill_config_env_overrides(workspace_dir, channel_name) — contextmanager，在 super().reply() 期间施加技能级 env 覆盖
BUILTIN_SKILL_LANGUAGES = ("en", "zh") — 内置技能仅这两种本地化变体（-en / -zh 目录命名约定）
reconcile_pool_manifest / reconcile_workspace_manifest — 同步内置技能到 pool / workspace

十、Plan System 关键路径

plan/hints.py 核心 API：

def set_plan_gate(plan_notebook, enabled: bool)         # 切换 _plan_tool_gate
def clear_plan_awaiting_user_confirm(plan_notebook)     # 每轮开始清 3 个 flag
def check_plan_tool_gate(plan_notebook, tool_name)      # 返回错误字符串或 None
def should_skip_auto_continue(plan_notebook)            # plan 进行中跳过 auto-continue

QwenPawAgent._acting 通过 check_plan_tool_gate 实现执行门，预锁机制（react_agent.py:803-811）确保 asyncio.gather 并发的兄弟工具在 mutation tool 真正执行前就被锁住。

十一、部署模式

6 种部署方式：

pip 安装 — 开发者 / 服务器，完整控制
脚本安装 — 新手 / 桌面用户，零配置，自动管理 Python + Node
Docker — 服务器 / NAS，隔离运行，持久化卷管理
阿里云 ECS — 云上一键部署
魔搭创空间 — 无需安装，云端运行，需设为非公开
桌面应用（Tauri） — Beta，零配置，跨平台

十二、关键文件索引

模块	文件	关键导出
Agent 主体	`agents/react_agent.py:85`	`QwenPawAgent`
工具安全	`agents/tool_guard_mixin.py:79`	`ToolGuardMixin`, `_normalize_tool_guard_ui_lang`
安全引擎	`security/tool_guard/engine.py:54`	`ToolGuardEngine`, `_guard_enabled`
执行级别	`security/tool_guard/execution_level.py:15`	`ToolExecutionLevel` (STRICT/SMART/AUTO/OFF)
数据模型	`security/tool_guard/models.py`	`GuardSeverity`, `GuardThreatCategory`, `ToolGuardResult`, `GuardFinding`
Provider 抽象	`providers/provider.py:195`	`Provider(ProviderInfo, ABC)`
Provider 管理	`providers/provider_manager.py`	`ProviderManager.get_instance()`
多 Agent	`app/multi_agent_manager.py:22`	`MultiAgentManager`, `Workspace` 懒加载
Agent 路由	`app/agent_context.py:45`	`get_agent_for_request` (4 级优先级)
渠道基类	`app/channels/base.py`	`BaseChannel(ABC)`, `TextContent`, `ContentType`
渠道注册	`app/channels/registry.py:20-37`	`_BUILTIN_SPECS` (18 个内置渠道)
渠道管理	`app/channels/manager.py`	`ChannelManager._process_batch`
记忆抽象	`agents/memory/base_memory_manager.py:19`	`BaseMemoryManager(ABC)`
上下文	`agents/context/light_context_manager.py:64`	`LightContextManager`
技能	`agents/skill_system/__init__.py`	`SkillPoolService`, `SkillService`, `resolve_effective_skills`
插件类型	`plugins/architecture.py:12`	`PluginType(Enum, 6 值)`
Plan 门控	`plan/hints.py:46,55,71,84`	`set_plan_gate`, `clear_plan_awaiting_user_confirm`, `check_plan_tool_gate`, `should_skip_auto_continue`
MCP 客户端	`app/mcp/stateful_client.py`	`HttpStatefulClient`, `StdIOStatefulClient`
MCP 管理	`app/mcp/manager.py`	MCPClientManager (含 OAuth 注入)
应用启动	`app/_app.py:223-510`	`lifespan` (Phase 1 < 100ms / Phase 2 background)

十三、核心设计哲学

1. 本地优先 — 密钥分离到 ~/.qwenpaw.secret/，数据 ~/.qwenpaw/，绝不自动上传。 2. 技能驱动 — resolve_effective_skills() 按 workspace + channel 动态筛选。 3. 安全默认 — _guard_enabled() 缺省 True；插件工具无 config 不注册；技能安装前自动扫描。 4. 渐进式复杂性 — 三行命令 → 多 Agent 协作。 5. 韧性设计 — LLM 重试（指数退避，参见 retry_chat_model.py）、MCP 三段恢复（reconnect → rebuild → reconnect）、媒体三层容错、Plan 并发竞态预锁。 6. 冷热分离 + 两阶段启动（_app.py:230-333 注释明确）：

Phase 1（target < 100ms）：restore cleanup → env register → 迁移 → 核心 manager 实例化（MultiAgentManager / ProviderManager / LocalModelManager）→ 暴露 app.state → 日志输出 Server ready in {fast_elapsed:.3f}s
Phase 2（background）：asyncio.create_task(_background_startup()) 并行 plugin_loader + start_all_configured_agents() → 本地模型 resume → 插件 provider 注册 → 控制命令注册 → 启动钩子执行

十四、关键设计亮点

MRO 协同：QwenPawAgent._acting + ToolGuardMixin._acting + ReActAgent._acting 通过 super() 链式覆盖，任何中间层遗漏 super()._acting() 会破坏整条拦截链（注释行 99-101 明确警告）。
Plan 预锁（react_agent.py:803-811）解决 asyncio.gather 并发下 mutation tool 还没执行就被兄弟工具绕过门控的竞态。
formatter 标志位（formatter._qwenpaw_force_strip_media）实现请求时规范化，区别于 memory 块级剥离的"事后清理"。
Media 块嵌套剥离（react_agent.py:1387-1405）：不仅剥离顶层 media 块，还深入 ToolResultBlock.output 内部剥离嵌套的 image / audio。
_auto_continue_if_text_only 双向语言支持：根据 agent_config.language 选 _AUTO_CONTINUE_HINT_EN/ZH，并把 assistant 上轮文本尾段 600 字符包入让模型自审。

十五、总结

QwenPaw 是一个架构精良、设计思想清晰的 AI 助理平台。其核心优势：

1. 清晰的六层分层架构，每层职责单一、接口清晰 2. 丰富的扩展机制（Skills + Plugin + MCP + Custom Provider），无锁定 3. 多层安全防线，本地优先的隐私设计 4. 渐进式复杂度，新手到专家皆可驾驭 5. 弹性韧性设计，从 LLM 调用重试到 MCP 故障恢复，全链路保障 6. 完整的生态系统：多端接入（18 个内置渠道）、定时任务、记忆进化（3 后端）、多 Agent 协作

潜在的改进空间：Agent 间通信目前主要通过工具调用实现，缺少原生消息总线；Plan 系统与 Agent 核心的耦合度较高；测试覆盖率有进一步提升空间（当前 fail_under = 30）。