📚 SOP-First

SOP-First 是 VTeamOS 的核心理念，通过流程知识驱动 Agent 的行为，而非硬编码的行为规则。

🎯 什么是 SOP

定义

SOP（Standard Operating Procedure） = 流程知识

它告诉 Agent：

✅ 遇到什么情况
✅ 按什么流程走
✅ 用什么工具
✅ 怎么验证结果

SOP vs 规则

维度	传统规则	SOP-First
本质	行为约束	能力导航
灵活性	硬编码	知识驱动
进化	手动修改	自动结晶
复用	低	高

🔄 SOP 工作流

基本流程

用户需求
    ↓
任务分析
    ↓
sop_search("相关 SOP")
    ↓
匹配到 SOP？
    ├─ 是 → 加载 SOP → 按流程执行
    └─ 否 → 自由探索 → 记录成功路径 → SOP 结晶

详细示例

用户: "帮我查一下今天的天气"

1. Agent 分析需求
   → 意图: 天气查询
   → 实体: 今天、天气

2. Agent 搜索 SOP
   → sop_search("天气查询")
   → 匹配到: weather_query_sop

3. Agent 加载 SOP
   → skill_view("weather")
   → 获取: 天气查询技能

4. Agent 按 SOP 执行
   → 获取用户位置
   → 调用天气 API
   → 格式化结果

5. Agent 返回结果
   → "今天天气晴朗，气温 25°C"

📝 SOP 结构

SOP 定义

yaml

# weather_query_sop.yaml
name: "weather_query"
description: "查询天气信息的标准流程"
version: "1.0.0"
triggers:
  - "天气"
  - "温度"
  - "weather"
  - "temperature"

steps:
  - id: 1
    name: "获取位置"
    tool: "memory_recall"
    input:
      query: "用户位置"
    output: "user_location"
    on_failure: "ask_user"

  - id: 2
    name: "调用天气 API"
    tool: "web"
    input:
      url: "https://api.weather.com/..."
      params:
        location: "{{user_location}}"
    output: "weather_data"

  - id: 3
    name: "格式化结果"
    tool: "shell"
    input:
      command: "format_weather {{weather_data}}"
    output: "formatted_result"

validation:
  - check: "weather_data.temp != null"
    error: "无法获取温度数据"
  - check: "formatted_result.length > 0"
    error: "格式化失败"

success:
  reply: "{{formatted_result}}"
  
failure:
  reply: "抱歉，无法获取天气信息，请稍后重试"
  suggest: "检查网络连接或稍后重试"

🔍 SOP 发现

sop_search 工具

// sop_search 工具定义
type SOPSearchTool struct {
    sopStore *SOPStore
}

func (t *SOPSearchTool) Execute(ctx context.Context, input map[string]any) (map[string]any, error) {
    query := input["query"].(string)
    
    // 搜索匹配的 SOP
    sops := t.sopStore.Search(query)
    
    // 按相关性排序
    sort.Slice(sops, func(i, j int) bool {
        return sops[i].Score > sops[j].Score
    })
    
    return map[string]any{
        "sops": sops,
        "count": len(sops),
    }, nil
}

匹配算法

// SOP 匹配算法
func (s *SOPStore) Search(query string) []*SOP {
    results := make([]*SOP, 0)
    
    for _, sop := range s.sops {
        score := 0.0
        
        // 1. 关键词匹配
        for _, trigger := range sop.Triggers {
            if strings.Contains(query, trigger) {
                score += 10.0
            }
        }
        
        // 2. 语义相似度
        similarity := s.calculateSimilarity(query, sop.Description)
        score += similarity * 5.0
        
        // 3. 历史成功率
        score += sop.SuccessRate * 2.0
        
        // 4. 最近使用
        if time.Since(sop.LastUsed) < 24*time.Hour {
            score += 1.0
        }
        
        if score > 0 {
            sop.Score = score
            results = append(results, sop)
        }
    }
    
    return results
}

🧪 SOP 结晶

SOPCrystallizer

SOPCrystallizer 负责从成功执行中自动生成 SOP：

// SOP 结晶器
type SOPCrystallizer struct {
    traceStore  *TraceStore
    sopStore    *SOPStore
    llm         LLMProvider
}

func (c *SOPCrystallizer) Crystallize(ctx context.Context, taskID string) (*SOP, error) {
    // 1. 获取执行轨迹
    trace, err := c.traceStore.Get(taskID)
    if err != nil {
        return nil, err
    }
    
    // 2. 检查是否成功
    if !trace.Success {
        return nil, fmt.Errorf("task failed, cannot crystallize")
    }
    
    // 3. 分析轨迹
    analysis := c.analyzeTrace(trace)
    
    // 4. 生成 SOP
    sop, err := c.generateSOP(ctx, analysis)
    if err != nil {
        return nil, err
    }
    
    // 5. 保存 SOP
    if err := c.sopStore.Save(sop); err != nil {
        return nil, err
    }
    
    return sop, nil
}

结晶条件

// 结晶条件检查
func (c *SOPCrystallizer) shouldCrystallize(trace *Trace) bool {
    // 1. 任务成功
    if !trace.Success {
        return false
    }
    
    // 2. 执行步骤 ≥ 3
    if len(trace.Steps) < 3 {
        return false
    }
    
    // 3. 总耗时 < 5 分钟
    if trace.TotalDuration > 5*time.Minute {
        return false
    }
    
    // 4. 没有已存在的类似 SOP
    similar := c.sopStore.FindSimilar(trace)
    if len(similar) > 0 {
        return false
    }
    
    return true
}

📊 SOP 优化

A/B 测试

