🦞 基于 OpenClaw 的全流程 Bug 解决助理 Agent

# 多智能体通信协议（基于 Redis Pub/Sub） import redis import json from dataclasses import dataclass from enum import Enum from typing import Dict, Any, Optional class AgentRole(Enum): COLLECTOR = "collector" LOCATOR = "locator" FIXER = "fixer" TESTER = "tester" DEPLOYER = "deployer" AUDITOR = "auditor" class MessageType(Enum): BUG_REPORT = "bug_report" LOCATION_RESULT = "location_result" FIX_PROPOSAL = "fix_proposal" TEST_RESULT = "test_result" DEPLOY_STATUS = "deploy_status" AUDIT_ALERT = "audit_alert" @dataclass class AgentMessage: """智能体间通信消息""" msg_id: str msg_type: MessageType sender: AgentRole receiver: AgentRole payload: Dict[str, Any] timestamp: float bug_id: str priority: int = 3 # 1-5, 1 最高 def to_json(self) -> str: return json.dumps({ "msg_id": self.msg_id, "msg_type": self.msg_type.value, "sender": self.sender.value, "receiver": self.receiver.value, "payload": self.payload, "timestamp": self.timestamp, "bug_id": self.bug_id, "priority": self.priority }) @classmethod def from_json(cls, json_str: str) -> "AgentMessage": data = json.loads(json_str) return cls( msg_id=data["msg_id"], msg_type=MessageType(data["msg_type"]), sender=AgentRole(data["sender"]), receiver=AgentRole(data["receiver"]), payload=data["payload"], timestamp=data["timestamp"], bug_id=data["bug_id"], priority=data.get("priority", 3) ) class AgentCommunicationHub: """智能体通信中心""" def __init__(self, redis_url: str = "redis://localhost:6379"): self.redis = redis.from_url(redis_url) self.pubsub = self.redis.pubsub() self.channel_prefix = "openclaw:bugfix:" def publish(self, message: AgentMessage): """发布消息到指定频道""" channel = f"{self.channel_prefix}{message.receiver.value}" self.redis.publish(channel, message.to_json()) def subscribe(self, role: AgentRole): """订阅指定角色的消息""" channel = f"{self.channel_prefix}{role.value}" self.pubsub.subscribe(channel) return self.pubsub def broadcast(self, message: AgentMessage): """广播消息到所有智能体""" channel = f"{self.channel_prefix}all" self.redis.publish(channel, message.to_json())

# 基于状态机的任务编排 from transitions import Machine from datetime import datetime class BugFixWorkflow: """Bug 修复工作流状态机""" states = [ 'collected', # 已收集 'triaged', # 已分类 'locating', # 定位中 'located', # 已定位 'fixing', # 修复中 'fixed', # 已修复 'testing', # 测试中 'tested', # 已测试 'deploying', # 部署中 'deployed', # 已部署 'verified', # 已验证 'failed', # 失败 'rolled_back' # 已回滚 ] def __init__(self, bug_id: str): self.bug_id = bug_id self.start_time = datetime.now() self.logs = [] # 定义状态机 self.machine = Machine( model=self, states=BugFixWorkflow.states, initial='collected', auto_transitions=False ) # 添加转换 self.machine.add_transition('triage', 'collected', 'triaged') self.machine.add_transition('start_location', 'triaged', 'locating') self.machine.add_transition('complete_location', 'locating', 'located') self.machine.add_transition('start_fix', 'located', 'fixing') self.machine.add_transition('complete_fix', 'fixing', 'fixed') self.machine.add_transition('start_test', 'fixed', 'testing') self.machine.add_transition('pass_test', 'testing', 'tested') self.machine.add_transition('fail_test', 'testing', 'fixing') self.machine.add_transition('start_deploy', 'tested', 'deploying') self.machine.add_transition('complete_deploy', 'deploying', 'deployed') self.machine.add_transition('verify', 'deployed', 'verified') self.machine.add_transition('fail', '*', 'failed') self.machine.add_transition('rollback', ['deployed', 'deploying'], 'rolled_back') def log(self, action: str, details: str): """记录日志""" self.logs.append({ "timestamp": datetime.now().isoformat(), "state": self.state, "action": action, "details": details }) def get_progress(self) -> dict: """获取进度信息""" return { "bug_id": self.bug_id, "current_state": self.state, "elapsed_seconds": (datetime.now() - self.start_time).total_seconds(), "log_count": len(self.logs) }

