AI Coding Agent 工程化实战：从单 Agent 到多 Agent 协作的五大架构模式与生产级代码实现

前言

2026年，AI编程已经不再是"辅助补全代码"的玩具，而是真正走进了生产流水线。

Claude Code 100%自编写、Devin 能独立完成完整功能模块、Cursor 的 Agent 模式可以直接替你接手一个 GitHub Issue……这些不是营销噱头，是真实发生在无数工程团队里的日常。但真正的问题是：当 AI Coding Agent 从 demo 走向生产，你的系统能接住吗？

这篇文章从真实工程问题出发，拆解 AI Coding Agent 的五种核心架构模式，配上可直接运行的代码，展示如何从单 Agent 扩展到多 Agent 协作，以及如何处理成本控制、可靠性保障等实际问题。

一、为什么 AI Coding Agent 工程化这么难？

1.1 两种范式的根本差异

传统代码补全工具是单轮响应系统：你给一个 prompt，AI 返回一段代码，任务结束。

AI Coding Agent 是多轮自主执行系统：你给一个目标，AI 自主决定调用哪些工具、阅读哪些文件、修改哪些代码，可能经历几十甚至上百次工具调用才能完成任务。

传统补全：
User → [Prompt] → LLM → [Code] → Done

AI Coding Agent：
User → [Goal] → Agent → [Plan] → [Tool: read_file]
       → [Tool: grep] → [Tool: write_file] → [Tool: test]
       → [Loop: 10-100次] → [Done]

1.2 生产环境的五大核心挑战

挑战一：工具调用的正确性

LLM 调用工具时，会犯两类错误：工具选择错误（如应该用 grep 找代码，却调用了 read_file 读取整个仓库）和工具参数错误。

挑战二：长程任务的状态管理

修一个 bug 可能跨越几十个工具调用，涉及几百个文件。上一次调用的结果，需要成为下一次调用的上下文。

挑战三：结果的可验证性

AI 说"我已经修好了 bug"，你信吗？生产环境需要：可验证、可回滚、可审计。

挑战四：成本控制

一次复杂任务可能消耗数万 token，如果不加控制，轻松烧掉大量预算。

挑战五：可靠性保障

LLM 不是确定性的，需要处理：输出格式不稳定（JSON 解析失败）、调用超时、API 限流（429 错误）。

二、架构模式一：单 Agent 直接执行

最简单也最常见的模式：单个 Agent 接收任务，直接执行，用一个循环不断调用工具直到任务完成。

2.1 Python 实现：最简 Agent 框架

import json, re, time, subprocess
from dataclasses import dataclass, field
from typing import Optional
from abc import ABC, abstractmethod

@dataclass
class ToolResult:
    success: bool
    output: str
    error: Optional[str] = None
    execution_time_ms: float = 0

class Tool(ABC):
    name: str = ""
    description: str = ""
    parameters: dict = field(default_factory=dict)
    
    @abstractmethod
    def execute(self, **kwargs) -> ToolResult: pass
    
    def to_schema(self) -> dict:
        return {"type": "function", "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters
        }}

# 文件系统工具
class ReadFileTool(Tool):
    name = "read_file"
    description = "读取文件内容，用于理解代码、配置、文档。"
    parameters = {"type": "object", "properties": {
        "path": {"type": "string"},
        "offset": {"type": "integer", "default": 0},
        "limit": {"type": "integer", "default": 500}
    }, "required": ["path"]}
    
    def __init__(self, root): self.root = root
    def execute(self, path, offset=0, limit=500):
        from pathlib import Path
        try:
            lines = Path(self.root)/path
            content = lines.read_text(errors="ignore").split("\n")
            return ToolResult(success=True, output="\n".join(content[offset:offset+limit]))
        except Exception as e:
            return ToolResult(success=False, output="", error=str(e))

class WriteFileTool(Tool):
    name = "write_file"
    description = "写入或覆盖文件内容。会自动创建父目录。"
    parameters = {"type": "object", "properties": {
        "path": {"type": "string"}, "content": {"type": "string"}
    }, "required": ["path", "content"]}
    
    def __init__(self, root): self.root = root
    def execute(self, path, content):
        from pathlib import Path
        try:
            f = Path(self.root)/path; f.parent.mkdir(parents=True, exist_ok=True)
            f.write_text(content)
            return ToolResult(success=True, output=f"写入成功: {path}")
        except Exception as e:
            return ToolResult(success=False, output="", error=str(e))

