1
┌─────────────────────────────────────────┐
2
│           用户交互层                      │
3
│  ( 多渠道接入：钉钉、飞书、QQ 等 )          │
4
└──────────────┬──────────────────────────┘
5
               │
6
┌──────────────▼──────────────────────────┐
7
│           消息队列层                      │
8
│  ( 消息缓冲、优先级调度、可靠投递 )         │
9
└──────────────┬──────────────────────────┘
10
               │
11
┌──────────────▼──────────────────────────┐
12
│           Agent 核心                     │
13
│  ┌─────────────────────────────────┐   │
14
│  │  Prompt 系统                    │   │
15
│  └─────────────────────────────────┘   │
16
│  ┌─────────────────────────────────┐   │
17
│  │  Hooks 系统                     │   │
18
│  └─────────────────────────────────┘   │
19
│  ┌─────────────────────────────────┐   │
20
│  │  Tools 系统                    │   │
21
│  └─────────────────────────────────┘   │
22
│  ┌─────────────────────────────────┐   │
23
│  │  Memory 系统                   │   │
24
│  └─────────────────────────────────┘   │
25
└──────────────┬──────────────────────────┘
26
               │
27
┌──────────────▼──────────────────────────┐
28
│           模型层                         │
29
│  ( 云端 API、本地模型、推理服务 )          │
30
└─────────────────────────────────────────┘

1
Prompt 系统
2
├── Prompt Loader ( 加载器 )
3
├── Prompt Builder ( 构建器 )
4
├── Prompt Renderer ( 渲染器 )
5
└── Prompt Cache ( 缓存 )

1
# Prompt 加载顺序
2
FILE_ORDER = [
3
    ("AGENTS.md", True),    # 必需，Agent 角色定义
4
    ("SOUL.md", True),      # 必需，核心个性
5
    ("PROFILE.md", False),  # 可选，用户档案
6
    ("SKILLS.md", False),   # 可选，技能定义
7
]

1
# Agent 角色定义
2

3
你是 CoPaw，一个智能 AI 助手。
4

5
## 角色
6
- 个人助理
7
- 信息助手
8
- 创意助手
9

10
## 目标
11
- 帮助用户解决问题
12
- 提供准确信息
13
- 生成创意内容
14

15
## 原则
16
- 友好
17
- 专业
18
- 准确
19
- 有帮助

1
# 核心 Soul
2

3
你是一个温暖、友善的 AI 助手。
4

5
## 性格
6
- 温暖：对用户充满关怀
7
- 友善：总是乐于助人
8
- 专业：提供高质量服务
9
- 幽默：适度使用幽默
10

11
## 价值观
12
- 用户至上
13
- 隐私保护
14
- 持续学习

1
# 用户档案
2

3
## 基本信息
4
姓名：用户
5
语言：中文
6
时区：Asia/Shanghai
7

8
## 偏好
9
- 回答风格：简洁明了
10
- 详细程度：适中
11
- 语气：友好
12

13
## 习惯
14
- 工作时间：9:00-18:00
15
- 休息时间：12:00-13:00

1
class PromptBuilder:
2
    """Prompt 构建器"""
3

4
    def __init__(self, workspace: str):
5
        self.workspace = workspace
6
        self.prompts = {}
7

8
    def load_prompts(self) -> Dict[str, str]:
9
        """加载所有 Prompt 文件"""
10
        for filename, required in FILE_ORDER:
11
            path = os.path.join(self.workspace, filename)
12
            if os.path.exists(path):
13
                with open(path, 'r', encoding='utf-8') as f:
14
                    self.prompts[filename] = f.read()
15
            elif required:
16
                raise FileNotFoundError(f"必需的 Prompt 文件缺失: {filename}")
17
        return self.prompts
18

19
    def build_system_prompt(self) -> str:
20
        """构建系统提示词"""
21
        parts = []
22
        for filename in FILE_ORDER:
23
            if filename[0] in self.prompts:
24
                parts.append(self.prompts[filename[0]])
25
        return "\n\n".join(parts)

1
class PromptRenderer:
2
    """Prompt 渲染器"""
3

4
    def render(self, template: str, variables: Dict[str, Any]) -> str:
5
        """渲染 Prompt"""
6
        from jinja2 import Template
7

8
        # 使用 Jinja2 模板
9
        template_obj = Template(template)
10
        return template_obj.render(**variables)

1
template = """
2
你好 {{ user_name }}！
3

4
今天是 {{ date }}，有什么可以帮你的吗？
5
"""
6

7
variables = {
8
    'user_name': ' 张三 ',
9
    'date': '2026-03-12'
10
}
11

12
result = renderer.render(template, variables)
13
# 输出：你好 张三！
14
# 今天是 2026-03-12，有什么可以帮你的吗？

1
Hooks 系统
2
├── Hook Registry ( 注册表 )
3
├── Hook Executor ( 执行器 )
4
├── Hook Middleware ( 中间件 )
5
└── Hook Context ( 上下文 )