# Skill: git-blame-enhanced (增强版 Git 定位) """ SKILL.md # git-blame-enhanced 增强版 Git Blame，支持 AI 代码归属追踪和跨文件调用链分析。 ## 输入 - file_path: str - 文件路径 - line_number: int - 行号 - include_trace: bool - 是否包含 Agent Trace ## 输出 - author: str - 作者 - commit_hash: str - 提交哈希 - commit_time: datetime - 提交时间 - agent_trace: dict - AI 贡献信息（如有） - call_chain: list - 调用链 - impact_scope: dict - 影响范围 """ import subprocess import json from pathlib import Path class GitBlameEnhanced: def __init__(self, repo_path: str): self.repo_path = Path(repo_path) self.notes_ref = "refs/notes/agent-traces" def execute(self, file_path: str, line_number: int, include_trace: bool = True) -> dict: """执行增强版 Git Blame""" # 1. 标准 Git Blame blame_info = self._standard_blame(file_path, line_number) # 2. 查询 Agent Trace if include_trace: blame_info["agent_trace"] = self._query_agent_trace(blame_info["commit_hash"]) # 3. 分析调用链 blame_info["call_chain"] = self._analyze_call_chain(file_path, line_number) # 4. 评估影响范围 blame_info["impact_scope"] = self._assess_impact(file_path, line_number) return blame_info def _standard_blame(self, file_path: str, line_number: int) -> dict: """执行标准 Git Blame""" cmd = [ 'git', 'blame', '-L', f"{line_number},{line_number}", '--porcelain', '-e', # 显示作者邮箱 file_path ] result = subprocess.run( cmd, cwd=self.repo_path, capture_output=True, text=True ) return self._parse_blame(result.stdout) def _query_agent_trace(self, commit_hash: str) -> dict: """查询 Agent Trace 记录""" cmd = [ 'git', 'notes', '--ref', self.notes_ref, 'show', commit_hash ] result = subprocess.run( cmd, cwd=self.repo_path, capture_output=True, text=True ) if result.returncode == 0: try: return json.loads(result.stdout) except json.JSONDecodeError: return {} return {} def _analyze_call_chain(self, file_path: str, line_number: int) -> list: """分析函数调用链""" # 使用 tree-sitter 或 ctags 分析调用关系 # 简化实现：grep 函数调用 return [] def _assess_impact(self, file_path: str, line_number: int) -> dict: """评估影响范围""" return { "direct_impact": ["function_a", "function_b"], "indirect_impact": ["module_c"], "risk_level": "medium" } def _parse_blame(self, blame_output: str) -> dict: """解析 Git Blame 输出""" info = {} for line in blame_output.splitlines(): if line.startswith("author "): info["author"] = line[7:] elif line.startswith("author-mail "): info["email"] = line[12:].strip('<>') elif line.startswith("author-time "): info["commit_time"] = int(line[12:]) elif not line.startswith(" ") and len(line) >= 40: info["commit_hash"] = line[:40] return info

# Skill 链编排配置（YAML） # skill_chain.yaml chain_name: bugfix_pipeline_v3 version: 3.0.0 description: 全流程 Bug 解决 Skill 链 skills: - name: github-issue-listener version: 2.1.0 config: poll_interval: 30s labels: [bug, critical, high] output_mapping: bug_id: $.id title: $.title description: $.body - name: git-blame-enhanced version: 1.5.0 input_mapping: file_path: $.payload.file_path line_number: $.payload.line_number include_trace: true timeout: 60s - name: claude-code-fix version: 3.2.0 config: model: claude-3-5-sonnet-20260101 max_tokens: 4096 temperature: 0.3 input_mapping: bug_description: $.title code_context: $.payload.code_snippet location_info: $.previous_output - name: pytest-runner version: 2.0.0 config: test_dirs: [tests/unit, tests/integration] coverage_threshold: 80 parallel: true timeout: 300s - name: k8s-deploy version: 1.8.0 config: namespace: staging strategy: canary canary_weight: 10 requires_approval: true - name: audit-logger version: 1.2.0 config: log_level: info storage: elasticsearch retention_days: 90 execute_on: [success, failure] error_handling: max_retries: 3 retry_delay: 5s on_failure: rollback notify_channels: [slack, email] approval_gates: - before_skill: k8s-deploy approvers: [tech_lead, security_team] timeout: 3600s

