AI Agent 自动化工作流系统架构与实现

2.2 闭环控制流 (Closed-Loop FSM)

在 core/kernel/mixins/closed_loop.py 中，实现了一个复杂的有限状态机 (FSM)。

• 状态 (LoopState): PENDING, RUNNING, PAUSED, COMPLETED, FAILED
• 阶段 (LoopPhase):
  1. EXECUTE: 执行阶段，调用 Swarm。
  2. INTEGRATE: 整合阶段，调用 Integrator。
  3. VALIDATE: 验证阶段，运行测试。
  4. RESEARCH: (仅在失败时) 调研阶段，调用 Search Agent。
  5. FIX: (仅在失败时) 修复阶段，应用修复方案。
  6. DELIVER: 交付阶段。

核心逻辑:

while True:
    if phase == VALIDATE and score < threshold:
        next_phase = RESEARCH  # 自动进入修复循环
    elif phase == FIX:
        next_phase = EXECUTE   # 修复后重新执行

这种死循环保护机制是系统'自治'的关键。

2.3 知识边界感知 (Metacognition)

在 core/knowledge_boundary.py 中，系统实现了一个简单的元认知模块。

• 功能: 在执行任务前，先问自己'我知道怎么做吗？'
• 输出:
  • Thinking Mode: 'Fast' (直接干) vs 'Slow' (先调研)。
  • Confidence: 置信度分数 (0.0 - 1.0)。
• 实现: 如果置信度低于阈值，EnhancedDecomposer 会自动在任务列表头部插入 TaskType.RESEARCH 类型的任务。

第三章：群体智能编排 (Swarm Orchestration)

3.1 蜂群数据库设计 (Schema Design)

Swarm 不依赖内存状态，而是使用 SQLite (swarm_core.db) 进行持久化。这意味着即使进程崩溃，任务状态也不会丢失。

• 代码位置: core/swarm.py
• 核心表结构:
  • swarm_sessions: 记录整个会话的状态，主任务，子任务总数。
  • tasks: 记录每个原子任务。
    • task_id: UUID
    • parent_id: 关联的 Session
    • worker_type: 指定需要的 Agent 类型 (e.g., 'backend-architect')
    • status: 'pending', 'running', 'completed', 'failed'
    • priority: 优先级 (决定执行顺序)
    • payload: JSON 格式的具体指令

3.2 Worker 协议与通信

Worker 是独立的执行单元。所有的 Worker 都继承自 BaseWorker。

• 通信协议: JSON Payload。
  • Input: {"action": "...", "context": {...}}
  • Output: {"status": "success", "data": {...}}
• 并行机制: SwarmOrchestrator.get_parallel_subtasks() 方法会查询数据库，找出所有 status='pending' 且依赖已满足的任务，一次性返回给主进程并行调用工具执行。

3.3 核心 Worker 解析：Backend Architect

在 core/workers/worker_coder.py 中，CoderWorker (即 Backend Architect) 展现了惊人的细节。

静态分析能力: 它不仅仅是写文件。它有一个 analyze 动作，使用 Python 原生的 ast (Abstract Syntax Tree) 模块解析代码。

tree = ast.parse(data)
analysis = {
    "classes": [node.name for node in ...],
    "functions": [node.name for node in ...],
    "imports": [...]
}

这意味着它在修改代码前，真的'看懂'了代码结构，而不是盲目替换文本。

第四章：长期记忆系统 (Long-Term Memory System)

4.1 记忆的三层架构

在 core/ltm/models.py 和 manager.py 中，系统实现了一个类脑的记忆结构：

1. Episodic Memory (情景记忆): 记录每一次交互的流水账。
   • 包含：时间戳、用户指令、执行结果、情感效价。
2. Semantic Memory (语义记忆): 从情景记忆中抽象出的知识点。
   • 例如：多次在 Python 中使用 FastAPI 成功，就会形成一条关于 FastAPI 的语义记忆。
   • Consolidation (固化): 当某类情景记忆重复出现（阈值默认 3 次），系统会自动将其转化为语义记忆。
3. Procedural Memory (程序记忆): 针对特定任务的最佳实践步骤（SOP）。

4.2 情感计算与重要性评估

这是最令人惊讶的部分。系统会计算'情感'。

• Emotional Valence (情感效价):
  • 成功且高质量的任务 = 正向情感 (+0.7 ~ +1.0)。
  • 失败或严重的错误 = 负向情感 (-0.5 ~ -1.0)。