class BashTool(Tool):
    name = "bash"
    description = "执行 bash 命令，用于运行测试、构建等。超时上限 120 秒。"
    parameters = {"type": "object", "properties": {
        "command": {"type": "string"},
        "cwd": {"type": "string", "default": "."},
        "timeout": {"type": "integer", "default": 60}
    }, "required": ["command"]}
    
    def execute(self, command, cwd=".", timeout=60):
        try:
            r = subprocess.run(command, shell=True, cwd=cwd,
                             capture_output=True, text=True, timeout=timeout)
            return ToolResult(success=r.returncode==0, output=r.stdout[:5000],
                            error=r.stderr[:1000] if r.stderr else None)
        except subprocess.TimeoutExpired:
            return ToolResult(success=False, output="", error=f"超时 {timeout}s")

# LLM 客户端
@dataclass
class LLMResponse:
    content: str; tool_calls: list; usage: dict; model: str; finish_reason: str

class LLMClient(ABC):
    @abstractmethod
    def chat(self, messages: list, tools: list, temperature=0.2) -> LLMResponse: pass

class OpenAIClient(LLMClient):
    def __init__(self, api_key, model="gpt-4o", base_url=None):
        self.api_key = api_key; self.model = model
        self.base_url = base_url or "https://api.openai.com/v1"
    
    def chat(self, messages, tools=None, temperature=0.2):
        import urllib.request, urllib.error
        payload = {"model": self.model, "messages": messages, "temperature": temperature}
        if tools: payload["tools"] = tools; payload["tool_choice"] = "auto"
        body = json.dumps(payload).encode()
        req = urllib.request.Request(f"{self.base_url}/chat/completions",
                                     data=body, method="POST")
        req.add_header("Authorization", f"Bearer {self.api_key}")
        req.add_header("Content-Type", "application/json")
        with urllib.request.urlopen(req, timeout=120) as resp:
            data = json.loads(resp.read())
        msg = data["choices"][0]["message"]
        tool_calls = [{"id": t["id"], "name": t["function"]["name"],
                       "arguments": t["function"]["arguments"]}
                      for t in msg.get("tool_calls", [])]
        return LLMResponse(content=msg.get("content",""), tool_calls=tool_calls,
                          usage=data.get("usage",{}), model=data.get("model",""),
                          finish_reason=data["choices"][0].get("finish_reason",""))

@dataclass
class AgentMessage:
    role: str; content: str
    tool_call_id: Optional[str] = None; tool_name: Optional[str] = None

@dataclass
class AgentConfig:
    max_iterations: int = 50
    max_tool_calls_per_iter: int = 5
    temperature: float = 0.2
    system_prompt: str = ""