# 反思与自我优化机制 import json from datetime import datetime from typing import List, Dict class ReflectionEngine: """反思引擎：评估执行效果并优化策略""" def __init__(self): self.experience_db = [] # 经验数据库 self.performance_metrics = {} def reflect(self, workflow_logs: List[dict], outcome: str) -> dict: """执行反思，生成优化建议""" reflection = { "timestamp": datetime.now().isoformat(), "outcome": outcome, "analysis": self._analyze_performance(workflow_logs), "bottlenecks": self._identify_bottlenecks(workflow_logs), "optimizations": [], "lessons_learned": [] } # 生成优化建议 if outcome == "success": reflection["optimizations"] = self._generate_success_optimizations(workflow_logs) reflection["lessons_learned"] = self._extract_best_practices(workflow_logs) else: reflection["optimizations"] = self._generate_failure_fixes(workflow_logs) reflection["lessons_learned"] = self._extract_failure_patterns(workflow_logs) # 存储经验 self.experience_db.append(reflection) return reflection def _analyze_performance(self, logs: List[dict]) -> dict: """分析性能指标""" if not logs: return {} timestamps = [datetime.fromisoformat(log["timestamp"]) for log in logs] durations = [(timestamps[i+1] - timestamps[i]).total_seconds() for i in range(len(timestamps)-1)] return { "total_duration": sum(durations), "avg_step_duration": sum(durations) / len(durations) if durations else 0, "max_step_duration": max(durations) if durations else 0, "step_count": len(logs) } def _identify_bottlenecks(self, logs: List[dict]) -> List[str]: """识别瓶颈步骤""" bottlenecks = [] for log in logs: if log.get("duration", 0) > 300: # 超过 5 分钟 bottlenecks.append(f"慢步骤：{log['action']} ({log['duration']:.1f}s)") if log.get("retries", 0) > 2: bottlenecks.append(f"多次重试：{log['action']} ({log['retries']}次)") return bottlenecks def _generate_success_optimizations(self, logs: List[dict]) -> List[str]: """生成成功场景的优化建议""" optimizations = [] # 分析哪些步骤可以并行 sequential_steps = [log["action"] for log in logs if log.get("sequential")] if len(sequential_steps) > 3: optimizations.append("考虑将部分顺序步骤改为并行执行") return optimizations def _generate_failure_fixes(self, logs: List[dict]) -> List[str]: """生成失败场景的修复建议""" fixes = [] # 查找最后一个成功的步骤 last_success = None for log in reversed(logs): if log.get("status") == "success": last_success = log break if last_success: fixes.append(f"从步骤'{last_success['action']}'后开始重试") return fixes def apply_optimization(self, optimization: str): """应用优化建议到 Skill 链配置""" # 动态调整 Skill 链配置 pass

# 审计日志系统 import logging import json from datetime import datetime from typing import Dict, Any class AuditLogger: """审计日志记录器""" def __init__(self, log_path: str = "/var/log/openclaw/audit.log"): self.logger = logging.getLogger("openclaw.audit") self.logger.setLevel(logging.INFO) # 文件处理器 handler = logging.FileHandler(log_path) handler.setFormatter(logging.Formatter( '%(asctime)s - %(levelname)s - %(message)s' )) self.logger.addHandler(handler) def log_command(self, command: str, agent: str, bug_id: str, status: str, output: str = ""): """记录命令执行""" audit_entry = { "type": "command_execution", "timestamp": datetime.now().isoformat(), "agent": agent, "bug_id": bug_id, "command": command, "status": status, "output": output[:1000] # 限制长度 } self.logger.info(json.dumps(audit_entry)) def log_file_access(self, file_path: str, operation: str, agent: str, bug_id: str): """记录文件访问""" audit_entry = { "type": "file_access", "timestamp": datetime.now().isoformat(), "agent": agent, "bug_id": bug_id, "file_path": file_path, "operation": operation # read/write/delete } self.logger.info(json.dumps(audit_entry)) def log_security_event(self, event_type: str, severity: str, details: Dict[str, Any]): """记录安全事件""" audit_entry = { "type": "security_event", "timestamp": datetime.now().isoformat(), "event_type": event_type, "severity": severity, # low/medium/high/critical "details": details } self.logger.warning(json.dumps(audit_entry)) # 高危事件立即告警 if severity in ["high", "critical"]: self._send_alert(audit_entry) def _send_alert(self, event: dict): """发送安全告警""" # 集成 Slack/钉钉/邮件告警 pass