• Importance Score (重要性):
  • 错误修复类任务、复杂任务会被标记为高重要性。
  • 系统会优先保留高重要性、高情感强度的记忆（模仿人类遗忘机制）。

4.3 错误反向检索生成 (Error RAG)

在 core/memory.py 中实现。

• 工作流:
  1. 遇到报错 -> 生成签名 -> 存入 errors/ 目录。
  2. 修复报错 -> 记录修复方案 -> 关联到该签名。
  3. 下次遇到同签名错误 -> 直接读取修复方案 -> 自动修复。

Error Signature (错误签名):

def _extract_error_signature(self, error_message):
    # 去除行号和文件路径的差异，只保留核心堆栈信息
    clean = re.sub(r'line \d+', 'line N', error_message)
    return md5(clean)

第五章：智能决策与路由 (Intelligence & Routing)

5.1 5 场景决策树 (5-Scenario Decision Tree)

core/scenario_selector.py 定义了系统如何面对不同难度的任务。这避免了'杀鸡用牛刀'。

1. Prompt Enhancement (复杂度 1-2): 简单的改名、润色。不启动 Swarm，直接 LLM 返回。
2. Skill Reuse (复杂度 3-5 + 有 Skill): 发现有现成工具（如 search），直接调用。
3. Plan + Review (复杂度 3-5 + 无 Skill): 默认模式。先生成计划，用户确认，执行，最后 Review。
4. Lead-Member (复杂度 6-10): 复杂的项目开发。指定一个 Leader Agent 负责协调，多个 Member Agent 并行开发。
5. Composite (复杂度 10+): 复合模式，任务套任务。

5.2 增强型任务拆解 (Enhanced Decomposition)

core/enhanced_decomposer.py 是任务的'粉碎机'。

• Prompt Patterns: 使用正则表达式匹配用户意图（如 r"创建 | 新建 | 编写" -> TaskType.CODE_WRITE）。
• Agent Mapping: 自动将任务类型映射到最合适的 Agent（如 CODE_WRITE -> backend-architect, RESEARCH -> search）。
• Dependency Graph: 构建任务依赖图，进行拓扑排序，确保执行顺序正确。

5.3 Skill 自动发现机制

core/skill_discovery.py 允许系统扩展能力。

• 它会扫描 .trae/skills/ 目录下的 skill.yaml。
• 如果发现本地安装了新 Skill（比如用户刚写了一个 NovelWriter），它会自动将其注册到能力列表中。
• 支持 Fallback Chain: Local Skill -> Built-in Agent -> General LLM。

第六章：技术栈与关键算法 (Tech Stack & Algorithms)

6.1 核心技术栈

• Language: Python 3.9+ (大量使用了 Type Hints, Dataclasses, Enum)。
• Storage:
  • SQLite: 结构化数据 (Swarm State)。
  • JSON: 配置文件与中间产物。
  • Markdown: 记忆存储 (便于人类阅读和 LLM 理解)。
• Code Analysis: ast (Python Abstract Syntax Tree)。
• Concurrency: 基于 subprocess 和数据库状态锁的伪并行（更安全，易于调试）。

6.2 关键算法实现

1. 拓扑排序 (Topological Sort): 用于任务依赖解析，确保子任务执行顺序。
2. MD5 Hashing: 用于生成一致性的错误签名和任务 ID。
3. 加权评分算法:
   • 在 Integrator 中计算 quality_score。
   • 在 LTM 中计算 importance_score。
   • 公式示例：Score = Base + Weight_A * Factor_A + Weight_B * Factor_B。

第七章：整合与交付 (Integration & Delivery)

7.1 冲突检测与合并策略

core/integrator.py 是最后一道防线。

• 它定义了 ConflictType: DUPLICATE_CONTENT, INCONSISTENT_INTERFACE 等。
• 合并策略 (Integration Rules):
  • Code: merge_with_imports (智能合并 import 语句)。
  • Docs: concatenate_with_toc (生成目录并拼接)。
  • Config: deep_merge (递归合并 JSON/YAML)。

7.2 质量评分体系

系统不会盲目交付。它会计算一个 Quality Score。

• 如果 Score < 0.7 (阈值)，系统会将状态标记为 FAILED，这会触发 KernelClosedLoopMixin 进入 RESEARCH -> FIX 流程。
• 评分维度包括：冲突数量、测试通过率、代码规范性（通过 AST 分析）。