第十六章：竞赛实战——从零搭建多 Agent 系统

// 伪代码：数据分析流程图
graph.AddNode("parse_requirement", parseRequirementNode)
graph.AddNode("query_data", queryDataNode)
graph.AddNode("validate_data", validateDataNode)  // 数据质量检查
graph.AddNode("analyze", analyzeNode)
graph.AddNode("generate_report", reportNode)

// 条件路由：数据质量不达标 → 重新查询
graph.AddConditionalEdge("validate_data", func(state State) string {
    if state.DataQuality < 0.8 {
        return "query_data"  // 循环回去重新查
    }
    return "analyze"
})

// cmd/hello/main.go
package main

import (
    "context"
    "fmt"
    "log"

    "trpc.group/trpc-go/trpc-agent-go/agent"
    "trpc.group/trpc-go/trpc-agent-go/runner"
)

func main() {
    // 1. 创建一个最简单的 Agent
    helloAgent := agent.NewAgent(
        agent.WithName("HelloAgent"),
        agent.WithSystemPrompt("你是一个友好的助手，用中文回复。"),
        agent.WithModel("gpt-4o-mini"), // 开发阶段用便宜的模型
    )

    // 2. 创建 Runner（Agent 的运行容器）
    r := runner.NewRunner(
        runner.WithAgent(helloAgent),
    )

    // 3. 运行一次请求
    ctx := context.Background()
    result, err := r.Run(ctx, "你好！请介绍一下你自己。")
    if err != nil {
        log.Fatalf("运行失败: %v", err)
    }

    fmt.Println("Agent 回复:", result.Content)
}

// internal/workflow/pipeline.go
package workflow

import (
    "context"
    "trpc.group/trpc-go/trpc-agent-go/team"
    "trpc.group/trpc-go/trpc-agent-go/agent"
    
    "github.com/your-team/devops-agent/internal/agents"
    "github.com/your-team/devops-agent/internal/tools"
)

// BuildPipeline 构建完整的 DevOps 流水线 Team
func BuildPipeline() *team.Team {
    // 创建各个专业 Agent
    reviewer := agents.NewCodeReviewer()
    testgen := agents.NewTestGenerator()
    deployer := agents.NewDeployer()
    
    // 创建协调者
    coordinator := agents.NewPipelineManager()
    
    // 组装 Team
    pipeline := team.NewTeam(
        team.WithMode(team.ModeCoordinator),
        team.WithCoordinator(coordinator),
        team.WithMembers(reviewer, testgen, deployer),
        team.WithMaxRounds(10), // 最多 10 轮交互
    )
    
    return pipeline
}

// internal/agents/reviewer.go
package agents

import (
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "github.com/your-team/devops-agent/internal/tools"
)

const reviewerPrompt = `你是一位资深代码审查员。你的职责是审查 Pull Request 中的代码变更。

## 你的能力
- 分析代码差异（git diff）
- 检查代码风格是否符合团队规范
- 识别潜在的安全问题
- 发现性能瓶颈
- 验证逻辑正确性

## 审查维度（按优先级）
1. 安全：SQL注入、XSS、硬编码密钥、未验证的输入
2. 逻辑：边界条件、空指针、并发安全、资源泄漏
3. 性能：N+1查询、无必要的内存分配、缺少缓存
4. 风格：命名规范、代码组织、注释质量

## 输出格式
你必须以 JSON 格式输出审查结果：
{
  "verdict": "approve" | "request_changes" | "reject",
  "summary": "一句话总结",
  "issues": [
    {
      "severity": "critical" | "major" | "minor" | "suggestion",
      "file": "文件路径",
      "line": 行号,
      "description": "问题描述",
      "suggestion": "修改建议"
    }
  ],
  "score": 0-100
}

## 重要约束
- 你只负责审查，不负责修改代码
- 没有 critical 或 major 问题时应该 approve
- 不要吹毛求疵——minor 问题不应阻止合入
- 如果代码变更太大（>500行），建议拆分 PR`

// NewCodeReviewer 创建代码审查 Agent
func NewCodeReviewer() *agent.Agent {
    return agent.NewAgent(
        agent.WithName("CodeReviewer"),
        agent.WithSystemPrompt(reviewerPrompt),
        agent.WithModel("gpt-4o"),  // 审查需要强推理能力
        agent.WithTools(
            tools.GitDiff(),       // 获取代码差异
            tools.FileRead(),      // 读取完整文件
            tools.ASTAnalyze(),    // 语法树分析
            tools.StyleCheck(),    // 风格检查
            tools.SecurityScan(),  // 安全扫描
        ),
        agent.WithTemperature(0.1), // 审查需要确定性
    )
}

// internal/tools/git.go
package tools

import (
    "context"
    "encoding/json"
    "fmt"
    "os/exec"
    "strings"
    
    "trpc.group/trpc-go/trpc-agent-go/tool"
)

// GitDiff 创建获取 Git 差异的工具
func GitDiff() *tool.Tool {
    return tool.NewTool(
        tool.WithName("git_diff"),
        tool.WithDescription("获取指定 PR 或 commit 的代码差异。返回 unified diff 格式的变更内容。"),
        tool.WithParameters(tool.Parameters{
            Type: "object",
            Properties: map[string]tool.Property{
                "base_branch": {
                    Type:        "string",
                    Description: "基准分支，默认为 main",
                },
                "target_branch": {
                    Type:        "string",
                    Description: "目标分支（PR 的 head branch）",
                },
                "file_filter": {
                    Type:        "string",
                    Description: "文件过滤模式（如 '*.go'），可选",
                },
            },
            Required: []string{"target_branch"},
        }),
        tool.WithHandler(handleGitDiff),
    )
}

