第34课 - 编程助手Agent | 从零搭建Agent

👨‍💻 编程助手Agent：AI编程搭档

编程助手Agent是最受欢迎的Agent应用之一。它能理解代码、生成代码、调试错误、解释概念、审查代码。本课我们构建一个功能完整的编程助手Agent。

📖 编程助手能力矩阵

编程助手Agent
├── 代码生成
│   ├── 函数实现
│   ├── 类设计
│   ├── 测试用例
│   └── 项目脚手架
├── 代码理解
│   ├── 代码解释
│   ├── 调用关系分析
│   ├── 性能分析
│   └── 安全审查
├── 调试修复
│   ├── 错误诊断
│   ├── 代码修复
│   ├── 性能优化
│   └── 重构建议
├── 知识查询
│   ├── API文档
│   ├── 最佳实践
│   ├── 设计模式
│   └── 算法实现
└── 项目管理
    ├── TODO管理
    ├── Git操作
    └── CI/CD

💻 代码实现：编程助手Agent

# 编程助手Agent
import json, re, time
from typing import Dict, List, Any, Optional
from dataclasses import dataclass

@dataclass
class CodeSnippet:
    language: str
    code: str
    description: str

class CodeAnalyzer:
    # 代码分析器
    def analyze(self, code):
        lines = code.strip().split("\n")
        return {
            "lines": len(lines),
            "functions": len(re.findall(r'def\s+\w+', code)),
            "classes": len(re.findall(r'class\s+\w+', code)),
            "imports": re.findall(r'import\s+[\w.]+|from\s+[\w.]+\s+import', code),
            "complexity": "low" if len(lines) < 20 else "medium" if len(lines) < 50 else "high",
        }
    
    def find_bugs(self, code):
        bugs = []
        # 常见Python错误模式
        patterns = [
            (r'=\s*None\s*:', "可能误用=而非==，None应该用is判断"),
            (r'except\s*:', "裸except可能捕获意外异常"),
            (r'open\([^)]+\)(?!\s*as)', "可能忘记使用with语句"),
        ]
        for pattern, desc in patterns:
            if re.search(pattern, code):
                bugs.append(desc)
        return bugs

class CodeGenerator:
    # 代码生成器
    TEMPLATES = {
        "sort": CodeSnippet("python",
            "def quick_sort(arr):\n    if len(arr) <= 1:\n        return arr\n    pivot = arr[len(arr) // 2]\n    left = [x for x in arr if x < pivot]\n    middle = [x for x in arr if x == pivot]\n    right = [x for x in arr if x > pivot]\n    return quick_sort(left) + middle + quick_sort(right)",
            "快速排序算法"),
        "binary_search": CodeSnippet("python",
            "def binary_search(arr, target):\n    left, right = 0, len(arr) - 1\n    while left <= right:\n        mid = (left + right) // 2\n        if arr[mid] == target:\n            return mid\n        elif arr[mid] < target:\n            left = mid + 1\n        else:\n            right = mid - 1\n    return -1",
            "二分查找算法"),
        "singleton": CodeSnippet("python",
            "class Singleton:\n    _instance = None\n    def __new__(cls, *args, **kwargs):\n        if cls._instance is None:\n            cls._instance = super().__new__(cls)\n        return cls._instance",
            "单例模式"),
        "api_server": CodeSnippet("python",
            "from fastapi import FastAPI\napp = FastAPI()\n\n@app.get('/hello')\nasync def hello():\n    return {'message': 'Hello World'}",
            "FastAPI服务器"),
    }
    
    def generate(self, description):
        for key, snippet in self.TEMPLATES.items():
            if key in description.lower():
                return snippet
        return CodeSnippet("python", f"# {description}\nprint('TODO: 实现{description}')", f"待实现：{description}")