class SingleAgent:
    def __init__(self, llm: LLMClient, tools: list[Tool], config: AgentConfig = None):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.config = config or AgentConfig()
        self.messages: list[AgentMessage] = []
        self.tool_schemas = [t.to_schema() for t in tools]
    
    def _llm_messages(self) -> list:
        result = []
        for m in self.messages:
            if m.role == "system": result.append({"role":"system","content":m.content})
            elif m.role == "user": result.append({"role":"user","content":m.content})
            elif m.role == "assistant":
                if m.tool_call_id:
                    result.append({"role":"assistant","tool_calls":[{
                        "id":m.tool_call_id,"type":"function",
                        "function":{"name":m.tool_name,"arguments":m.content}}]})
                else: result.append({"role":"assistant","content":m.content})
            elif m.role == "tool":
                result.append({"role":"tool","tool_call_id":m.tool_call_id,"content":m.content})
        return result
    
    def _parse_args(self, name, args_str):
        try: return json.loads(args_str)
        except json.JSONDecodeError:
            fixed = re.sub(r',(\s*[}]])', r'\1', args_str.strip())
            try: return json.loads(fixed)
            except: return {"_raw": args_str}
    
    def run(self, task: str, context: dict = None) -> dict:
        sys_prompt = self.config.system_prompt or """你是一个专业的 AI Coding Agent。
可用工具：read_file(读文件)、write_file(写文件)、bash(执行命令)。
工作原则：
1. 先理解任务，再制定计划
2. 优先用 grep/read 定位代码，不要盲目读整个仓库
3. 修改前先读原文件
4. 写完代码后运行测试
5. 每次只做一件明确的事"""
        if context: sys_prompt += f"\n\n## 当前上下文\n{json.dumps(context,ensure_ascii=False)}"
        self.messages = [AgentMessage("system", sys_prompt), AgentMessage("user", task)]
        
        total_usage = {"prompt_tokens":0,"completion_tokens":0}
        tool_count = 0
        
        for i in range(self.config.max_iterations):
            try: resp = self.llm.chat(self._llm_messages(), self.tool_schemas, self.config.temperature)
            except Exception as e:
                self.messages.append(AgentMessage("user", f"LLM 调用失败: {e}，请重试"))
                continue
            
            for k,v in resp.usage.items(): total_usage[k] = total_usage.get(k,0)+v
            
            if not resp.tool_calls:
                self.messages.append(AgentMessage("assistant", resp.content))
                return {"success":True,"final_message":resp.content,
                        "iterations":i+1,"tool_calls":tool_count,
                        "cost_usd":(total_usage.get("prompt_tokens",0)*0.0000025+total_usage.get("completion_tokens",0)*0.00001),
                        "messages":self.messages}
            
            for tc in resp.tool_calls[:self.config.max_tool_calls_per_iter]:
                tool_count += 1
                args = self._parse_args(tc["name"], tc["arguments"])
                self.messages.append(AgentMessage("assistant", tc["arguments"], tc["id"], tc["name"]))
                
                if tc["name"] not in self.tools:
                    r = ToolResult(False, "", f"未知工具: {tc['name']}")
                else:
                    try: r = self.tools[tc["name"]].execute(**args)
                    except Exception as e: r = ToolResult(False, "", str(e))
                
                result_text = f"[{r.output}]" if r.success else f"[ERROR: {r.error}]"
                self.messages.append(AgentMessage("tool", result_text[:3000], tc["id"]))
        
        return {"success":False,"final_message":f"达到最大迭代次数 {self.config.max_iterations}",
                "iterations":self.config.max_iterations,"tool_calls":tool_count,
                "cost_usd":total_usage.get("prompt_tokens",0)*0.0000025+total_usage.get("completion_tokens",0)*0.00001,
                "messages":self.messages}

# ===== 使用示例 =====
# import os
# llm = OpenAIClient(api_key=os.environ["OPENAI_API_KEY"])
# agent = SingleAgent(llm, [ReadFileTool("/project"), WriteFileTool("/project"), BashTool()],
#                     AgentConfig(max_iterations=30))
# result = agent.run("修复 src/auth/login.py 中的 session 过期 bug")
# print(f"成功: {result['success']}, 迭代: {result['iterations']}, 成本: ${result['cost_usd']:.4f}")

核心流程：构建消息历史 → 调用 LLM → 解析工具调用 → 执行工具 → 将结果写回消息历史 → 循环直到 LLM 不再调用工具。

三、架构模式二：Harness 模式——带质量门的多 Agent 编排

Harness Engineering 理念的核心是：不要让单个 Agent 决定任务是否完成，而是让流水线中的多个 Agent 相互校验。

3.1 三 Agent 流水线实现

from enum import Enum
import uuid, re

class AgentRole(Enum):
    PLANNER = "planner"   # 任务规划与分解
    CODER = "coder"       # 代码编写
    REVIEWER = "reviewer" # 代码审查

ROLE_PROMPTS = {
    AgentRole.PLANNER: """你是一个任务规划专家。
职责：1. 理解需求，分解为可执行的小任务；2. 指定验收标准；3. 确定执行顺序
输出格式：[PLAN]
{"tasks":[{"id":"task-1","description":"...","acceptance_criteria":["..."]}]}
[/PLAN]""",
    
    AgentRole.CODER: """你是一个专业程序员。
职责：1. 严格按验收标准编写代码；2. 遵循项目规范；3. 确保可运行有测试
每次只实现一个明确的功能。""",
    
    AgentRole.REVIEWER: """你是资深代码审查专家（15年经验）。
职责：严格审查代码质量和正确性，发现潜在 bug、安全漏洞、性能问题
输出：[REVIEW]
{"passed":true/false,"issues":[{"severity":"critical/major/minor","description":"...","suggestion":"..."}]}
[/REVIEW]"""
}