1
from copaw.hooks import Hook, HookContext
2

3
class CustomHook(Hook):
4
    """自定义 Hook"""
5

6
    def __init__(self, config: Dict[str, Any]):
7
        super().__init__(config)
8
        self.enabled = config.get('enabled', True)
9

10
    async def before_llm_call(self, context: HookContext) -> None:
11
        """LLM 调用前"""
12
        if not self.enabled:
13
            return
14

15
        # 自定义逻辑
16
        context.data['custom_value'] = 'custom'
17
        self.log("Pre-LLM hook executed")
18

19
    async def after_llm_call(self, context: HookContext) -> None:
20
        """LLM 调用后"""
21
        if not self.enabled:
22
            return
23

24
        # 自定义逻辑
25
        result = context.data.get('result')
26
        self.log(f"Post-LLM hook executed: {result}")
27

28
    def log(self, message: str) -> None:
29
        """记录日志"""
30
        print(f"[CustomHook] {message}")

1
from copaw.hooks import HookRegistry
2

3
# 注册 Hook
4
registry = HookRegistry()
5
registry.register('custom', CustomHook)
6

7
# 使用 Hook
8
hook = registry.get('custom')
9
await hook.before_llm_call(context)

1
class HookChain:
2
    """Hook 链"""
3

4
    def __init__(self, hooks: List[Hook]):
5
        self.hooks = hooks
6

7
    async def execute_before(self, context: HookContext) -> None:
8
        """执行所有 Pre-Hooks"""
9
        for hook in self.hooks:
10
            await hook.before_llm_call(context)
11

12
    async def execute_after(self, context: HookContext) -> None:
13
        """执行所有 Post-Hooks"""
14
        for hook in reversed(self.hooks):
15
            await hook.after_llm_call(context)

1
Tools 系统
2
├── Tool Registry ( 注册表 )
3
├── Tool Manager ( 管理器 )
4
├── Tool Executor ( 执行器 )
5
└── Tool Cache ( 缓存 )

1
class ReadFileTool(Tool):
2
    """读取文件工具"""
3

4
    name = "read_file"
5
    description = "读取文件内容"
6

7
    async def execute(self, file_path: str) -> str:
8
        """执行工具"""
9
        with open(file_path, 'r', encoding='utf-8') as f:
10
            return f.read()

1
class WebSearchTool(Tool):
2
    """网络搜索工具"""
3

4
    name = "web_search"
5
    description = "搜索网络信息"
6

7
    async def execute(self, query: str) -> List[Dict[str, Any]]:
8
        """执行工具"""
9
        # 调用搜索 API
10
        results = await self.search_api.search(query)
11
        return results

1
class CalculatorTool(Tool):
2
    """计算器工具"""
3

4
    name = "calculator"
5
    description = "数学计算"
6

7
    async def execute(self, expression: str) -> float:
8
        """执行工具"""
9
        # 安全计算
10
        result = eval(expression, {'__builtins__': {}}, {})
11
        return result

1
from copaw.tools import Tool
2

3
class CustomTool(Tool):
4
    """自定义工具"""
5

6
    name = "custom_tool"
7
    description = "自定义工具描述"
8

9
    def __init__(self, config: Dict[str, Any]):
10
        super().__init__(config)
11
        self.param1 = config.get('param1', 'default')
12

13
    async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
14
        """执行工具"""
15
        try:
16
            result = self.process(input_data)
17
            return {
18
                'success': True,
19
                'result': result
20
            }
21
        except Exception as e:
22
            return {
23
                'success': False,
24
                'error': str(e)
25
            }
26

27
    def process(self, input_data: Dict[str, Any]) -> Any:
28
        """处理输入数据"""
29
        # 处理逻辑
30
        return processed_result

1
from copaw.tools import ToolRegistry
2

3
# 注册工具
4
registry = ToolRegistry()
5
registry.register('custom', CustomTool)
6

7
# 使用工具
8
tool = registry.get('custom')
9
result = await tool.execute({'param': 'value'})

1
class ToolChain:
2
    """工具链"""
3

4
    def __init__(self, tools: List[Tool]):
5
        self.tools = tools
6

7
    async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
8
        """执行工具链"""
9
        current_data = input_data
10

11
        for tool in self.tools:
12
            result = await tool.execute(current_data)
13
            if result.get('success'):
14
                current_data = result.get('result')
15
            else:
16
                return {
17
                    'success': False,
18
                    'error': result.get('error'),
19
                    'failed_at': tool.name
20
                }
21

22
        return {
23
            'success': True,
24
            'result': current_data
25
        }

1
Memory 系统
2
├── Memory Manager ( 管理器 )
3
├── Memory Storage ( 存储 )
4
├── Memory Retriever ( 检索器 )
5
└── Memory Index ( 索引 )

1
class FileMemoryStorage:
2
    """文件存储"""
3

4
    def __init__(self, path: str):
5
        self.path = path
6
        os.makedirs(path, exist_ok=True)
7

