AI编码Agent实战指南：从Issue到PR的自动化工程实践

# Agent PR 审查框架
AGENT_PR_REVIEW_CHECKLIST = {
    "任务理解": {
        "是否正确理解了Issue的真实目标": "必查",
        "是否识别了隐含的约束条件": "必查",
        "是否考虑了边界情况和异常路径": "必查"
    },
    "代码变更范围": {
        "是否修改了无关文件": "必查",
        "是否删除了必要的现有代码": "必查",
        "变更范围是否与Issue匹配": "必查",
        "是否产生了不必要的大规模重构": "建议查"
    },
    "测试覆盖": {
        "是否新增了必要的测试": "必查",
        "现有测试是否仍然通过": "必查",
        "是否测试了边界情况": "建议查",
        "测试是否有意义（非无意义的占位测试）": "必查"
    },
    "执行过程": {
        "跑了哪些命令": "必查",
        "是否遇到过失败，如何修复": "必查",
        "是否进行了不必要的探索性操作": "建议查"
    },
    "依赖与影响": {
        "是否引入了新依赖": "必查",
        "新依赖是否必要且安全": "必查",
        "是否改变了API接口": "必查",
        "是否改变了数据结构": "必查"
    },
    "可观测性": {
        "是否添加了必要的日志": "必查",
        "是否更新了监控指标": "建议查",
        "是否留下了不确定性标记": "建议查"
    },
    "人工决策点": {
        "是否标注了需要人工确认的决策": "必查",
        "是否在不确定时选择了保守方案": "建议查",
        "是否避免了危险的自动化操作": "必查"
    }
}

def review_agent_pr(pr_data, agent_log):
    """审查Agent提交的PR"""
    findings = []
    
    for category, items in AGENT_PR_REVIEW_CHECKLIST.items():
        for check, level in items.items():
            result = perform_check(check, pr_data, agent_log)
            if result['status'] == 'fail':
                findings.append({
                    'category': category,
                    'check': check,
                    'level': level,
                    'detail': result['detail']
                })
    
    return {
        'passed': len([f for f in findings if f['level'] == '必查']) == 0,
        'warnings': [f for f in findings if f['level'] == '建议查'],
        'blockers': [f for f in findings if f['level'] == '必查'],
        'total_checks': len(findings)
    }

# 日志规范示例代码
import structlog
logger = structlog.get_logger()

# 正确的日志方式
logger.info("user_registered", 
    user_id=user.id, 
    source="api", 
    plan="pro")

# 错误的日志方式（禁止使用）
# print(f"User {user.id} registered from API with pro plan")

# 异常处理中的日志
try:
    process_payment(order)
except PaymentError as e:
    logger.error("payment_failed",
        order_id=order.id,
        error_type=type(e).__name__,
        error_message=str(e),
        amount=order.amount
    )
    raise

# 在Agent配置中注入日志规则
AGENT_SYSTEM_PROMPT = (
    "你是一个专业的后端开发Agent。在编写代码时，你必须遵守以下规则：
"
    "1. [强制] 所有生成的代码必须使用结构化日志
"
    "2. [强制] 所有函数入口和关键分支必须有INFO级别日志
"
    "3. [强制] 所有异常捕获必须有ERROR级别日志，包含完整上下文
"
    "4. [强制] 禁止使用print()进行日志输出
"
    "5. [建议] 在关键业务逻辑中添加DEBUG级别日志用于排查

"
    "如果你不确定如何实现某个日志点，请在代码中添加注释标记：
"
    "# REVIEW: 此处日志需要人工确认是否合适"
)