@dataclass
class HarnessResult:
    task_id: str; success: bool; total_iterations: int
    total_cost_usd: float; quality_score: float; errors: list

class HarnessPipeline:
    def __init__(self, llm_factory, tools, max_retries=2):
        self.llm_factory = llm_factory
        self.tools = tools
        self.max_retries = max_retries
        self.role_agents = {
            r: SingleAgent(llm_factory(), tools, AgentConfig(max_iterations=20))
            for r in AgentRole
        }
    
    def run(self, task_desc: str, criteria: list[str]) -> HarnessResult:
        task_id = str(uuid.uuid4())[:8]
        errors = []
        
        # 阶段1：规划
        plan_msg = f"任务：{task_desc}\n验收标准：\n" + "\n".join(f"- {c}" for c in criteria)
        plan_result = self.role_agents[AgentRole.PLANNER].run(plan_msg)
        if not plan_result["success"]:
            return HarnessResult(task_id, False, 1, 0, 0, ["规划失败"])
        
        sub_tasks = self._parse_tasks(plan_result["final_message"])
        total_cost = plan_result["cost_usd"]
        total_iter = 1
        
        # 阶段2-3：编码 + 审查（每个子任务）
        for st in sub_tasks:
            task_msg = f"任务：{st['description']}\n验收标准：\n" + "\n".join(f"- {c}" for c in st.get("acceptance_criteria",[]))
            
            # Coder
            coder = self.role_agents[AgentRole.CODER]
            coder_r = coder.run(task_msg)
            total_cost += coder_r["cost_usd"]
            total_iter += coder_r["iterations"]
            
            if not coder_r["success"]:
                errors.append(f"子任务 {st['id']} 编码失败"); continue
            
            # Reviewer
            reviewer = self.role_agents[AgentRole.REVIEWER]
            review_r = reviewer.run(f"审查代码：\n{coder_r['final_message']}\n\n任务：{st['description']}")
            total_cost += review_r["cost_usd"]
            total_iter += review_r["iterations"]
            
            passed = self._check_review_passed(review_r["final_message"])
            if not passed:
                issues = re.findall(r'"description":\s*"([^"]+)"', review_r["final_message"])
                errors.append(f"子任务 {st['id']} 审查问题：{issues[:2]}")
        
        return HarnessResult(task_id, len(errors)==0, total_iter, total_cost,
                            1.0 if len(errors)==0 else 0.5, errors)
    
    def _parse_tasks(self, text: str) -> list:
        m = re.search(r'\[PLAN\](.*?)\[/PLAN\]', text, re.DOTALL)
        if not m: return [{"id":"main","description":text[:500],"acceptance_criteria":[]}]
        try: return json.loads(m.group(1)).get("tasks",[])
        except: return [{"id":"main","description":text[:500],"acceptance_criteria":[]}]
    
    def _check_review_passed(self, text: str) -> bool:
        m = re.search(r'passed["\']?\s*:\s*(true|false)', text, re.IGNORECASE)
        return m and m.group(1).lower()=="true"

# ===== 使用示例 =====
# pipeline = HarnessPipeline(lambda: llm, [ReadFileTool("/p"), WriteFileTool("/p"), BashTool()])
# result = pipeline.run("实现 JWT 认证服务",
#     ["所有接口有单元测试","密码 bcrypt hash","JWT secret 从环境变量读"])
# print(f"成功: {result.success}, 迭代: {result.total_iterations}, 成本: ${result.total_cost_usd:.4f}")

Harness 模式让 Planner（规划）、Coder（编码）、Reviewer（审查）三个角色相互协作，每个子任务都经过编码→审查的校验循环，不通过则记录问题继续，最终给出质量评分。

四、架构模式三：上下文压缩与长期记忆

