Rescale 如何基于 Strands Agents 与 AI-DLC 构建智能测试分诊代理

传统 SDLC 流程:   [需求定义] ──> [人工编码] ──> [持续构建] ──> [静态规则测试] ──> [人工排错分诊] ──> [部署]
AI-DLC 闭环系统:   [需求语义] ──> [AI 协同生成] ──> [容器化构建] ──> [并行测试执行]
                                       ▲                                    │
                                       └─── [AI 智能分诊代理] ◄──────────────┘
                                            ├─ 日志语义清洗与故障分类
                                            ├─ 结合 Git Diff 的根因定位
                                            └─ 自动提单 / 代码补丁建议

 ┌────────────────────────────────────────────────────────────────────────┐
 │                       持续集成流水线 (CI/CD Pipeline)                   │
 │                 (GitHub Actions / Jenkins / GitLab CI)                 │
 └───────────────────────────────────┬────────────────────────────────────┘
                                     │ Webhook 故障事件触发
                                     ▼
 ┌────────────────────────────────────────────────────────────────────────┐
 │                     Strands 数据预处理管道                             │
 │  ├─ 敏感凭证过滤引擎 (脱敏 Token、IP 地址、内网域名)                    │
 │  ├─ 噪音削减算法 (过滤周期性心跳日志与冗余控制台输出)                  │
 │  └─ 语义切片单元 (精准截取崩溃上下文关键窗口)                          │
 └───────────────────────────────────┬────────────────────────────────────┘
                                     │
                                     ▼
 ┌────────────────────────────────────────────────────────────────────────┐
 │                     Strands 智能分诊代理核心编排层                     │
 │                                                                        │
 │  ┌───────────────────────┐          ┌───────────────────────────────┐  │
 │  │    快速分类代理       │          │          根因诊断代理         │  │
 │  │ (基础设施/代码缺陷)   │ ───────> │    (深度比对 Diff 与调用栈)   │  │
 │  └───────────────────────┘          └───────────────┬───────────────┘  │
 │             ▲                                       │                  │
 │             │ 检索历史模式                          │ 执行工具调用     │
 │  ┌──────────┴────────────┐          ┌───────────────▼───────────────┐  │
 │  │    向量知识数据库     │          │         行动执行代理          │  │
 │  │ (OpenSearch / pgvector)│         │ (调用 Jira / Slack / Git API) │  │
 │  └───────────────────────┘          └───────────────────────────────┘  │
 └───────────────────────────────────┬────────────────────────────────────┘
                                     │
                                     ▼
 ┌────────────────────────────────────────────────────────────────────────┐
 │                    自动化工程反馈与自愈动作                            │
 │    - 自动隔离已知 Flaky 测试用例                                       │
 │    - 在 Pull Request 评论区推送结构化根因分析报告                      │
 │    - 针对瞬时网络中断自动重启特定测试任务                              │
 └────────────────────────────────────────────────────────────────────────┘

import os
import json
from typing import List
from pydantic import BaseModel, Field
import requests

# 定义结构化分析结果模型
class TriageVerdict(BaseModel):
    failure_category: str = Field(description="故障类别: INFRASTRUCTURE, FLAKY, CODE_DEFECT, CONFIGURATION")
    root_cause_summary: str = Field(description="故障技术根因的清晰简明阐述")
    confidence_score: float = Field(description="模型推断置信度，范围 0.0 到 1.0")
    affected_files: List[str] = Field(description="调用链中直接相关的源码文件列表")
    suggested_action: str = Field(description="系统建议动作: RERUN, QUARANTINE, FILE_BUG, FIX_DIFF")

class TestTriageAgent:
    def __init__(self, api_key: str, base_url: str = "https://api.n1n.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

    def clean_logs(self, raw_logs: str, max_lines: int = 150) -> str:
        """截取异常核心窗口，降低 Token 消耗并聚焦关键逻辑"""
        lines = raw_logs.splitlines()
        trigger_tokens = ["ERROR", "FAIL", "Traceback", "Exception", "TimeoutError"]
        matched_indices = [i for i, line in enumerate(lines) if any(tok in line for tok in trigger_tokens)]
        
        if not matched_indices:
            return "\n".join(lines[-max_lines:])
        
        # 截取首个异常点前 10 行至后 140 行的内容
        start = max(0, matched_indices[0] - 10)
        return "\n".join(lines[start:start + max_lines])

    def analyze_failure(self, test_name: str, clean_trace: str, git_diff: str) -> TriageVerdict:
        system_prompt = (
            "你是一个高级 CI/CD 自动化故障诊断专家。请仔细比对提供的代码变动与运行异常日志。\n"
            "你的职责是精准识别是代码逻辑缺陷还是瞬态基础设施报错。\n"
            "必须严格输出符合规范的 JSON 字典。"
        )

        user_content = f"""测试用例名称: {test_name}

[最近代码提交变动 Git Diff]:
{git_diff[:2000]}

[清洗后的堆栈错误日志]:
{clean_trace}
"""

        body = {
            "model": "claude-3-5-sonnet-20241022",
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_content}
            ],
            "response_format": {"type": "json_object"},
            "temperature": 0.1
        }

        response = requests.post(f"{self.base_url}/chat/completions", headers=self.headers, json=body)
        response.raise_for_status()
        
        raw_output = response.json()["choices"][0]["message"]["content"]
        parsed_json = json.loads(raw_output)
        return TriageVerdict(**parsed_json)

# 运行演示
if __name__ == "__main__":
    # 推荐使用 n1n.ai 提供的极速聚合 API 接入点
    client_key = os.getenv("N1N_API_KEY", "sk-sample-key")
    agent = TestTriageAgent(api_key=client_key, base_url="https://api.n1n.ai/v1")

    sample_log = """
    [INFO] 执行初始化: test_user_authentication_flow
    [DEBUG] 连接 Redis 集群 (10.0.4.15:6379)... 成功
    [ERROR] 致命错误: redis.exceptions.ConnectionError: Connection closed by server.
    Traceback (most recent call last):
      File "/app/tests/test_auth.py", line 78, in test_user_authentication_flow
        session = auth_service.generate_token(user_id="usr_9918")
      File "/app/services/auth.py", line 34, in generate_token
        self.cache_client.setex(f"session:{user_id}", 10, token)
    redis.exceptions.ConnectionError: Connection closed by server.
    """

    sample_diff = """
    diff --git a/app/services/auth.py b/app/services/auth.py
    - self.cache_client.setex(f"session:{user_id}", 3600, token)
    + self.cache_client.setex(f"session:{user_id}", 10, token)
    """

    processed_log = agent.clean_logs(sample_log)
    verdict = agent.analyze_failure("test_user_authentication_flow", processed_log, sample_diff)
    print("\n=== 分诊评估结果 ===")
    print(verdict.json(indent=2, ensure_ascii=False))