func handleGitDiff(ctx context.Context, params json.RawMessage) (string, error) {
    var input struct {
        BaseBranch   string `json:"base_branch"`
        TargetBranch string `json:"target_branch"`
        FileFilter   string `json:"file_filter"`
    }
    
    if err := json.Unmarshal(params, &input); err != nil {
        return "", fmt.Errorf("参数解析失败: %w", err)
    }
    
    // 默认基准分支
    if input.BaseBranch == "" {
        input.BaseBranch = "main"
    }
    
    // 构建 git diff 命令
    args := []string{"diff", input.BaseBranch + "..." + input.TargetBranch}
    if input.FileFilter != "" {
        args = append(args, "--", input.FileFilter)
    }
    
    cmd := exec.CommandContext(ctx, "git", args...)
    output, err := cmd.Output()
    if err != nil {
        return "", fmt.Errorf("git diff 执行失败: %w", err)
    }
    
    diff := string(output)
    
    // 如果差异太大，截断并提示
    if len(diff) > 10000 {
        diff = diff[:10000] + "\n\n... [差异内容过长，已截断。请使用 file_filter 缩小范围]"
    }
    
    return diff, nil
}

// internal/tools/security.go
package tools

import (
    "context"
    "encoding/json"
    "fmt"
    "regexp"
    "strings"
    
    "trpc.group/trpc-go/trpc-agent-go/tool"
)

// SecurityScan 创建安全扫描工具
func SecurityScan() *tool.Tool {
    return tool.NewTool(
        tool.WithName("security_scan"),
        tool.WithDescription("对代码进行安全扫描，检查常见安全问题如硬编码密钥、SQL注入风险等。"),
        tool.WithParameters(tool.Parameters{
            Type: "object",
            Properties: map[string]tool.Property{
                "code": {
                    Type:        "string",
                    Description: "要扫描的代码内容",
                },
                "language": {
                    Type:        "string",
                    Description: "编程语言（go/python/javascript）",
                },
            },
            Required: []string{"code", "language"},
        }),
        tool.WithHandler(handleSecurityScan),
    )
}

// 安全规则（简化示例）
var securityRules = []struct {
    Name    string
    Pattern *regexp.Regexp
    Severity string
    Message string
}{
    {
        Name:     "hardcoded_secret",
        Pattern:  regexp.MustCompile(`(?i)(password|secret|api_key|token)\s*[:=]\s*["']([^"']{8,})["']`),
        Severity: "critical",
        Message:  "发现硬编码的敏感信息，应使用环境变量或密钥管理服务",
    },
    {
        Name:     "sql_injection",
        Pattern:  regexp.MustCompile(`(?i)(fmt\.Sprintf|string\.format).*?(SELECT|INSERT|UPDATE|DELETE)`),
        Severity: "critical",
        Message:  "可能存在 SQL 注入风险，应使用参数化查询",
    },
    {
        Name:     "unsafe_deserialization",
        Pattern:  regexp.MustCompile(`(?i)json\.Unmarshal.*interface\{\}`),
        Severity: "major",
        Message:  "反序列化到 interface{} 可能导致类型安全问题",
    },
}

func handleSecurityScan(ctx context.Context, params json.RawMessage) (string, error) {
    var input struct {
        Code     string `json:"code"`
        Language string `json:"language"`
    }
    
    if err := json.Unmarshal(params, &input); err != nil {
        return "", fmt.Errorf("参数解析失败: %w", err)
    }
    
    var findings []map[string]interface{}
    lines := strings.Split(input.Code, "\n")
    
    for _, rule := range securityRules {
        for i, line := range lines {
            if rule.Pattern.MatchString(line) {
                findings = append(findings, map[string]interface{}{
                    "rule":     rule.Name,
                    "severity": rule.Severity,
                    "line":     i + 1,
                    "message":  rule.Message,
                    "context":  strings.TrimSpace(line),
                })
            }
        }
    }
    
    result, _ := json.MarshalIndent(map[string]interface{}{
        "total_findings": len(findings),
        "findings":       findings,
    }, "", "  ")
    
    return string(result), nil
}

// internal/agents/testgen.go
package agents

import (
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "github.com/your-team/devops-agent/internal/tools"
)

const testgenPrompt = `你是一位测试工程专家。你的职责是为代码变更生成高质量的测试用例。

## 你的能力
- 阅读源代码并理解其逻辑
- 设计覆盖率高的测试用例
- 编写可独立运行的测试代码
- 运行测试并分析结果

## 测试设计原则
1. 每个公开函数至少一个正常路径测试
2. 每个错误分支至少一个测试
3. 边界条件必须覆盖（空值、零值、极大值）
4. 并发场景如果适用必须测试
5. 使用 Table-Driven 风格组织 Go 测试

## 输出格式
生成测试后输出：
{
  "test_file": "文件路径",
  "test_count": 测试数量,
  "coverage_estimate": "预估覆盖率",
  "test_code": "完整测试代码"
}

## 重要约束
- 测试必须能独立运行（不依赖外部服务）
- 需要外部依赖时必须 mock
- 测试命名必须描述行为（TestUserLogin_WithInvalidPassword_ReturnsError）
- 不修改被测代码，只生成测试代码`

func NewTestGenerator() *agent.Agent {
    return agent.NewAgent(
        agent.WithName("TestGenerator"),
        agent.WithSystemPrompt(testgenPrompt),
        agent.WithModel("gpt-4o"),
        agent.WithTools(
            tools.FileRead(),      // 读取源文件
            tools.FileWrite(),     // 写入测试文件
            tools.RunTests(),      // 执行测试
            tools.Coverage(),      // 获取覆盖率
        ),
        agent.WithTemperature(0.3), // 略有创意但不要太跳
    )
}

// internal/tools/testing.go
package tools

import (
    "context"
    "encoding/json"
    "fmt"
    "os/exec"
    "strings"
    "time"
    
    "trpc.group/trpc-go/trpc-agent-go/tool"
)

// RunTests 创建测试运行工具
func RunTests() *tool.Tool {
    return tool.NewTool(
        tool.WithName("run_tests"),
        tool.WithDescription("运行指定包或文件的 Go 测试，返回测试结果和覆盖率。"),
        tool.WithParameters(tool.Parameters{
            Type: "object",
            Properties: map[string]tool.Property{
                "package_path": {
                    Type:        "string",
                    Description: "要测试的包路径（如 ./internal/service/...）",
                },
                "test_name": {
                    Type:        "string",
                    Description: "指定运行的测试名（正则），可选",
                },
                "timeout": {
                    Type:        "string",
                    Description: "超时时间（如 30s），默认 60s",
                },
            },
            Required: []string{"package_path"},
        }),
        tool.WithHandler(handleRunTests),
    )
}