随着工具调用次数增加，消息历史越来越长，最终超出 LLM 的上下文窗口。

4.1 三层记忆架构

Layer 3: Semantic Memory（语义记忆）
  SQLite FTS5 索引项目代码，支持快速检索相关文件
  
Layer 2: Conversation History（对话历史）
  自动压缩/摘要早期消息，保留最近 N 条
  
Layer 1: Tool Results Cache（工具结果缓存）
  LRU + TTL 缓存，避免重复读取相同文件

4.2 上下文管理器实现

import time, hashlib, json
from collections import OrderedDict

class ContextWindow:
    """智能上下文窗口：监控大小，自动压缩早期消息"""
    def __init__(self, max_tokens=120000):
        self.max_tokens = max_tokens
        self.messages: list[AgentMessage] = []
    
    def _tokens(self, text: str) -> int:
        cjk = sum(1 for c in text if ord(c) > 127)
        return int(cjk * 0.5 + (len(text)-cjk) * 0.25)
    
    def add(self, msg: AgentMessage):
        self.messages.append(msg)
        self._compress()
    
    def _compress(self):
        total = sum(self._tokens(m.content) for m in self.messages)
        if total < self.max_tokens: return
        
        # 保留系统消息 + 最近 10 条
        keep = max(1, len(self.messages)-10)
        if keep <= 1: return
        
        mid = self.messages[1:keep]
        
        # 统计摘要
        tools = [m.tool_name for m in mid if m.tool_name]
        reads = [m.content[:60] for m in mid if m.tool_name=="read_file"]
        decisions = [m.content[:100] for m in mid if m.role=="assistant" and not m.tool_call_id]
        
        parts = []
        if tools: parts.append(f"执行了 {len(set(tools))} 种工具共 {len(tools)} 次")
        if reads: parts.append(f"阅读了 {len(set(reads))} 个文件")
        if decisions: parts.append(f"做出 {len(decisions)} 个决策")
        
        summary = "; ".join(parts) if parts else "无重要操作"
        
        self.messages = (
            [self.messages[0]] +
            [AgentMessage("system", f"[早期对话摘要] {summary}")] +
            self.messages[keep:]
        )
    
    def get(self) -> list[AgentMessage]: return self.messages
    def total_tokens(self) -> int: return sum(self._tokens(m.content) for m in self.messages)

class ToolResultCache:
    """工具结果缓存：LRU + TTL，避免重复读文件"""
    def __init__(self, ttl=1800, max_entries=5000):
        self.cache: OrderedDict = OrderedDict()
        self.ttl = ttl; self.max_entries = max_entries
    
    def get(self, key: str):
        if key not in self.cache: return None
        entry = self.cache[key]
        if time.time()-entry["ts"] < self.ttl:
            self.cache.move_to_end(key); return entry["content"]
        del self.cache[key]; return None
    
    def set(self, key: str, content: str):
        self.cache[key] = {"content": content, "ts": time.time()}
        self.cache.move_to_end(key)
        if len(self.cache) > self.max_entries:
            for _ in range(int(self.max_entries*0.1)): self.cache.popitem(last=False)

