LingFlow v1.1.0 核心业务流程

版本: v1.1.0 日期: 2026-03-17

核心架构

┌─────────────────────────────────────────────────────────────┐
│                     LingFlow 系统                          │
└─────────────────────────────────────────────────────────────┘
                              │
                              ├── 技能库 (10 个技能)
                              ├── 代理协调器 (6 个代理)
                              ├── 测试引擎 (3 种引擎)
                              └── 上下文压缩器

核心流程

流程 1: 代理协调流程

┌─────────────────┐
│  任务提交      │
│  Task.submit() │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  注册表查询    │
│  find_agent()  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  上下文压缩    │
│  compress()    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  代理执行      │
│  execute()     │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  结果返回      │
│  TaskResult    │
└─────────────────┘

流程 2: 并行执行流程

┌─────────────────┐
│  批量任务      │
│  Task[]        │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  依赖解析      │
│  resolve_deps() │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  调度排序      │
│  schedule()    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  并行分配      │
│  distribute()  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  并发执行      │
│  asyncio.gather│
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  结果聚合      │
│  aggregate()   │
└─────────────────┘

流程 3: 上下文压缩流程

┌─────────────────┐
│  原始上下文    │
│  Context(dict) │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  密度分析      │
│  analyze()     │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  信息过滤      │
│  filter()      │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  语义压缩      │
│  semantic()    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  列表优化      │
│  optimize()    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  压缩上下文    │
│  Compressed()   │
└─────────────────┘

核心类和函数

1. AgentCoordinator

核心职责: - 任务调度和执行 - 代理选择和分配 - 状态监控和管理

核心方法:

submit_task(task)           # 提交任务
schedule_tasks()            # 调度任务
execute_parallel(tasks)      # 并行执行
execute_workflow(tasks)      # 执行工作流
get_status()                # 获取状态

2. AgentRegistry

核心职责: - 代理注册和发现 - 能力匹配查询 - 代理状态管理

核心方法:

register_agent(agent)       # 注册代理
find_agents_for_task(task)  # 查找代理
get_agent(name)            # 获取代理
list_agents()              # 列出代理

3. ContextCompressor

核心职责: - 上下文压缩 - 信息密度分析 - 压缩统计

核心方法:

compress(context)          # 压缩上下文
estimate_tokens(text)     # 估算 token
get_stats()                # 获取统计

数据流

任务执行数据流

Task (定义)
  ↓
Task + Context (压缩后)
  ↓
Agent (执行)
  ↓
TaskResult (结果)
  ↓
CompletedTasks (存储)

并行执行数据流

Task[] (多个任务)
  ↓
ScheduleTasks[] (已调度)
  ↓
ExecuteParallel[] (并行执行)
  ↓
TaskResult[] (多个结果)
  ↓
Results (聚合结果)

关键决策点

1. 代理选择

决策因素: - 任务类型匹配 - 代理能力匹配 - 代理可用性

2. 任务调度

决策因素: - 依赖关系 - 任务优先级 - 可用资源

3. 上下文压缩

决策因素: - 上下文大小 - 压缩阈值 - 信息重要性

错误处理流程

任务执行失败

Task.execute() 失败
  ↓
捕获异常
  ↓
返回 TaskResult(success=False, error=...)
  ↓
记录到 failed_tasks
  ↓
继续执行其他任务

无可用代理

find_agents_for_task() 返回空列表
  ↓
返回 TaskResult(success=False, error="No suitable agent")
  ↓
记录到 failed_tasks
  ↓
跳过该任务

依赖未满足

schedule_tasks() 发现依赖未满足
  ↓
跳过该任务
  ↓
等待依赖完成
  ↓
下一轮调度重新尝试

性能优化点

1. 并行执行

优化: - 使用 asyncio 实现真正的并发 - 可配置的并行度 - 信号量控制资源

2. 上下文压缩

优化: - 减少传输 token 数 - 保持关键信息 - 可配置压缩级别

3. 缓存机制

优化: - 代理查找缓存 - 压缩结果缓存 - 状态查询缓存

监控和日志

关键指标

执行指标
任务成功率
平均执行时间
并行效率
资源指标
代理使用率
Token 消耗
内存使用
压缩指标
压缩率
Token 节省
信息保留率

日志级别

INFO: 正常流程
WARNING: 非致命错误
ERROR: 任务失败
DEBUG: 详细调试信息

集成点

1. 技能集成

dispatching-parallel-agents 技能使用协调器
subagent-driven-development 技能调用并行执行

2. 测试引擎集成

任务完成后运行验证测试
集成三种测试引擎

3. 文档系统

自动生成执行报告
记录压缩统计
保存任务结果

扩展点

1. 自定义代理

可以通过继承 Agent 类创建自定义代理：

class CustomAgent(Agent):
    def can_execute(self, task):
        # 自定义能力匹配逻辑
        pass

    async def execute_task(self, task, context):
        # 自定义执行逻辑
        pass

2. 自定义压缩器

可以通过继承 ContextCompressor 创建自定义压缩器：

class CustomCompressor(ContextCompressor):
    def compress(self, context):
        # 自定义压缩逻辑
        pass

3. 自定义调度器

可以通过重写 schedule_tasks() 方法实现自定义调度：

class CustomCoordinator(AgentCoordinator):
    def schedule_tasks(self):
        # 自定义调度逻辑
        pass

安全考虑

1. 资源限制

最大并行度限制
任务超时机制
内存使用监控

2. 错误隔离

单个任务失败不影响其他任务
代理错误不导致系统崩溃
异常捕获和优雅处理

3. 权限控制

代理访问控制
任务执行权限
敏感信息保护

总结

LingFlow v1.1.0 的核心业务流程围绕三个主要组件：

AgentCoordinator: 任务调度和执行
AgentRegistry: 代理管理和发现
ContextCompressor: 上下文优化

通过这些组件的协同工作，实现了高效、可靠的多代理任务执行系统。

文档版本: 1.0 最后更新: 2026-03-17