func handleRunTests(ctx context.Context, params json.RawMessage) (string, error) {
    var input struct {
        PackagePath string `json:"package_path"`
        TestName    string `json:"test_name"`
        Timeout     string `json:"timeout"`
    }
    
    if err := json.Unmarshal(params, &input); err != nil {
        return "", fmt.Errorf("参数解析失败: %w", err)
    }
    
    if input.Timeout == "" {
        input.Timeout = "60s"
    }
    
    // 构建测试命令
    args := []string{"test", "-v", "-cover", "-timeout", input.Timeout}
    if input.TestName != "" {
        args = append(args, "-run", input.TestName)
    }
    args = append(args, input.PackagePath)
    
    // 设置命令超时
    timeout, _ := time.ParseDuration(input.Timeout)
    cmdCtx, cancel := context.WithTimeout(ctx, timeout+5*time.Second)
    defer cancel()
    
    cmd := exec.CommandContext(cmdCtx, "go", args...)
    output, err := cmd.CombinedOutput()
    
    // 解析结果
    result := map[string]interface{}{
        "command": "go " + strings.Join(args, " "),
        "output":  string(output),
        "success": err == nil,
    }
    
    // 提取覆盖率
    for _, line := range strings.Split(string(output), "\n") {
        if strings.Contains(line, "coverage:") {
            result["coverage"] = strings.TrimSpace(line)
            break
        }
    }
    
    resultJSON, _ := json.MarshalIndent(result, "", "  ")
    return string(resultJSON), nil
}

// internal/agents/deployer.go
package agents

import (
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "github.com/your-team/devops-agent/internal/tools"
)

const deployerPrompt = `你是一位部署工程专家。你的职责是安全地将代码变更部署到生产环境。

## 你的能力
- 构建 Docker 镜像
- 管理 Kubernetes 部署
- 执行健康检查
- 处理回滚

## 部署策略
1. 先灰度（5%流量），观察 5 分钟
2. 扩大灰度（25%），观察 5 分钟
3. 全量发布
4. 任何阶段异常指标 → 立即回滚

## 回滚条件（触发任一即回滚）
- 错误率 > 1%
- P99 延迟 > 基准的 2 倍
- 健康检查连续 3 次失败
- CPU/内存使用率异常飙升

## 输出格式
{
  "action": "deploy" | "rollback" | "abort",
  "stage": "build" | "canary" | "full",
  "status": "success" | "failed" | "rolled_back",
  "details": "详细说明",
  "metrics": {
    "error_rate": "错误率",
    "latency_p99": "P99延迟",
    "health_checks_passed": true/false
  }
}

## 重要约束
- 安全第一：宁可回滚也不能带病上线
- 每一步都要有健康检查确认
- 部署失败必须自动回滚，不能停在中间状态
- 所有操作都必须有审计日志`

func NewDeployer() *agent.Agent {
    return agent.NewAgent(
        agent.WithName("Deployer"),
        agent.WithSystemPrompt(deployerPrompt),
        agent.WithModel("gpt-4o"),
        agent.WithTools(
            tools.DockerBuild(),   // 构建镜像
            tools.K8sDeploy(),     // Kubernetes 部署
            tools.HealthCheck(),   // 健康检查
            tools.MetricsQuery(),  // 查询监控指标
            tools.Rollback(),      // 回滚
        ),
        agent.WithTemperature(0.0), // 部署必须完全确定性
    )
}

// internal/tools/health.go
package tools

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
    
    "trpc.group/trpc-go/trpc-agent-go/tool"
)

// HealthCheck 创建健康检查工具
func HealthCheck() *tool.Tool {
    return tool.NewTool(
        tool.WithName("health_check"),
        tool.WithDescription("对目标服务执行健康检查，验证服务是否正常运行。"),
        tool.WithParameters(tool.Parameters{
            Type: "object",
            Properties: map[string]tool.Property{
                "endpoint": {
                    Type:        "string",
                    Description: "健康检查的 URL（如 http://service:8080/health）",
                },
                "retries": {
                    Type:        "integer",
                    Description: "重试次数，默认 3",
                },
                "interval_seconds": {
                    Type:        "integer",
                    Description: "重试间隔秒数，默认 5",
                },
            },
            Required: []string{"endpoint"},
        }),
        tool.WithHandler(handleHealthCheck),
    )
}

func handleHealthCheck(ctx context.Context, params json.RawMessage) (string, error) {
    var input struct {
        Endpoint        string `json:"endpoint"`
        Retries         int    `json:"retries"`
        IntervalSeconds int    `json:"interval_seconds"`
    }
    
    if err := json.Unmarshal(params, &input); err != nil {
        return "", fmt.Errorf("参数解析失败: %w", err)
    }
    
    if input.Retries <= 0 {
        input.Retries = 3
    }
    if input.IntervalSeconds <= 0 {
        input.IntervalSeconds = 5
    }
    
    client := &http.Client{Timeout: 10 * time.Second}
    var lastErr error
    
    for i := 0; i < input.Retries; i++ {
        resp, err := client.Get(input.Endpoint)
        if err != nil {
            lastErr = err
            time.Sleep(time.Duration(input.IntervalSeconds) * time.Second)
            continue
        }
        resp.Body.Close()
        
        if resp.StatusCode == http.StatusOK {
            result, _ := json.MarshalIndent(map[string]interface{}{
                "healthy":      true,
                "status_code":  resp.StatusCode,
                "attempts":     i + 1,
                "endpoint":     input.Endpoint,
            }, "", "  ")
            return string(result), nil
        }
        
        lastErr = fmt.Errorf("HTTP %d", resp.StatusCode)
        time.Sleep(time.Duration(input.IntervalSeconds) * time.Second)
    }
    
    result, _ := json.MarshalIndent(map[string]interface{}{
        "healthy":     false,
        "error":       lastErr.Error(),
        "attempts":    input.Retries,
        "endpoint":    input.Endpoint,
    }, "", "  ")
    return string(result), nil
}

