阿里开源桌面 Agent「CoPaw」源码深度拆解

1.2 模块关系

整个系统以 CoPawAgent 为核心，该代理继承自 AgentScope 的 ReActAgent，实现了工具调用和记忆管理功能。系统通过命令行接口启动应用服务，应用服务接收来自各种消息通道（钉钉、飞书、QQ、Discord 等）的消息，传递给 AI 代理进行处理，并返回响应。

二、核心算法分析

2.1 ReAct 推理算法

CoPaw 的核心算法基于 ReAct（Reasoning and Acting）范式，实现在 react_agent.py 中：

from agentscope.agent import ReActAgent

class CoPawAgent(ReActAgent):

ReAct 算法允许 AI 在推理（reasoning）和行动（acting）之间交替进行，能够根据当前情况选择合适的工具并基于结果做出下一步决策。

ReAct（Reasoning and Acting）是一种结合推理和行动的 AI 代理框架，它让 AI 模型能够在推理（思考）和行动（调用工具）之间交替进行。这种机制使 AI 能够更有效地解决问题，特别是在需要外部知识或工具的复杂任务中。

**ReAct 算法的核心思想是在解决问题时，AI 模型不仅要进行内部推理，还要能够与外部环境交互。**它的工作流程如下：

1. 观察输入：接收用户的问题或当前任务
2. 推理：分析问题，制定解决策略
3. 行动：根据推理结果选择适当的工具执行操作
4. 观察结果：获取工具执行的结果
5. 重复循环：回到推理步骤，直到得到最终答案

CoPaw 中的 ReAct 实现 在 CoPaw 中，ReAct 算法的实现基于 AgentScope 框架。CoPawAgent 继承自 AgentScope 的 ReActAgent，实现了 ReAct 算法的具体逻辑。

class ReActAgent(AgentBase):
    def __call__(self, msg: Union[dict, list, MessageBase]) -> Union[None, MessageBase, list]:
        # 将输入消息添加到记忆中
        self.memory.add(msg)
        # 开始 ReAct 循环
        for _ in range(self.max_steps):
            # 推理步骤：生成思维过程
            thought = self._think()
            # 如果推理得出结论，则返回结果
            if self._halt(thought):
                return thought
            # 行动步骤：执行工具调用
            action_result = self._act()
            # 将行动结果记录到记忆中
            self.memory.add(action_result)
        # 达到最大步数仍未得出结论，返回最后的尝试
        return self._make_response()

2.2 记忆管理系统

记忆管理是 CoPaw 的关键特性之一，采用双层记忆结构：

1. 短期记忆：在单次会话中维护上下文
2. 长期记忆：通过文件系统持久化重要信息

记忆系统受到 ReMe 的启发，具有以下特点： • 自动压缩对话内容以防止上下文窗口溢出 • 将关键信息写入 Markdown 文件进行长期存储 • 提供语义搜索功能以随时检索记忆

2.3 工具选择机制

内置丰富工具集，支持动态扩展，核心工具包括：

2.4 技能管理算法

系统支持动态加载和管理用户自定义的 Python 技能脚本，位于工作目录的 skills 子目录下。技能管理器会扫描这些脚本并将其中定义的函数注册为可调用工具。

三、主要功能函数详解

3.1 CoPawAgent 核心类

这是整个项目的核心入口，整合模型、工具、记忆等能力：

class CoPawAgent(ReActAgent):
    def __init__(
        self,
        name: str = "CoPaw Agent",
        sys_prompt: str = "",
        model_configs: dict = None,
        memory_config: dict = None,
        use_memory_compaction: bool = True,
        **kwargs: Any,
    ) -> None:
        # 初始化模型和格式化器
        model, formatter = create_model_and_formatter(model_configs)
        # 确保技能初始化
        ensure_skills_initialized()
        # 构建系统提示词
        sys_prompt = sys_prompt or build_system_prompt_from_working_dir()
        # 初始化工具包
        toolkit = Toolkit(
            # 内置工具
            browser_use,
            desktop_screenshot,
            execute_shell_command,
            get_current_time,
            read_file,
            write_file,
            edit_file,
            send_file_to_user,
            # 动态加载的技能
            *list_available_skills(),
            # 记忆搜索工具
            create_memory_search_tool(memory_config),
        )
        # 设置记忆管理器
        memory_manager = MemoryManager(memory_config)
        # 添加钩子（如果需要记忆压缩）
        hooks = []
        if use_memory_compaction:
            hooks.append(MemoryCompactionHook(
                ratio=MEMORY_COMPACT_RATIO,
                keep_recent=MEMORY_COMPACT_KEEP_RECENT,
            ))
        super().__init__(
            name=name,
            sys_prompt=sys_prompt,
            model=model,
            use_memory=True,
            memory_config=memory_config,
            toolkit=toolkit,
            hooks=hooks,
            **kwargs,
        )

关键逻辑说明：