# main_controller.py - Bugfix Agent 主控制器 import asyncio import logging from typing import Dict, Any from .agents import ( CollectorAgent, LocatorAgent, FixerAgent, TesterAgent, DeployerAgent, AuditorAgent ) from .workflow import BugFixWorkflow from .communication import AgentCommunicationHub, AgentMessage from .reflection import ReflectionEngine logger = logging.getLogger(__name__) class BugfixAgentController: """Bugfix Agent 主控制器""" def __init__(self, config: Dict[str, Any]): self.config = config # 初始化智能体 self.collector = CollectorAgent(config["collector"]) self.locator = LocatorAgent(config["locator"]) self.fixer = FixerAgent(config["fixer"]) self.tester = TesterAgent(config["tester"]) self.deployer = DeployerAgent(config["deployer"]) self.auditor = AuditorAgent(config["auditor"]) # 通信中心 self.comm_hub = AgentCommunicationHub(config["redis_url"]) # 反思引擎 self.reflection = ReflectionEngine() # 工作流跟踪 self.workflows: Dict[str, BugFixWorkflow] = {} async def start(self): """启动控制器""" logger.info("🚀 启动 Bugfix Agent Controller...") # 订阅各智能体消息 for role in ["collector", "locator", "fixer", "tester", "deployer"]: await self._subscribe_to_agent(role) # 启动心跳任务 asyncio.create_task(self._heartbeat()) logger.info("✅ Bugfix Agent Controller 已启动") async def process_bug(self, bug_id: str): """处理单个 Bug 的完整流程""" workflow = BugFixWorkflow(bug_id) self.workflows[bug_id] = workflow try: # 1. 问题收集 workflow.log("collect", "开始收集 Bug 信息") bug_event = await self.collector.collect(bug_id) workflow.triage() # 2. Git 定位 workflow.start_location() location_result = await self.locator.locate(bug_event) workflow.complete_location() # 3. 代码修复 workflow.start_fix() fix_result = await self.fixer.fix(bug_event, location_result) workflow.complete_fix() # 4. 自动化测试 workflow.start_test() test_result = await self.tester.test(fix_result) if test_result.passed: workflow.pass_test() else: workflow.fail_test() return await self._handle_test_failure(workflow, test_result) # 5. 部署上线 workflow.start_deploy() deploy_result = await self.deployer.deploy(fix_result) workflow.complete_deploy() # 6. 验证与归档 workflow.verify() await self.auditor.archive(workflow.logs) # 7. 反思优化 reflection = self.reflection.reflect(workflow.logs, "success") await self._apply_optimizations(reflection) logger.info(f"✅ Bug {bug_id} 处理完成") except Exception as e: logger.error(f"❌ Bug {bug_id} 处理失败：{e}", exc_info=True) workflow.fail() await self._handle_failure(workflow, e) async def _subscribe_to_agent(self, role: str): """订阅智能体消息""" pass async def _heartbeat(self): """心跳任务：定期检查系统状态""" while True: await asyncio.sleep(60) # 检查各智能体健康状态 pass async def _handle_test_failure(self, workflow, test_result): """处理测试失败""" pass async def _handle_failure(self, workflow, error): """处理流程失败""" pass async def _apply_optimizations(self, reflection: dict): """应用优化建议""" for opt in reflection.get("optimizations", []): logger.info(f"🔧 应用优化：{opt}") # 动态调整配置 pass if __name__ == "__main__": config = { "redis_url": "redis://localhost:6379", "collector": {}, "locator": {}, "fixer": {}, "tester": {}, "deployer": {}, "auditor": {} } controller = BugfixAgentController(config) asyncio.run(controller.start())