// internal/agents/coordinator.go
package agents

import (
    "trpc.group/trpc-go/trpc-agent-go/agent"
    "github.com/your-team/devops-agent/internal/tools"
)

const coordinatorPrompt = `你是 DevOps 流水线的总调度员。你负责分析代码变更，决定流水线的执行策略，并协调各专业 Agent 的工作。

## 你的团队
- CodeReviewer：负责代码审查
- TestGenerator：负责生成和运行测试
- Deployer：负责构建和部署

## 你的决策逻辑

### 变更类型判断
1. 文档变更（只改 .md/.txt）→ 只走 CodeReviewer（轻量审查）
2. 配置变更（改 .yaml/.env）→ CodeReviewer + Deployer（跳过测试）
3. 代码变更（改 .go/.py 等）→ 全流程
4. 紧急修复（hotfix 分支）→ 简化审查 + 快速部署

### 决策流程
1. 获取 PR 信息
2. 判断变更类型
3. 指派合适的 Agent 执行任务
4. 收集执行结果
5. 做出最终决策（部署/修改/拒绝）

## 输出格式
在每个决策点输出：
{
  "decision": "你的决策",
  "reasoning": "决策原因",
  "next_action": "下一步动作",
  "assigned_to": "指派给谁"
}

## 重要约束
- 你不直接执行代码审查、测试、部署——这是专业 Agent 的工作
- 你负责全局视角和决策
- 如果 CodeReviewer 给出 reject，流程立即终止
- 测试覆盖率 < 80% 不允许部署`

func NewPipelineManager() *agent.Agent {
    return agent.NewAgent(
        agent.WithName("PipelineManager"),
        agent.WithSystemPrompt(coordinatorPrompt),
        agent.WithModel("gpt-4o"),
        agent.WithTools(
            tools.GetPRInfo(),     // 获取 PR 元信息
            tools.Notify(),        // 发送通知
        ),
        agent.WithTemperature(0.1),
    )
}

// internal/workflow/pipeline.go
package workflow

import (
    "context"
    "fmt"
    "log"
    
    "trpc.group/trpc-go/trpc-agent-go/team"
    "trpc.group/trpc-go/trpc-agent-go/runner"
    
    "github.com/your-team/devops-agent/internal/agents"
)

// DevOpsPipeline 封装完整的 DevOps 流水线
type DevOpsPipeline struct {
    team   *team.Team
    runner *runner.Runner
}

// NewDevOpsPipeline 创建 DevOps 流水线
func NewDevOpsPipeline() *DevOpsPipeline {
    // 1. 创建各 Agent
    coordinator := agents.NewPipelineManager()
    reviewer := agents.NewCodeReviewer()
    testgen := agents.NewTestGenerator()
    deployer := agents.NewDeployer()
    
    // 2. 组装 Team
    t := team.NewTeam(
        team.WithName("DevOps Pipeline"),
        team.WithMode(team.ModeCoordinator),
        team.WithCoordinator(coordinator),
        team.WithMembers(reviewer, testgen, deployer),
        team.WithMaxRounds(15),          // 最多 15 轮
        team.WithTimeout(300),            // 5 分钟超时
        team.WithParallelExecution(true), // 允许并行
    )
    
    // 3. 创建 Runner
    r := runner.NewRunner(
        runner.WithAgent(t),  // Team 也实现了 Agent 接口
    )
    
    return &DevOpsPipeline{
        team:   t,
        runner: r,
    }
}

// Execute 执行流水线
func (p *DevOpsPipeline) Execute(ctx context.Context, prInfo string) (*PipelineResult, error) {
    prompt := fmt.Sprintf("请处理以下 Pull Request：\n%s", prInfo)
    
    result, err := p.runner.Run(ctx, prompt)
    if err != nil {
        return nil, fmt.Errorf("流水线执行失败: %w", err)
    }
    
    // 解析结果
    return parsePipelineResult(result), nil
}

// tests/integration_test.go
package tests

import (
    "context"
    "testing"
    "time"
    
    "github.com/your-team/devops-agent/internal/workflow"
)

func TestPipeline_SimpleCodeChange(t *testing.T) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
    
    pipeline := workflow.NewDevOpsPipeline()
    
    // 模拟一个简单的代码变更 PR
    prInfo := `
    PR #42: 添加用户登录日志
    分支: feature/login-logging -> main
    变更文件:
    - internal/auth/login.go (+15, -2)
    - internal/auth/login_test.go (+30, -0)
    
    变更内容摘要：在登录成功/失败时记录日志，包含用户名和时间戳。
    `
    
    result, err := pipeline.Execute(ctx, prInfo)
    if err != nil {
        t.Fatalf("流水线执行失败: %v", err)
    }
    
    // 验证结果
    if result == nil {
        t.Fatal("结果不应为 nil")
    }
    
    // 验证所有 Agent 都被调用了
    if !result.ReviewCompleted {
        t.Error("代码审查未完成")
    }
    if !result.TestsGenerated {
        t.Error("测试未生成")
    }
    
    t.Logf("流水线结果: %+v", result)
}

func TestPipeline_DocOnlyChange(t *testing.T) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
    defer cancel()
    
    pipeline := workflow.NewDevOpsPipeline()
    
    // 只改文档——应该走轻量路径
    prInfo := `
    PR #43: 更新 README
    分支: docs/update-readme -> main
    变更文件:
    - README.md (+5, -3)
    - docs/api.md (+20, -0)
    
    变更内容摘要：更新安装说明和 API 文档。
    `
    
    result, err := pipeline.Execute(ctx, prInfo)
    if err != nil {
        t.Fatalf("流水线执行失败: %v", err)
    }
    
    // 文档变更不应该触发部署
    if result.DeployTriggered {
        t.Error("文档变更不应触发部署")
    }
}

// 问题代码：所有 Agent 共享一个 session
t := team.NewTeam(
    team.WithSharedMemory(true),  // 这会导致对话混乱
)