• create_model_and_formatter：适配多模型提供商的工厂函数
• ensure_skills_initialized：保证技能目录结构完整性
• MemoryCompactionHook：上下文压缩钩子，防止窗口溢出
• Toolkit：整合所有可调用工具，供 ReActAgent 决策调用

3.2 模型工厂函数

model_factory.py 中的 create_model_and_formatter 函数负责根据配置创建模型实例：

def create_model_and_formatter(model_configs: Optional[dict] = None):
    # 加载配置
    configs = load_config()
    model_config = model_configs or configs.model
    # 根据提供商类型创建模型
    if model_config.get("provider") == "ollama":
        # Ollama 模型配置
        pass
    elif model_config.get("provider") == "local":
        # 本地模型配置
        pass
    # 其他提供商...
    return model, formatter

3.3 技能管理器核心函数

skills_manager.py 负责动态加载和管理用户定义的技能：

def ensure_skills_initialized() -> None:
    """确保技能目录存在并包含基础文件"""
    skills_dir = get_working_skills_dir()
    os.makedirs(skills_dir, exist_ok=True)
    # 创建基础文件
    for filename, content in [
        ("README.md", README_CONTENT),
        ("__init__.py", ""),
        ("SKILL.md", SKILL_TEMPLATE),
    ]:
        filepath = os.path.join(skills_dir, filename)
        if not os.path.exists(filepath):
            with open(filepath, "w", encoding="utf-8") as f:
                f.write(content)

def list_available_skills() -> List[Callable]:
    """列出所有可用的技能"""
    skills_dir = get_working_skills_dir()
    skills = []
    for file_path in Path(skills_dir).rglob("*.py"):
        if file_path.name.startswith("_"):
            continue
        # 动态导入模块并查找可调用对象
        # ...
    return skills

3.4 记忆管理器

CoPaw 的记忆管理器采用分层设计，包括短期记忆和长期记忆两个层面：

短期记忆（上下文窗口管理） • 维护当前对话的上下文 • 管理有限的上下文窗口大小 • 通过记忆压缩技术防止上下文溢出

长期记忆（持久化存储） • 将关键信息持久化存储在文件系统中 • 支持语义搜索和关键词搜索 • 通过异步索引更新保持高效检索

3.4.1 上下文管理

CoPaw 的记忆管理器具备自动压缩对话内容的功能，以防止上下文窗口溢出。这是与其他 AI 助手的重要区别之一。

# 记忆压缩钩子
hooks.append(MemoryCompactionHook(
    ratio=MEMORY_COMPACT_RATIO,
    keep_recent=MEMORY_COMPACT_KEEP_RECENT,
))

3.4.2 长期记忆管理

• 将重要信息写入 Markdown 文件（MEMORY.md）进行长期存储 • 按日期分割日志（daily log），存储在 memory/YYYY-MM-DD.md 格式的文件中 • 支持混合记忆搜索（向量语义搜索+BM25 全文搜索）

graph TB
User[User / Agent] --> MM[MemoryManager]
MM --> ContextMgmt[Context Management]
MM --> MemoryMgmt[Long-term Memory Management]
ContextMgmt --> Summary[Context Compression]
MemoryMgmt --> FileTools[Memory Update]
MemoryMgmt --> Watcher[Memory Index Update]
MemoryMgmt --> SearchLayer[Hybrid Memory Search]
FileTools --> LTM[MEMORY.md]
FileTools --> DailyLog[memory/YYYY-MM-DD.md]
Watcher --> Index[Async Database Update]
SearchLayer --> VectorSearch[Vector Semantic Search]
SearchLayer --> BM25[BM25 Full-text Search]

记忆管理器的工作流程

graph TB
A[用户输入] --> B[短期记忆]
B --> C[上下文管理]
C --> D{是否重要?}
D -->|是| E[写入长期记忆]
D -->|否| F[仅保留在短期记忆]
E --> G[异步索引更新]
G --> H[MEMORY.md]
G --> I[daily log YYYY-MM-DD.md]
J[用户查询] --> K[混合搜索]
K --> L[向量语义搜索]
K --> M[BM25 全文搜索]
L --> N[合并结果]
M --> N
N --> O[返回相关记忆]

3.5 命令行接口

CLI 模块 (cli) 提供了完整的命令行工具，包括： • copaw init：交互式初始化配置 • copaw app：启动主应用程序 • copaw cron：管理定时任务 • copaw skills：管理自定义技能 • copaw channels：配置通信渠道

3.6 通信渠道系统

CoPaw 支持多种通信渠道，包括钉钉、飞书、QQ、Discord 等。每个渠道都是一个独立的模块，通过标准接口与主系统交互。渠道系统允许用户在一个 CoPaw 实例中同时管理多个通信平台。

3.7 定时任务系统

定时任务系统（cron）允许用户创建计划任务，例如： • 每天定时发送消息 • 定期询问 AI 并发送回复 • 执行周期性检查任务 定时任务可以通过 CLI 或 Web API 管理，支持暂停、恢复、删除等操作。