class SemanticIndex:
    """语义索引：SQLite FTS5 对项目代码建立全文索引"""
    def __init__(self, db_path="/tmp/semantic_index.db"):
        import sqlite3
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS chunks (
                id INTEGER PRIMARY KEY, path TEXT, chunk TEXT,
                summary TEXT, symbols TEXT, modified REAL, idx INTEGER)""")
        self.conn.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS fts USING fts5(
                path, summary, symbols, content=chunks, content_rowid=id)""")
        self.conn.commit()
    
    def index(self, path: str, content: str, modified: float):
        lines = content.split("\n")
        cs = 200  # chunk size
        c = self.conn.cursor()
        c.execute("DELETE FROM chunks WHERE path=?", (path,))
        for i, chunk in enumerate(["\n".join(lines[j:j+cs]) for j in range(0,len(lines),cs)]):
            syms = self._extract_syms(chunk)
            smry = self._gen_summary(chunk)
            c.execute("INSERT INTO chunks VALUES(NULL,?,?,?,?,?,?,?)",
                     (path,chunk,smry,",".join(syms),modified,i))
        self.conn.commit()
    
    def search(self, query: str, top_k=5) -> list:
        cur = self.conn.execute(
            "SELECT path,chunk,summary FROM fts WHERE fts MATCH ? ORDER BY rank LIMIT ?",
            (query, top_k))
        return [{"path":r[0],"chunk":r[1],"summary":r[2]} for r in cur.fetchall()]
    
    def _extract_syms(self, code: str) -> list:
        import re
        syms = set()
        for p in [r'(?:def|class)\s+([a-zA-Z_]\w*)', r'([A-Z][a-zA-Z0-9_]+)\s*[=\(]']:
            syms.update(re.findall(p, code))
        return list(syms)[:20]
    
    def _gen_summary(self, code: str) -> str:
        import re
        fs = re.findall(r'def\s+(\w+)', code)
        cs = re.findall(r'class\s+(\w+)', code)
        p = []
        if cs: p.append(f"类: {','.join(cs)}")
        if fs: p.append(f"函数: {','.join(fs[:5])}")
        return "; ".join(p) if p else code[:100]
    
    def close(self): self.conn.close()

五、架构模式四：MCP 协议——标准化工具接入

如果每个系统都写一套专用工具适配器，维护成本是灾难性的。MCP（Model Context Protocol）就是来解决这个问题的。

5.1 MCP Client 实现

import json, subprocess, select
from dataclasses import dataclass
from typing import Optional

@dataclass
class MCPTool:
    name: str; description: str; input_schema: dict

class MCPClient:
    """MCP 客户端：通过 stdio 或 HTTP SSE 连接 MCP Host"""
    def __init__(self, command=None, url=None, env=None):
        self.command = command; self.url = url
        self.env = env or {}; self.process = None
        self.req_id = 0; self.tools: list[MCPTool] = []
        self._connected = False
    
    def connect(self, timeout=30):
        if self.command:
            self.process = subprocess.Popen(
                self.command, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                stderr=subprocess.PIPE, env={**subprocess.os.environ,**self.env},
                text=True, bufsize=1)
        elif self.url:
            import threading
            self._thread = threading.Thread(target=self._sse_reader, args=(self.url,timeout), daemon=True)
            self._thread.start()
        
        self._send({"jsonrpc":"2.0","method":"initialize","params":{
            "protocolVersion":"2024-11-05","clientInfo":{"name":"agent","version":"1.0"}},"id":1})
        r = self._send({"jsonrpc":"2.0","method":"tools/list","id":2})
        self.tools = [MCPTool(t["name"],t.get("description",""),t.get("inputSchema",{}))
                      for t in r.get("tools",[])]
        self._connected = True
    
    def _send(self, req: dict) -> dict:
        self.req_id += 1
        req["id"] = self.req_id
        if self.process:
            self.process.stdin.write(json.dumps(req)+"\n"); self.process.stdin.flush()
            while True:
                ready,_,_ = select.select([self.process.stdout],[],[],30)
                if ready:
                    res = json.loads(self.process.stdout.readline())
                    if res.get("id") == self.req_id:
                        return res.get("result",res.get("error",{}))
        return {}
    
    def _sse_reader(self, url, timeout):
        import requests
        try:
            with requests.get(url+"/sse", stream=True, timeout=timeout) as r:
                for line in r.iter_lines():
                    if line.startswith(b"data: "):
                        self._handle(json.loads(line[6:]))
        except: pass
    
    def _handle(self, data): pass  # 处理通知
    
    def call(self, name: str, args=None) -> dict:
        return self._send({"jsonrpc":"2.0","method":"tools/call",
                          "params":{"name":name,"arguments":args or {}},"id":self.req_id+1})
    
    def close(self):
        if self.process:
            self.process.terminate(); self.process.wait(timeout=5)
        self._connected = False

