【独家】炸场！阿里开源桌面 Agent「CoPaw」，源码深度拆解

文章重点从从代码架构、核心算法、主要功能函数解析三方面拆解阿里开源桌面Agent 助手CoPaw

liangtongtongqq2

1077人浏览 · 2026-03-03 18:28:54

liangtongtongqq2 · 2026-03-03 18:28:54 发布

CoPaw 是基于 AgentScope 框架构建的可扩展个人 AI 助手系统，采用分层架构设计，具备记忆管理、多工具调用、多渠道接入等核心能力。本文将从代码架构、核心算法、功能模块三个维度对项目进行全面解析。

开源代码链接：

GitHub - agentscope-ai/CoPaw: Your Personal AI Assistant; easy to install, deploy on your own machine or on the cloud; supports multiple chat apps with easily extensible capabilities.

一、代码架构分析

CoPaw 采用模块化分层架构，各模块职责清晰、低耦合高内聚，便于扩展和维护。

1.1 整体架构

Agents模块

CoPawAgent是CoPaw系统的核心类，继承自AgentScope的ReActAgent。该类实现了AI代理的基本行为，包括推理和行动。
主要功能：
• 初始化模型和格式化器
• 加载并整合所有可用技能
• 管理工具包和记忆系统
• 支持钩子机制，如记忆压缩

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,
        )

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[返回相关记忆]