国产 IM 适配版，基于 AgentScope 框架，深度支持微信、飞书、钉钉等国内主流即时通讯平台。

核心特性

统一消息总线

• 采用适配器模式统一消息格式
• 不同平台消息自动转换
• 统一的 API 接口
• 消息队列管理

深度平台集成

• 微信个人号/企业微信支持
• 飞书机器人深度集成
• 钉钉企业应用对接
• 支持富文本、图片、文件等多种消息类型

插件热重载

• 配置修改无需重启
• 动态加载新插件
• 插件版本管理
• 热更新机制

企业级特性

• 多租户支持
• 权限管理
• 消息审计
• 数据加密

快速开始

前置要求

• Python 3.9+
• AgentScope 框架
• 对应平台的开发者账号

安装

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

# 安装依赖
pip install -r requirements.txt

# 安装 AgentScope
pip install agentscope

配置

创建 config.yaml:

# 基础配置
app:
  name: "CoPaw"
  log_level: "INFO"

# LLM 配置
llm:
  provider: "anthropic"
  api_key: "${ANTHROPIC_API_KEY}"
  model: "claude-3-5-sonnet-20241022"

# 平台配置
platforms:
  # 微信配置
  wechat:
    enabled: true
    type: "personal"  # 或 "enterprise"
    app_id: "${WECHAT_APP_ID}"
    app_secret: "${WECHAT_APP_SECRET}"
    
  # 飞书配置
  feishu:
    enabled: true
    app_id: "${FEISHU_APP_ID}"
    app_secret: "${FEISHU_APP_SECRET}"
    verification_token: "${FEISHU_VERIFICATION_TOKEN}"
    encrypt_key: "${FEISHU_ENCRYPT_KEY}"
    
  # 钉钉配置
  dingtalk:
    enabled: true
    app_key: "${DINGTALK_APP_KEY}"
    app_secret: "${DINGTALK_APP_SECRET}"
    agent_id: "${DINGTALK_AGENT_ID}"

# 消息处理
message:
  queue_size: 1000
  timeout: 30
  retry: 3

启动

# 启动 CoPaw
python main.py --config config.yaml

# 后台运行
nohup python main.py --config config.yaml > copaw.log 2>&1 &

技术架构

核心架构

┌─────────────────────────────────────────┐
│           CoPaw Core                    │
│                                         │
│  ┌───────────────────────────────────┐  │
│  │   AgentScope Framework            │  │
│  └───────────────────────────────────┘  │
│              ↓                          │
│  ┌───────────────────────────────────┐  │
│  │   Message Bus (消息总线)          │  │
│  │  - Adapter Pattern                │  │
│  │  - Message Queue                  │  │
│  │  - Format Converter               │  │
│  └───────────────────────────────────┘  │
│              ↓                          │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐ │
│  │ WeChat  │  │ Feishu  │  │DingTalk │ │
│  │ Adapter │  │ Adapter │  │ Adapter │ │
│  └─────────┘  └─────────┘  └─────────┘ │
└─────────────────────────────────────────┘

核心技术栈

• 语言: Python 3.9+
• 框架: AgentScope
• 消息队列: Redis (可选)
• 数据库: SQLite / PostgreSQL

适配器模式

每个平台都有独立的适配器：

from agentscope.message import Msg
from copaw.adapters import BaseAdapter

class WeChatAdapter(BaseAdapter):
    def __init__(self, config):
        super().__init__(config)
        self.client = self._init_client()
    
    def receive(self) -> Msg:
        """接收消息并转换为统一格式"""
        raw_msg = self.client.get_message()
        return self._convert_to_msg(raw_msg)
    
    def send(self, msg: Msg):
        """将统一格式消息发送到微信"""
        raw_msg = self._convert_from_msg(msg)
        self.client.send_message(raw_msg)

使用示例

基本使用

from copaw import CoPaw
from agentscope.agents import DialogAgent

# 初始化 CoPaw
copaw = CoPaw(config_path="config.yaml")

# 创建 Agent
agent = DialogAgent(
    name="助手",
    sys_prompt="你是一个友好的AI助手",
    model_config_name="claude"
)

# 启动服务
copaw.run(agent)

多平台同时运行

# 同时监听多个平台
copaw = CoPaw(config_path="config.yaml")

# 为不同平台设置不同的 Agent
copaw.register_agent("wechat", wechat_agent)
copaw.register_agent("feishu", feishu_agent)
copaw.register_agent("dingtalk", dingtalk_agent)