# 在 SingleAgent 中集成 MCP 工具
class MCPAgent(SingleAgent):
    def __init__(self, llm, mcp_clients: list[MCPClient], config=None):
        wrapped = []
        for client in mcp_clients:
            for t in client.tools:
                mcp_t = self._wrap(client, t)
                if mcp_t: wrapped.append(mcp_t)
        super().__init__(llm, wrapped, config)
    
    def _wrap(self, client, mcp_tool):
        class W(Tool):
            name = mcp_tool.name; description = mcp_tool.description
            parameters = mcp_tool.input_schema
            def execute(self, **kw):
                r = client.call(self.name, kw)
                ok = not r.get("isError")
                parts = [c["text"] for c in r.get("content",[]) if c.get("type")=="text"]
                return ToolResult(ok, "\n".join(parts),
                                r.get("isError") and str(r) or None)
        return W()

# 使用示例：
# mcp_fs = MCPClient(command=["./mcp-filesystem","/workspace"])
# mcp_gh = MCPClient(command=["./mcp-github","--token",os.environ["GITHUB_TOKEN"]])
# mcp_fs.connect(); mcp_gh.connect()
# agent = MCPAgent(llm, [mcp_fs, mcp_gh])
# result = agent.run("分析这个 GitHub 项目的代码结构")

六、架构模式五：成本控制与可靠性保障

6.1 自适应成本路由

from enum import Enum
from dataclasses import dataclass

class Tier(Enum):
    FAST = "fast"      # gpt-4o-mini: 便宜快
    BALANCED = "balanced"  # gpt-4o: 均衡

COST = {
    Tier.FAST:    {"model":"gpt-4o-mini","prompt":0.15,"completion":0.60},
    Tier.BALANCED:{"model":"gpt-4o",     "prompt":2.50,"completion":10.00},
}

@dataclass
class Budget:
    max_task: float = 5.0
    max_day: float = 100.0
    warn_at: float = 0.8

class CostTracker:
    def __init__(self, cfg: Budget):
        self.cfg = cfg
        self.task_costs = {}; self.day_costs = {}
        self.today = time.strftime("%Y-%m-%d")
    
    def check(self, task_id: str, est: float) -> bool:
        today = time.strftime("%Y-%m-%d")
        if today != self.today:
            self.today = today; self.day_costs.clear()
        if self.task_costs.get(task_id,0)+est > self.cfg.max_task: return False
        if sum(self.day_costs.values())+est > self.cfg.max_day: return False
        return True
    
    def record(self, task_id: str, cost: float):
        self.task_costs[task_id] = self.task_costs.get(task_id,0)+cost
        self.day_costs[self.today] = self.day_costs.get(self.today,0)+cost
        ratio = sum(self.day_costs.values())/self.cfg.max_day
        if ratio > self.cfg.warn_at:
            print(f"警告：今日成本已达 {ratio:.0%}")

class AdaptiveRouter:
    def __init__(self, tracker: CostTracker):
        self.tracker = tracker
    
    def route(self, task: str, ctx_size=0) -> Tier:
        if not self.tracker.check("default", 0.1): return Tier.FAST
        is_simple = len(task)<200 and "修复" in task and ctx_size<10000
        is_complex = any(k in task for k in ["架构","设计","重构"]) or ctx_size>50000
        return Tier.FAST if is_simple else Tier.BALANCED

6.2 熔断器实现

import time, random

class CircuitBreaker:
    def __init__(self, threshold=5, recovery=60):
        self.threshold = threshold
        self.recovery = recovery
        self.failures = 0; self.last_fail = 0; self.state = "closed"
    
    def can_execute(self) -> bool:
        if self.state == "closed": return True
        if self.state == "open":
            if time.time()-self.last_fail > self.recovery:
                self.state = "half_open"; return True
            return False
        return True
    
    def record_success(self):
        self.failures = 0; self.state = "closed"
    
    def record_failure(self):
        self.failures += 1; self.last_fail = time.time()
        if self.failures >= self.threshold:
            self.state = "open"
            print(f"Circuit breaker OPEN after {self.failures} failures")

class RetryManager:
    def __init__(self, max_attempts=3, base_delay=1.0, max_delay=30.0):
        self.max_attempts = max_attempts
        self.base_delay = base_delay; self.max_delay = max_delay
    
    def execute(self, func, *args, **kwargs):
        last_err = None
        for attempt in range(self.max_attempts):
            try: return func(*args, **kwargs)
            except Exception as e:
                last_err = e
                if attempt < self.max_attempts - 1:
                    delay = min(self.base_delay*(2**attempt)*(0.5+random.random()), self.max_delay)
                    if "429" in str(e): delay = max(delay, 60)
                    print(f"Attempt {attempt+1} failed, retry in {delay:.1f}s")
                    time.sleep(delay)
        return {"error": str(last_err), "degraded": True}