# 在Agent的代码生成流程中验证日志规范
def validate_logging(code):
    """验证生成的代码是否符合日志规范"""
    violations = []
    
    # 检查是否使用了print
    if 'print(' in code and 'logging' not in code:
        violations.append("使用了print而非logging")
    
    # 检查异常处理中是否有日志
    if 'except' in code:
        lines = code.split('
')
        in_except = False
        has_log_in_except = False
        for line in lines:
            if 'except' in line:
                in_except = True
                has_log_in_except = False
            elif in_except:
                if 'logger' in line or 'log' in line:
                    has_log_in_except = True
                if line.strip() == '' or line.strip().startswith('except'):
                    if not has_log_in_except:
                        violations.append("异常处理中缺少日志")
                    in_except = False
    
    return {
        'passed': len(violations) == 0,
        'violations': violations
    }

# 会话管理最佳实践
import time

class AgentSessionManager:
    """AI编码Agent的会话管理器"""
    
    def __init__(self, max_context_tokens=100000):
        self.max_context = max_context_tokens
        self.sessions = {}
    
    def create_session(self, task_type, task_description):
        """为每个独立任务创建新会话"""
        session_id = f"{task_type}_{int(time.time())}"
        
        # 加载任务相关的上下文
        context = self._build_task_context(task_type, task_description)
        
        self.sessions[session_id] = {
            'task': task_description,
            'context': context,
            'messages': [],
            'token_count': len(context) // 4  # 粗略估算
        }
        return session_id
    
    def _build_task_context(self, task_type, description):
        """根据任务类型构建精准的上下文"""
        context_parts = []
        
        # 1. 系统提示（固定）
        context_parts.append(self._get_system_prompt(task_type))
        
        # 2. 项目约定（按需加载）
        if task_type == "feature":
            context_parts.append(self._get_project_conventions())
            context_parts.append(self._get_relevant_api_docs(description))
        
        # 3. 相关代码文件（精准检索）
        relevant_files = self._find_relevant_files(description)
        for f in relevant_files[:5]:  # 最多5个文件
            context_parts.append(f"--- {f['path']} ---
{f['content'][:5000]}")
        
        # 4. 相关历史决策
        past_decisions = self._get_related_decisions(description)
        if past_decisions:
            context_parts.append("## 相关历史决策
" + past_decisions)
        
        return "

".join(context_parts)
    
    def add_message(self, session_id, role, content):
        """添加消息并监控上下文使用量"""
        session = self.sessions[session_id]
        session['messages'].append({'role': role, 'content': content})
        session['token_count'] += len(content) // 4
        
        # 上下文使用超过70%时发出警告
        if session['token_count'] > self.max_context * 0.7:
            print(f"[警告] 会话 {session_id} 上下文使用率超过70%")
        
        # 超过90%时强制摘要
        if session['token_count'] > self.max_context * 0.9:
            self._summarize_and_compress(session_id)

# 任务拆分示例
def decompose_task(issue_description):
    """将复杂Issue拆分为可独立执行的子任务"""
    
    complexity = analyze_complexity(issue_description)
    
    if complexity['score'] < 3:
        # 简单任务，单会话完成
        return [issue_description]
    
    # 复杂任务，拆分为子任务
    subtasks = [
        {
            "id": "analysis",
            "description": "分析需求并设计实现方案",
            "session_type": "planning",
            "depends_on": []
        },
        {
            "id": "data_model",
            "description": "根据方案实现数据模型变更",
            "session_type": "implementation",
            "depends_on": ["analysis"]
        },
        {
            "id": "api",
            "description": "实现API端点和业务逻辑",
            "session_type": "implementation",
            "depends_on": ["analysis", "data_model"]
        },
        {
            "id": "tests",
            "description": "编写单元测试和集成测试",
            "session_type": "testing",
            "depends_on": ["api"]
        },
        {
            "id": "docs",
            "description": "更新API文档和变更日志",
            "session_type": "documentation",
            "depends_on": ["api"]
        }
    ]
    
    return subtasks

# Agent效能度量框架
def measure_agent_effectiveness(metrics):
    """度量AI编码Agent的效能"""
    return {
        '自动化率': metrics['agent_prs'] / metrics['total_prs'] * 100,
        '首次通过率': metrics['first_pass_approved'] / metrics['agent_prs'] * 100,
        '平均审查轮次': metrics['total_review_rounds'] / metrics['agent_prs'],
        '引入缺陷率': metrics['post_merge_bugs'] / metrics['agent_prs'] * 100,
        '时间节省': (metrics['human_avg_time'] - metrics['agent_avg_time']) / metrics['human_avg_time'] * 100,
        '成本节省': (metrics['human_cost'] - metrics['agent_cost']) / metrics['human_cost'] * 100
    }