// 修复：每个 Agent 独立 session，只通过 Coordinator 传递结构化结果
t := team.NewTeam(
    team.WithSharedMemory(false), // 各自独立
    team.WithMessageFilter(func(msg Message) bool {
        // 只传递最终结论，不传内部推理过程
        return msg.Role == "result"
    }),
)

// 解决：添加循环计数器和退出条件
type ReviewState struct {
    Attempts int
    MaxAttempts int  // = 3
}

func reviewDecision(state ReviewState) string {
    if state.Attempts >= state.MaxAttempts {
        return "escalate_to_human"  // 超过最大次数，转人工
    }
    if state.ReviewPassed {
        return "proceed_to_deploy"
    }
    return "retry_review"  // 继续循环
}

// 解决：为每个 Agent 设置独立超时
reviewer := agent.NewAgent(
    agent.WithName("CodeReviewer"),
    agent.WithTimeout(60 * time.Second),  // 单次 LLM 调用超时
    agent.WithMaxRetries(2),              // 重试 2 次
    agent.WithRetryDelay(5 * time.Second), // 重试间隔
)

// internal/workflow/review_loop.go
package workflow

import (
    "context"
    "fmt"
    
    "trpc.group/trpc-go/trpc-agent-go/stategraph"
)

// ReviewState 审查循环的状态
type ReviewState struct {
    PRDiff         string   `sg:"type:string"`
    ReviewResult   string   `sg:"type:string"`
    Suggestions    []string `sg:"type:list,reducer:append"`
    Attempts       int      `sg:"type:int"`
    MaxAttempts    int      `sg:"type:int"`
    FinalVerdict   string   `sg:"type:string"`
}

// BuildReviewGraph 构建代码审查循环图
func BuildReviewGraph() *stategraph.CompiledGraph {
    // 1. 创建图
    graph := stategraph.NewStateGraph[ReviewState]()
    
    // 2. 添加节点
    graph.AddNode("review", reviewNode)          // 执行审查
    graph.AddNode("generate_fix", generateFixNode) // 生成修复建议
    graph.AddNode("apply_fix", applyFixNode)     // 应用修复
    graph.AddNode("final_report", finalReportNode) // 生成最终报告
    
    // 3. 设置入口
    graph.SetEntryPoint("review")
    
    // 4. 添加条件路由（关键！）
    graph.AddConditionalEdge("review", reviewRouter)
    
    // 5. 添加普通边
    graph.AddEdge("generate_fix", "apply_fix")
    graph.AddEdge("apply_fix", "review")  // 回到审查（循环！）
    graph.AddEdge("final_report", stategraph.END)
    
    // 6. 编译
    compiled, err := graph.Compile()
    if err != nil {
        panic(fmt.Sprintf("图编译失败: %v", err))
    }
    
    return compiled
}

// reviewRouter 条件路由：决定审查后走哪个分支
func reviewRouter(state ReviewState) string {
    // 超过最大尝试次数 → 强制出报告
    if state.Attempts >= state.MaxAttempts {
        return "final_report"
    }
    
    // 审查通过 → 出报告
    if state.FinalVerdict == "approve" {
        return "final_report"
    }
    
    // 审查不通过但还有机会 → 生成修复
    return "generate_fix"
}

// reviewNode 执行代码审查
func reviewNode(ctx context.Context, state ReviewState) (ReviewState, error) {
    state.Attempts++
    
    // 这里调用 CodeReviewer Agent 做实际审查
    reviewer := agents.NewCodeReviewer()
    result, err := reviewer.Run(ctx, fmt.Sprintf(
        "请审查以下代码变更（第 %d 次审查）：\n%s",
        state.Attempts, state.PRDiff,
    ))
    if err != nil {
        return state, err
    }
    
    state.ReviewResult = result.Content
    // 解析 verdict
    state.FinalVerdict = parseVerdict(result.Content)
    
    return state, nil
}

// generateFixNode 生成修复建议
func generateFixNode(ctx context.Context, state ReviewState) (ReviewState, error) {
    // 基于审查意见生成修复建议
    suggestion := fmt.Sprintf("基于第 %d 次审查的修复建议: ...", state.Attempts)
    state.Suggestions = append(state.Suggestions, suggestion)
    return state, nil
}

// applyFixNode 应用修复（模拟）
func applyFixNode(ctx context.Context, state ReviewState) (ReviewState, error) {
    // 在真实场景中，这里会修改代码并更新 PRDiff
    // Demo 中我们直接更新 diff 的描述
    state.PRDiff = state.PRDiff + "\n[已应用修复建议]"
    return state, nil
}

// finalReportNode 生成最终报告
func finalReportNode(ctx context.Context, state ReviewState) (ReviewState, error) {
    // 汇总所有审查轮次的结果
    return state, nil
}

// internal/workflow/full_pipeline.go
package workflow

import (
    "context"
    "fmt"
    
    "trpc.group/trpc-go/trpc-agent-go/stategraph"
)

// PipelineState 完整流水线的状态
type PipelineState struct {
    // 输入
    PRInfo          string `sg:"type:string"`
    ChangeType      string `sg:"type:string"`  // code/docs/config/hotfix
    
    // 中间状态
    ReviewReport    string `sg:"type:string"`
    TestReport      string `sg:"type:string"`
    BuildStatus     string `sg:"type:string"`
    DeployStatus    string `sg:"type:string"`
    
    // 控制流
    ReviewPassed    bool   `sg:"type:bool"`
    TestsPassed     bool   `sg:"type:bool"`
    ShouldDeploy    bool   `sg:"type:bool"`
    
    // 输出
    FinalReport     string `sg:"type:string"`
}