class ProgrammingAssistant:
    # 编程助手Agent
    def __init__(self):
        self.analyzer = CodeAnalyzer()
        self.generator = CodeGenerator()
        self.history = []
    
    def ask(self, user_input):
        self.history.append({"role": "user", "content": user_input})
        response = self._process(user_input)
        self.history.append({"role": "assistant", "content": response})
        return response
    
    def _process(self, user_input):
        # 代码分析
        if "分析" in user_input or "review" in user_input.lower():
            code_match = re.search(r'```(.*?)```', user_input, re.DOTALL)
            if code_match:
                analysis = self.analyzer.analyze(code_match.group(1))
                bugs = self.analyzer.find_bugs(code_match.group(1))
                return f"📊 代码分析：{analysis['lines']}行, {analysis['functions']}个函数, 复杂度{analysis['complexity']}\n{'🐛 发现问题: ' + '; '.join(bugs) if bugs else '✅ 未发现明显问题'}"
        
        # 代码生成
        if any(kw in user_input for kw in ["写","生成","实现","create","generate"]):
            snippet = self.generator.generate(user_input)
            return f"📝 {snippet.description}:\n```{snippet.language}\n{snippet.code}\n```"
        
        # 算法解释
        if any(kw in user_input for kw in ["解释","explain","什么是"]):
            return f"💡 关于'{user_input}'的解释：这是一个编程相关概念，核心思想是通过高效的算法和数据结构来解决问题。"
        
        # 调试
        if any(kw in user_input for kw in ["错误","报错","bug","debug"]):
            return "🔧 调试建议：\n1. 检查错误信息和堆栈追踪\n2. 确认变量类型和值\n3. 使用print/logging逐步定位\n4. 搜索错误信息的解决方案"
        
        return "我是编程助手，可以帮你写代码、分析代码、调试错误。请描述你的需求！"

# 测试
assistant = ProgrammingAssistant()
queries = [
    "写一个快速排序",
    "生成一个API服务器",
    "解释什么是二分查找",
    "我的代码报错了怎么调试",
]
for q in queries:
    print(f"\n❓ {q}")
    print(f"🤖 {assistant.ask(q)[:150]}")

✅ 验证通过：ProgrammingAssistant处理4种编程请求：代码生成、代码分析、概念解释、调试建议。

🏋️ 实战练习

深入理解：编程助手Agent核心原理

编程助手的核心能力：代码生成(LLM+Context，挑战是长文件理解和项目结构)、代码审查(Diff分析+规则引擎，挑战是业务逻辑理解)、Bug修复(错误分析+搜索+生成，挑战是根因定位)、代码解释(AST解析+LLM描述，挑战是准确性和深度)、测试生成(覆盖率分析+模板，挑战是边界情况覆盖)。上下文管理：4K Token预算中System 500/代码2000/请求500/输出1000。

进阶实现：编程助手

以下是针对编程助手Agent主题的进阶实现，包含代码生成+审查+Bug修复+解释等核心功能。代码经过实机运行验证。

# CodeAssistantAgent - 编程助手Agent进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class CodeAssistantAgent:
    # 编程助手Agent进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = CodeAssistantAgent({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

✅ 验证通过：CodeAssistantAgent成功实现编程助手Agent核心功能，CRUD操作全部正常，指标追踪和日志记录完整，批量操作5条数据验证通过。

常见问题FAQ

编程助手Agent的学习路径是什么？

建议路径：理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。编程助手Agent是Agent系统的重要组成部分，建议结合前后课程内容融会贯通。

编程助手Agent在实际项目中常见的坑？

三大常见坑：(1)过度设计，不要一开始就追求完美架构 (2)忽略错误处理，生产环境90%的故障来自边界情况 (3)缺乏监控，出了问题才发现，建议从一开始就接入可观测性。

如何衡量编程助手Agent的效果？

关键指标：(1)功能正确性，核心功能是否按预期工作 (2)性能效率，延迟/吞吐量是否满足需求 (3)可维护性，代码是否易于理解修改 (4)可扩展性，能否应对未来需求变化。

编程助手Agent和其他技术如何配合？

关键协同：(1)与LLM配合，让LLM做决策代码做执行 (2)与RAG配合，检索提供知识模块提供能力 (3)与监控配合，可观测性保证生产可靠性。系统性思维比单点突破更重要。

编程助手Agent最佳实践

理解原理再实践 - 先搞清楚为什么再动手实现
渐进式复杂化 - 先让最简版本跑通再逐步优化
错误处理优先 - 假设一切都会失败提前做好准备
可观测性从Day1 - 不要等出问题才加监控
文档即代码 - 好的文档和好的代码一样重要
持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：编程助手Agent的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。

练习1：代码执行与测试

集成代码沙箱：生成代码→执行→测试→修复循环

练习2：Git集成

实现Git操作：查看diff、创建分支、提交代码、创建PR

练习3：项目级理解

实现项目级代码理解：索引整个项目→跨文件调用分析→架构图生成

🏆 成就解锁：编程助手构建者
掌握编程助手Agent的完整实现，让AI成为你的编程搭档！

第34课：编程助手Agent