6.3 完整生产级封装

class ProductionAgent:
    """整合所有架构模式的生产级 AI Coding Agent"""
    def __init__(self, config: dict):
        self.llm_factory = lambda tier: OpenAIClient(
            config["api_key"], COST[tier]["model"], config.get("base_url"))
        self.tracker = CostTracker(Budget(
            max_task=config.get("max_task_cost",5.0),
            max_day=config.get("max_daily_cost",100.0)))
        self.router = AdaptiveRouter(self.tracker)
        self.ctx_window = ContextWindow()
        self.tool_cache = ToolResultCache()
        self.sem_index = SemanticIndex()
        root = config.get("project_root",".")
        self.fs_tools = [ReadFileTool(root), WriteFileTool(root), BashTool()]
    
    def run(self, task: str, mode="single") -> dict:
        task_id = str(uuid.uuid4())[:8]
        tier = self.router.route(task)
        llm = self.llm_factory(tier)
        
        if mode == "single":
            agent = SingleAgent(llm, self.fs_tools, AgentConfig(max_iterations=30))
            result = agent.run(task)
            self.tracker.record(task_id, result["cost_usd"])
            return {"task_id":task_id,"mode":"single",**result}
        
        elif mode == "harness":
            pipeline = HarnessPipeline(lambda: self.llm_factory(tier), self.fs_tools)
            result = pipeline.run(task, [])
            self.tracker.record(task_id, result.total_cost_usd)
            return {"task_id":task_id,"mode":"harness","success":result.success,
                    "iterations":result.total_iterations,"cost":result.total_cost_usd,
                    "quality":result.quality_score,"errors":result.errors}

# 使用：
# agent = ProductionAgent({"api_key":os.environ["OPENAI_API_KEY"],
#                          "project_root":"/project","max_task_cost":10.0,"max_daily_cost":200.0})
# r = agent.run("实现用户认证模块","harness")

七、实战对比与选型建议

八、2026 年工程实践建议

1. 不要让 Agent 裸跑

没有质量门的 Agent 在生产环境中迟早出事。至少要有一个 Reviewer 角色帮你把关。

2. 上下文管理是生死线

超过上下文限制不是"慢"，是"彻底失败"。从第一天就把上下文压缩和记忆系统做进去。

3. 成本控制要主动

从第一天就设定预算上限，用 FAST 模型处理简单任务。等月底账单来了才心疼就晚了。

4. MCP 是正确方向

专用工具适配器的维护成本会随着系统增长而爆炸。MCP 协议虽然还年轻，但它解决了正确的问题。

5. 可观测性是必需的

记录每次工具调用的输入输出、执行时间、LLM 响应时间。这些数据是你优化 Agent 行为的基础。

6. 渐进式引入 AI

不要试图一步到位用 AI 重写整个开发流程。从代码审查、简单 bug 修复这类低风险任务开始，积累信心后再扩展。

结语

AI Coding Agent 工程化的核心挑战，不是让 AI 写出能跑的代码，而是构建一套能让 AI 持续、稳定、可控地产出高质量代码的系统工程。

从单 Agent 到 Harness 多 Agent 编排，从上下文压缩到 MCP 协议标准化，从成本控制到可靠性保障——每一步都是在为不确定的 AI 输出套上确定性的工程约束。

这条路没有银弹。但只要你开始认真对待 AI 编程的工程化问题，这篇文章里的五种架构模式和配套代码，应该能让你少走很多弯路。

2026 年，AI 编程的竞争已经不在"能不能写出代码"这个层面了——竞争在"谁能构建更可靠的 AI 编程系统"这个层面。工程化能力，将成为区分业余玩家和专业团队的关键分水岭。

动手吧。

本文代码基于 2026 年主流工程实践编写，可在生产环境中参考使用。如有问题，欢迎在评论区讨论。