// BuildFullPipeline 构建完整流水线图
func BuildFullPipeline() *stategraph.CompiledGraph {
    graph := stategraph.NewStateGraph[PipelineState]()
    
    // 节点
    graph.AddNode("classify", classifyChangeNode)   // 分类变更类型
    graph.AddNode("review", reviewPipelineNode)     // 代码审查
    graph.AddNode("test", testPipelineNode)         // 生成+运行测试
    graph.AddNode("build", buildNode)               // 构建镜像
    graph.AddNode("deploy_canary", deployCanaryNode) // 灰度部署
    graph.AddNode("monitor", monitorNode)           // 监控验证
    graph.AddNode("deploy_full", deployFullNode)    // 全量部署
    graph.AddNode("rollback", rollbackNode)         // 回滚
    graph.AddNode("report", generateReportNode)     // 生成报告
    
    // 入口
    graph.SetEntryPoint("classify")
    
    // 条件路由
    graph.AddConditionalEdge("classify", func(s PipelineState) string {
        switch s.ChangeType {
        case "docs":
            return "review"  // 文档只需审查
        case "hotfix":
            return "build"   // 紧急修复跳过审查
        default:
            return "review"  // 正常流程
        }
    })
    
    graph.AddConditionalEdge("review", func(s PipelineState) string {
        if !s.ReviewPassed {
            return "report"  // 审查不通过，直接出报告
        }
        if s.ChangeType == "docs" {
            return "report"  // 文档变更审查通过就结束
        }
        return "test"
    })
    
    graph.AddConditionalEdge("test", func(s PipelineState) string {
        if !s.TestsPassed {
            return "report"  // 测试不通过
        }
        return "build"
    })
    
    graph.AddEdge("build", "deploy_canary")
    
    graph.AddConditionalEdge("monitor", func(s PipelineState) string {
        if s.DeployStatus == "healthy" {
            return "deploy_full"
        }
        return "rollback"
    })
    
    graph.AddEdge("deploy_canary", "monitor")
    graph.AddEdge("deploy_full", "report")
    graph.AddEdge("rollback", "report")
    graph.AddEdge("report", stategraph.END)
    
    compiled, _ := graph.Compile()
    return compiled
}

// classifyChangeNode 分析 PR 判断变更类型
func classifyChangeNode(ctx context.Context, state PipelineState) (PipelineState, error) {
    // 简化的分类逻辑
    // 实际项目中会调用 PipelineManager Agent 做智能分类
    if containsOnlyDocs(state.PRInfo) {
        state.ChangeType = "docs"
    } else if isHotfix(state.PRInfo) {
        state.ChangeType = "hotfix"
    } else if isConfigChange(state.PRInfo) {
        state.ChangeType = "config"
    } else {
        state.ChangeType = "code"
    }
    return state, nil
}

// 其余节点函数的实现类似，各自调用对应的 Agent
// reviewPipelineNode 调用 CodeReviewer
// testPipelineNode 调用 TestGenerator
// buildNode 调用 DockerBuild 工具
// deployCanaryNode/deployFullNode 调用 Deployer
// monitorNode 调用 HealthCheck + MetricsQuery
// rollbackNode 调用 Rollback 工具
// generateReportNode 汇总所有结果

// cmd/demo/main.go
package main

import (
    "context"
    "fmt"
    "log"
    
    "github.com/your-team/devops-agent/internal/workflow"
)

func main() {
    // 构建图
    pipeline := workflow.BuildFullPipeline()
    
    // 初始状态
    initialState := workflow.PipelineState{
        PRInfo: `PR #100: 重构用户认证模块
分支: refactor/auth -> main
变更文件: internal/auth/*.go (8 files, +200, -150)`,
    }
    
    // 执行
    ctx := context.Background()
    finalState, err := pipeline.Invoke(ctx, initialState)
    if err != nil {
        log.Fatalf("流水线执行失败: %v", err)
    }
    
    fmt.Println("最终报告:")
    fmt.Println(finalState.FinalReport)
}

// internal/plugin/logger.go
package plugin

import (
    "context"
    "encoding/json"
    "log"
    "time"
    
    agentplugin "trpc.group/trpc-go/trpc-agent-go/plugin"
)

// LoggerPlugin 日志插件——记录每一步的输入输出
type LoggerPlugin struct{}

func NewLoggerPlugin() *LoggerPlugin {
    return &LoggerPlugin{}
}

// BeforeLLMCall 在 LLM 调用前执行
func (p *LoggerPlugin) BeforeLLMCall(ctx context.Context, req *agentplugin.LLMRequest) error {
    log.Printf("[LLM请求] Agent=%s Model=%s Messages=%d Tools=%d",
        req.AgentName,
        req.Model,
        len(req.Messages),
        len(req.Tools),
    )
    return nil
}

// AfterLLMCall 在 LLM 调用后执行
func (p *LoggerPlugin) AfterLLMCall(ctx context.Context, resp *agentplugin.LLMResponse) error {
    log.Printf("[LLM响应] Agent=%s Duration=%v TokensUsed=%d ToolCalls=%d",
        resp.AgentName,
        resp.Duration,
        resp.TokensUsed,
        len(resp.ToolCalls),
    )
    return nil
}

// BeforeToolCall 在工具调用前执行
func (p *LoggerPlugin) BeforeToolCall(ctx context.Context, call *agentplugin.ToolCall) error {
    log.Printf("[工具调用] Agent=%s Tool=%s Params=%s",
        call.AgentName,
        call.ToolName,
        string(call.Parameters),
    )
    return nil
}

// AfterToolCall 在工具调用后执行
func (p *LoggerPlugin) AfterToolCall(ctx context.Context, result *agentplugin.ToolResult) error {
    log.Printf("[工具结果] Agent=%s Tool=%s Duration=%v Success=%v",
        result.AgentName,
        result.ToolName,
        result.Duration,
        result.Error == nil,
    )
    return nil
}

// internal/plugin/ratelimit.go
package plugin

import (
    "context"
    "fmt"
    "sync"
    "time"
    
    agentplugin "trpc.group/trpc-go/trpc-agent-go/plugin"
)

// RateLimitPlugin 限流插件
type RateLimitPlugin struct {
    maxRPM     int           // 每分钟最大请求数
    window     time.Duration // 滑动窗口
    requests   []time.Time   // 请求时间记录
    mu         sync.Mutex
}

func NewRateLimitPlugin(maxRPM int) *RateLimitPlugin {
    return &RateLimitPlugin{
        maxRPM:   maxRPM,
        window:   time.Minute,
        requests: make([]time.Time, 0),
    }
}