# 启动
copaw.run_all()

自定义消息处理

from copaw.handlers import MessageHandler

class CustomHandler(MessageHandler):
    def handle_text(self, msg):
        """处理文本消息"""
        if msg.content.startswith("/"):
            return self.handle_command(msg)
        return self.agent(msg)
    
    def handle_image(self, msg):
        """处理图片消息"""
        # 图片识别逻辑
        pass
    
    def handle_file(self, msg):
        """处理文件消息"""
        # 文件处理逻辑
        pass

# 注册自定义处理器
copaw.register_handler(CustomHandler())

平台配置指南

微信配置

1. 个人微信（基于 itchat）：

# 安装依赖
pip install itchat

# 首次运行会弹出二维码，扫码登录
python main.py --platform wechat

2. 企业微信：

• 登录企业微信管理后台
• 创建应用获取 App ID 和 Secret
• 配置回调 URL

飞书配置

1. 创建飞书应用：

   • 访问 飞书开放平台
   • 创建企业自建应用
   • 获取 App ID 和 App Secret

2. 配置权限：

   • 开启机器人能力
   • 添加消息接收权限
   • 配置事件订阅

3. 配置回调：

# 在 config.yaml 中配置
feishu:
  callback_url: "https://your-domain.com/feishu/callback"
  verification_token: "your_token"
  encrypt_key: "your_key"

钉钉配置

1. 创建钉钉应用：

   • 登录钉钉开放平台
   • 创建企业内部应用
   • 获取 AppKey 和 AppSecret

2. 配置机器人：

   • 开启机器人功能
   • 配置消息接收地址
   • 设置 IP 白名单

高级功能

消息路由

# 根据消息内容路由到不同的 Agent
from copaw.router import MessageRouter

router = MessageRouter()

# 技术问题路由到技术 Agent
router.add_rule(
    pattern=r"技术|代码|bug",
    agent=tech_agent
)

# 业务问题路由到业务 Agent
router.add_rule(
    pattern=r"业务|需求|功能",
    agent=business_agent
)

copaw.set_router(router)

消息过滤

from copaw.filters import MessageFilter

# 过滤敏感词
filter = MessageFilter()
filter.add_blacklist(["敏感词1", "敏感词2"])

# 只处理特定用户的消息
filter.add_whitelist_user(["user_id_1", "user_id_2"])

copaw.add_filter(filter)

消息审计

# 启用消息审计
copaw = CoPaw(
    config_path="config.yaml",
    enable_audit=True,
    audit_path="/var/log/copaw/audit"
)

# 审计日志包含：
# - 消息时间
# - 发送者信息
# - 消息内容
# - AI 回复
# - 处理时长

性能指标

指标	数值
消息处理延迟	< 500ms
并发连接数	1000+
消息吞吐量	10000 msg/min
内存占用	< 200MB

适用场景

推荐使用

• 国内企业内部使用
• 需要接入飞书、钉钉的场景
• 微信个人号自动化
• 多平台统一管理
• 企业客服机器人

不推荐使用

• 需要国际化平台支持（Slack、Teams等）
• 对消息实时性要求极高（< 100ms）
• 需要处理大量多媒体内容

常见问题

Q: 微信个人号会被封号吗？

A: 使用 itchat 存在一定风险，建议：

• 不要频繁发送消息
• 避免群发广告
• 使用小号测试
• 企业使用建议用企业微信

Q: 如何处理消息并发？

A: CoPaw 内置消息队列，可以配置：

message:
  queue_size: 1000
  workers: 4  # 并发处理数

Q: 支持语音消息吗？

A: 支持，需要额外配置语音识别服务：

speech:
  provider: "aliyun"  # 或 "tencent"
  api_key: "${SPEECH_API_KEY}"

相关链接

• GitHub: agentscope-ai/CoPaw
• AgentScope: AgentScope 文档
• 问题反馈: Issues

贡献

欢迎贡献代码、报告问题或提出建议！

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

# 安装开发依赖
cd CoPaw
pip install -r requirements-dev.txt

# 运行测试
pytest tests/

# 运行示例
python examples/wechat_bot.py

许可证

Apache License 2.0 - 详见 LICENSE

---

访问 GitHub 仓库 →

上一页ClawWork - AI 外包工作者

下一页FastClaw - K8s 多租户平台