8
    async def save(self, key: str, value: Any) -> None:
9
        """保存记忆"""
10
        file_path = os.path.join(self.path, f"{key}.json")
11
        with open(file_path, 'w', encoding='utf-8') as f:
12
            json.dump(value, f, ensure_ascii=False)
13

14
    async def load(self, key: str) -> Optional[Any]:
15
        """加载记忆"""
16
        file_path = os.path.join(self.path, f"{key}.json")
17
        if os.path.exists(file_path):
18
            with open(file_path, 'r', encoding='utf-8') as f:
19
                return json.load(f)
20
        return None

1
class DatabaseMemoryStorage:
2
    """数据库存储"""
3

4
    def __init__(self, connection_string: str):
5
        self.connection_string = connection_string
6
        self.pool = await self.create_pool()
7

8
    async def save(self, key: str, value: Any) -> None:
9
        """保存记忆"""
10
        async with self.pool.acquire() as conn:
11
            await conn.execute(
12
                "INSERT INTO memories (key, value) VALUES ($1, $2)",
13
                key, json.dumps(value)
14
            )
15

16
    async def load(self, key: str) -> Optional[Any]:
17
        """加载记忆"""
18
        async with self.pool.acquire() as conn:
19
            record = await conn.fetchrow(
20
                "SELECT value FROM memories WHERE key = $1",
21
                key
22
            )
23
            return json.loads(record['value']) if record else None

1
class SemanticMemoryRetriever:
2
    """语义检索"""
3

4
    def __init__(self, index: Any):
5
        self.index = index
6

7
    async def search(self, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
8
        """语义搜索"""
9
        # 向量化查询
10
        query_vector = self.embed(query)
11

12
        # 搜索最相似的
13
        results = self.index.search(query_vector, top_k)
14

15
        return results
16

17
    def embed(self, text: str) -> List[float]:
18
        """文本向量化"""
19
        # 调用嵌入模型
20
        return self.embedding_model.embed(text)

1
class MemoryConsolidator:
2
    """记忆整合"""
3

4
    async def consolidate(self, memories: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
5
        """整合记忆"""
6
        # 按相似度分组
7
        groups = self.group_by_similarity(memories)
8

9
        # 合并相似记忆
10
        consolidated = []
11
        for group in groups:
12
            if len(group) > 1:
13
                merged = self.merge_group(group)
14
                consolidated.append(merged)
15
            else:
16
                consolidated.append(group[0])
17

18
        return consolidated
19

20
    def group_by_similarity(self, memories: List[Dict[str, Any]]) -> List[List[Dict[str, Any]]]:
21
        """按相似度分组"""
22
        # 实现分组逻辑
23
        pass
24

25
    def merge_group(self, group: List[Dict[str, Any]]) -> Dict[str, Any]:
26
        """合并记忆组"""
27
        # 实现合并逻辑
28
        pass

1
开始
2
  │
3
  ├─→ 接收用户消息
4
  │
5
  ├─→ Pre-Hooks
6
  │
7
  ├─→ 构建 Prompt
8
  │
9
  ├─→ 调用 LLM
10
  │
11
  ├─→ Post-Hooks
12
  │
13
  ├─→ 判断是否需要工具
14
  │   ├─ 是 → 执行工具 → 更新上下文 → 继续
15
  │   └─ 否 → 继续下一步
16
  │
17
  ├─→ 压缩上下文（如果需要）
18
  │
19
  ├─→ 判断是否完成
20
  │   ├─ 否 → 返回构建 Prompt 步骤
21
  │   └─ 是 → 继续下一步
22
  │
23
  └─→ 返回结果

1
class ContextCompactor:
2
    """上下文压缩器"""
3

4
    def __init__(self, config: Dict[str, Any]):
5
        self.threshold = config.get('threshold', 0.8)
6
        self.keep_recent = config.get('keep_recent', 5)
7

8
    async def compact(self, messages: List[Dict[str, Any]], max_length: int) -> List[Dict[str, Any]]:
9
        """压缩上下文"""
10
        current_length = self.calculate_length(messages)
11

12
        if current_length <= max_length:
13
            return messages
14

15
        # 压缩策略
16
        compressed = []
17

18
        # 保留最近的 N 条消息
19
        recent = messages[-self.keep_recent:]
20
        compressed.extend(recent)
21

22
        # 压缩中间部分
23
        middle = messages[self.keep_recent:-self.keep_recent]
24
        if middle:
25
            summary = await self.summarize(middle)
26
            compressed.insert(0, {
27
                'role': 'system',
28
                'content': f"[摘要] {summary}"
29
            })
30

31
        return compressed
32

33
    def calculate_length(self, messages: List[Dict[str, Any]]) -> int:
34
        """计算上下文长度"""
35
        length = 0
36
        for msg in messages:
37
            length += len(msg.get('content', ''))
38
        return length
39

40
    async def summarize(self, messages: List[Dict[str, Any]]) -> str:
41
        """总结消息"""
42
        # 调用 LLM 总结
43
        pass