func (p *RateLimitPlugin) BeforeLLMCall(ctx context.Context, req *agentplugin.LLMRequest) error {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    now := time.Now()
    
    // 清理过期记录
    cutoff := now.Add(-p.window)
    valid := p.requests[:0]
    for _, t := range p.requests {
        if t.After(cutoff) {
            valid = append(valid, t)
        }
    }
    p.requests = valid
    
    // 检查是否超限
    if len(p.requests) >= p.maxRPM {
        waitTime := p.requests[0].Add(p.window).Sub(now)
        return fmt.Errorf("限流：每分钟最多 %d 次请求，请等待 %v", p.maxRPM, waitTime)
    }
    
    // 记录本次请求
    p.requests = append(p.requests, now)
    return nil
}

// internal/plugin/tracer.go
package plugin

import (
    "context"
    "fmt"
    "time"
    
    agentplugin "trpc.group/trpc-go/trpc-agent-go/plugin"
)

// TracerPlugin 追踪插件——记录完整的执行链路
type TracerPlugin struct {
    spans []Span
}

type Span struct {
    TraceID   string
    SpanID    string
    ParentID  string
    Operation string
    Agent     string
    StartTime time.Time
    Duration  time.Duration
    Status    string
    Tags      map[string]string
}

func NewTracerPlugin() *TracerPlugin {
    return &TracerPlugin{
        spans: make([]Span, 0),
    }
}

func (p *TracerPlugin) BeforeLLMCall(ctx context.Context, req *agentplugin.LLMRequest) error {
    span := Span{
        TraceID:   getTraceID(ctx),
        SpanID:    generateSpanID(),
        Operation: "llm_call",
        Agent:     req.AgentName,
        StartTime: time.Now(),
        Tags: map[string]string{
            "model":    req.Model,
            "messages": fmt.Sprintf("%d", len(req.Messages)),
        },
    }
    p.spans = append(p.spans, span)
    return nil
}

// GetTimeline 获取完整执行时间线（用于调试和展示）
func (p *TracerPlugin) GetTimeline() []Span {
    return p.spans
}

// 在创建 Runner 时注册所有 Plugin
r := runner.NewRunner(
    runner.WithAgent(pipeline),
    runner.WithPlugins(
        plugin.NewLoggerPlugin(),
        plugin.NewRateLimitPlugin(60),  // 每分钟最多 60 次 LLM 调用
        plugin.NewTracerPlugin(),
    ),
)

// internal/workflow/ralph.go
package workflow

import (
    "context"
    "encoding/json"
    "fmt"
)

// RalphValidator 验证器接口
type RalphValidator interface {
    Validate(ctx context.Context, output string) (*ValidationResult, error)
}

// ValidationResult 验证结果
type ValidationResult struct {
    Valid       bool     `json:"valid"`
    Confidence float64  `json:"confidence"`  // 0.0 - 1.0
    Issues     []string `json:"issues"`
    Suggestion string   `json:"suggestion"`
}

// ReviewValidator 验证代码审查结果
type ReviewValidator struct{}

func (v *ReviewValidator) Validate(ctx context.Context, output string) (*ValidationResult, error) {
    // 1. 检查输出格式是否正确（必须是有效 JSON）
    var review map[string]interface{}
    if err := json.Unmarshal([]byte(output), &review); err != nil {
        return &ValidationResult{
            Valid:      false,
            Confidence: 1.0,
            Issues:     []string{"输出格式不是有效 JSON"},
            Suggestion: "请以 JSON 格式输出审查结果",
        }, nil
    }
    
    // 2. 检查必要字段
    requiredFields := []string{"verdict", "summary", "issues", "score"}
    var missing []string
    for _, field := range requiredFields {
        if _, ok := review[field]; !ok {
            missing = append(missing, field)
        }
    }
    if len(missing) > 0 {
        return &ValidationResult{
            Valid:      false,
            Confidence: 1.0,
            Issues:     []string{fmt.Sprintf("缺少必要字段: %v", missing)},
            Suggestion: "请确保输出包含 verdict/summary/issues/score 字段",
        }, nil
    }
    
    // 3. 检查 verdict 值是否合法
    verdict, _ := review["verdict"].(string)
    validVerdicts := map[string]bool{"approve": true, "request_changes": true, "reject": true}
    if !validVerdicts[verdict] {
        return &ValidationResult{
            Valid:      false,
            Confidence: 1.0,
            Issues:     []string{fmt.Sprintf("非法 verdict 值: %s", verdict)},
            Suggestion: "verdict 必须是 approve/request_changes/reject 之一",
        }, nil
    }
    
    // 4. 逻辑一致性检查
    score, _ := review["score"].(float64)
    if verdict == "approve" && score < 60 {
        return &ValidationResult{
            Valid:      false,
            Confidence: 0.8,
            Issues:     []string{"逻辑矛盾：verdict=approve 但 score<60"},
            Suggestion: "分数较低时应该 request_changes",
        }, nil
    }
    
    return &ValidationResult{
        Valid:      true,
        Confidence: 0.95,
    }, nil
}

// WithRalphLoop 包装 Agent 执行，加入 Ralph Loop 验证
func WithRalphLoop(agent AgentRunner, validator RalphValidator, maxRetries int) AgentRunner {
    return &ralphWrappedAgent{
        inner:      agent,
        validator:  validator,
        maxRetries: maxRetries,
    }
}

type ralphWrappedAgent struct {
    inner      AgentRunner
    validator  RalphValidator
    maxRetries int
}

func (a *ralphWrappedAgent) Run(ctx context.Context, input string) (*Result, error) {
    for i := 0; i < a.maxRetries; i++ {
        result, err := a.inner.Run(ctx, input)
        if err != nil {
            return nil, err
        }
        
        // 验证输出
        validation, err := a.validator.Validate(ctx, result.Content)
        if err != nil {
            return nil, fmt.Errorf("验证失败: %w", err)
        }
        
        if validation.Valid {
            return result, nil  // 验证通过，返回结果
        }
        
        // 验证不通过，把验证反馈加入下一轮输入
        input = fmt.Sprintf("%s\n\n[验证反馈] 你的上一次输出有问题：%v\n建议：%s\n请修正后重新输出。",
            input,
            validation.Issues,
            validation.Suggestion,
        )
    }
    
    return nil, fmt.Errorf("经过 %d 次重试仍未通过验证", a.maxRetries)
}