// SOP A/B 测试
type SOPExperiment struct {
    ID          string
    SOP_A       *SOP
    SOP_B       *SOP
    Traffic     float64  // 流量分配比例
    Metric      string   // 评估指标
    Duration    time.Duration
}

func (e *SOPExperiment) SelectSOP() *SOP {
    if rand.Float64() < e.Traffic {
        return e.SOP_A
    }
    return e.SOP_B
}

func (e *SOPExperiment) Evaluate() *ExperimentResult {
    // 收集数据
    dataA := e.collectData(e.SOP_A)
    dataB := e.collectData(e.SOP_B)
    
    // 统计分析
    result := e.analyze(dataA, dataB)
    
    return result
}

效果评估

// SOP 效果评估
type SOPEvaluation struct {
    SuccessRate     float64  // 成功率
    AvgDuration     time.Duration  // 平均耗时
    UserSatisfaction float64  // 用户满意度
    TokenUsage      int  // Token 使用量
}

func (s *SOPStore) Evaluate(sop *SOP) *SOPEvaluation {
    traces := s.traceStore.GetBySOP(sop.ID)
    
    eval := &SOPEvaluation{}
    
    for _, trace := range traces {
        if trace.Success {
            eval.SuccessRate++
        }
        eval.AvgDuration += trace.TotalDuration
        eval.TokenUsage += trace.TotalTokens
    }
    
    eval.SuccessRate /= float64(len(traces))
    eval.AvgDuration /= time.Duration(len(traces))
    
    return eval
}

🔧 SOP 管理

SOP 存储

// SOP 存储接口
type SOPStore interface {
    Save(sop *SOP) error
    Get(id string) (*SOP, error)
    Search(query string) []*SOP
    List() []*SOP
    Delete(id string) error
    Update(sop *SOP) error
}

SOP 版本控制

// SOP 版本
type SOPVersion struct {
    Version     string
    SOP         *SOP
    CreatedAt   time.Time
    CreatedBy   string
    ChangeLog   string
}

type SOPVersionStore interface {
    CreateVersion(sop *SOP, changeLog string) (*SOPVersion, error)
    GetVersion(sopID, version string) (*SOPVersion, error)
    ListVersions(sopID string) ([]*SOPVersion, error)
    Rollback(sopID, version string) error
}

💡 最佳实践

1. SOP 设计原则

✅ 单一职责: 每个 SOP 只做一件事
✅ 可测试: SOP 应该有明确的验证条件
✅ 可复用: SOP 应该足够通用
✅ 可进化: SOP 应该支持版本更新

2. SOP 编写规范

yaml

# 好的 SOP
name: "query_weather"
description: "查询指定城市的天气信息"
triggers: ["天气", "温度", "weather"]
steps:
  - name: "获取城市"
    tool: "memory_recall"
    input: {"query": "用户所在城市"}
  - name: "查询天气"
    tool: "web"
    input: {"url": "https://api.weather.com/{{city}}"}
validation:
  - check: "result.temp != null"

3. SOP 优化策略

📊 数据驱动: 基于执行数据优化 SOP
🔄 持续迭代: 定期更新和改进 SOP
🧪 A/B 测试: 验证优化效果
📈 监控指标: 跟踪成功率和用户满意度

🚀 快速开始

1. 创建 SOP

bash

# 创建 SOP 文件
cat > my_sop.yaml << EOF
name: "my_first_sop"
description: "我的第一个 SOP"
triggers: ["测试", "test"]
steps:
  - name: "执行测试"
    tool: "shell"
    input: {"command": "echo 'Hello, SOP!'"}
EOF

2. 加载 SOP

sopStore := sop.NewStore("./sops")
if err := sopStore.LoadFromFile("my_sop.yaml"); err != nil {
    log.Fatal(err)
}

3. 搜索并执行 SOP

// 搜索 SOP
results := sopStore.Search("测试")
if len(results) > 0 {
    sop := results[0]
    
    // 执行 SOP
    result, err := executor.Execute(ctx, sop, input)
    if err != nil {
        log.Fatal(err)
    }
    
    fmt.Println(result)
}

📚 下一步

AgentCore - 深入了解核心引擎
系统架构 - 整体架构设计
团队成员 - 认识主创团队

📚 SOP-First ​

🎯 什么是 SOP ​

定义 ​

SOP vs 规则 ​

🔄 SOP 工作流 ​

基本流程 ​

详细示例 ​

📝 SOP 结构 ​

SOP 定义 ​

🔍 SOP 发现 ​

sop_search 工具 ​

匹配算法 ​

🧪 SOP 结晶 ​

SOPCrystallizer ​

结晶条件 ​

📊 SOP 优化 ​

A/B 测试 ​

效果评估 ​

🔧 SOP 管理 ​

SOP 存储 ​

SOP 版本控制 ​

💡 最佳实践 ​

1. SOP 设计原则 ​

2. SOP 编写规范 ​

3. SOP 优化策略 ​

🚀 快速开始 ​

1. 创建 SOP ​

2. 加载 SOP ​

3. 搜索并执行 SOP ​

📚 下一步 ​

📚 SOP-First

🎯 什么是 SOP

定义

SOP vs 规则

🔄 SOP 工作流

基本流程

详细示例

📝 SOP 结构

SOP 定义

🔍 SOP 发现

sop_search 工具

匹配算法

🧪 SOP 结晶

SOPCrystallizer

结晶条件

📊 SOP 优化

A/B 测试

效果评估

🔧 SOP 管理

SOP 存储

SOP 版本控制

💡 最佳实践

1. SOP 设计原则

2. SOP 编写规范

3. SOP 优化策略

🚀 快速开始

1. 创建 SOP

2. 加载 SOP

3. 搜索并执行 SOP

📚 下一步