// internal/config/errors.go
package config

import (
    "errors"
    "fmt"
    "time"
)

// 自定义错误类型
var (
    ErrLLMTimeout      = errors.New("LLM 调用超时")
    ErrLLMRateLimit    = errors.New("LLM 请求被限流")
    ErrToolFailed      = errors.New("工具执行失败")
    ErrValidationFailed = errors.New("输出验证失败")
    ErrMaxRetries      = errors.New("超过最大重试次数")
)

// RetryConfig 重试配置
type RetryConfig struct {
    MaxRetries  int
    InitialDelay time.Duration
    MaxDelay     time.Duration
    Multiplier   float64  // 退避倍数
}

// DefaultRetryConfig 默认重试配置
var DefaultRetryConfig = RetryConfig{
    MaxRetries:   3,
    InitialDelay: 1 * time.Second,
    MaxDelay:     30 * time.Second,
    Multiplier:   2.0,
}

// RetryWithBackoff 指数退避重试
func RetryWithBackoff(config RetryConfig, fn func() error) error {
    delay := config.InitialDelay
    
    for i := 0; i < config.MaxRetries; i++ {
        err := fn()
        if err == nil {
            return nil
        }
        
        // 判断是否可重试
        if !isRetryable(err) {
            return err
        }
        
        // 等待后重试
        time.Sleep(delay)
        delay = time.Duration(float64(delay) * config.Multiplier)
        if delay > config.MaxDelay {
            delay = config.MaxDelay
        }
    }
    
    return fmt.Errorf("%w: 经过 %d 次重试后仍然失败", ErrMaxRetries, config.MaxRetries)
}

func isRetryable(err error) bool {
    return errors.Is(err, ErrLLMTimeout) || errors.Is(err, ErrLLMRateLimit)
}

// internal/storage/session.go
package storage

import (
    "context"
    "encoding/json"
    "fmt"
    "os"
    "path/filepath"
    "sync"
    "time"
)

// SessionStore 会话持久化存储
type SessionStore struct {
    baseDir string
    mu      sync.RWMutex
}

// Session 会话数据
type Session struct {
    ID        string    `json:"id"`
    CreatedAt time.Time `json:"created_at"`
    UpdatedAt time.Time `json:"updated_at"`
    Messages  []Message `json:"messages"`
    State     map[string]interface{} `json:"state"`
}

type Message struct {
    Role      string    `json:"role"`
    Content   string    `json:"content"`
    Timestamp time.Time `json:"timestamp"`
    Agent     string    `json:"agent,omitempty"`
}

func NewSessionStore(baseDir string) *SessionStore {
    os.MkdirAll(baseDir, 0755)
    return &SessionStore{baseDir: baseDir}
}

// Save 保存会话
func (s *SessionStore) Save(ctx context.Context, session *Session) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    
    session.UpdatedAt = time.Now()
    
    data, err := json.MarshalIndent(session, "", "  ")
    if err != nil {
        return fmt.Errorf("序列化失败: %w", err)
    }
    
    path := filepath.Join(s.baseDir, session.ID+".json")
    return os.WriteFile(path, data, 0644)
}

// Load 加载会话
func (s *SessionStore) Load(ctx context.Context, sessionID string) (*Session, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    
    path := filepath.Join(s.baseDir, sessionID+".json")
    data, err := os.ReadFile(path)
    if err != nil {
        if os.IsNotExist(err) {
            return nil, fmt.Errorf("会话 %s 不存在", sessionID)
        }
        return nil, err
    }
    
    var session Session
    if err := json.Unmarshal(data, &session); err != nil {
        return nil, fmt.Errorf("反序列化失败: %w", err)
    }
    
    return &session, nil
}

// cmd/demo/main.go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"
    
    "github.com/your-team/devops-agent/internal/workflow"
    "github.com/your-team/devops-agent/internal/plugin"
)

func main() {
    // 检查环境变量
    if os.Getenv("OPENAI_API_KEY") == "" {
        log.Fatal("请设置 OPENAI_API_KEY 环境变量")
    }
    
    fmt.Println("========================================")
    fmt.Println("  DevOps 智能流水线 Demo")
    fmt.Println("========================================")
    fmt.Println()
    
    // 创建流水线
    pipeline := workflow.NewDevOpsPipeline()
    
    // 模拟 PR 输入
    prInfo := `
PR #100: 重构用户认证模块
分支: refactor/auth -> main
作者: developer@example.com
变更文件:
  - internal/auth/handler.go (+45, -20)
  - internal/auth/middleware.go (+30, -10)
  - internal/auth/token.go (+60, -0) [新文件]
  - internal/auth/handler_test.go (+80, -0) [新文件]

变更摘要：
1. 将 JWT token 验证逻辑从 handler 抽取到独立的 token.go
2. 添加 token 刷新机制
3. 中间件支持配置化白名单路径
4. 新增单元测试覆盖 token 生成和验证逻辑
    `
    
    fmt.Println("[输入] PR 信息:")
    fmt.Println(prInfo)
    fmt.Println()
    
    // 执行流水线
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
    
    fmt.Println("[执行] 流水线开始运行...")
    fmt.Println()
    
    startTime := time.Now()
    result, err := pipeline.Execute(ctx, prInfo)
    duration := time.Since(startTime)
    
    if err != nil {
        log.Fatalf("流水线执行失败: %v", err)
    }
    
    // 输出结果
    fmt.Println("========================================")
    fmt.Println("  执行结果")
    fmt.Println("========================================")
    fmt.Printf("耗时: %v\n", duration)
    fmt.Println()
    
    resultJSON, _ := json.MarshalIndent(result, "", "  ")
    fmt.Println(string(resultJSON))
}