AI 编程的工程化落地：从补全到可验证的智能体

# rag/context_builder.py
import ast, re
from dataclasses import dataclass

@dataclass
class Chunk:
    path: str
    symbol: str
    code: str
    score: float = 0.0

def index_repo(root):
    chunks = []
    for p in Path(root).rglob("*.py"):
        tree = ast.parse(p.read_text(encoding="utf-8"))
        lines = p.read_text().splitlines()
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
                code = "\n".join(lines[node.lineno - 1: node.end_lineno])
                chunks.append(Chunk(str(p), node.name, code))
    return chunks

def build_context(chunks, query, k=8):
    q = set(re.findall(r"\w+", query.lower()))
    for c in chunks:
        # 符号名精确命中权重最高，其次词重叠
        name_hit = 1.0 if c.symbol.lower() in query.lower() else 0.0
        overlap = len(q & set(re.findall(r"\w+", c.code.lower()))) / (len(q) or 1)
        c.score = name_hit * 2 + overlap
    top = sorted(chunks, key=lambda c: c.score, reverse=True)[:k]
    # 按字符预算裁剪，防止撑爆上下文窗口
    budget, out = 6000, []
    for c in top:
        if budget - len(c.code) < 0: break
        out.append(f"# {c.path}::{c.symbol}\n{c.code}")
        budget -= len(c.code)
    return "\n\n".join(out)

# 用 grammar 约束输出为严格的补丁对象
GRAMMAR = r'''
root   ::= object
object ::= "{" pair ("," pair)* "}"
pair   ::= string ":" value
value  ::= string | number | object | array | "true" | "false" | "null"
string ::= "\"" [^"]* "\""
number ::= "-"? [0-9]+ ("." [0-9]+)?
array  ::= "[" (value ("," value)*)? "]"
'''

def request_patch(llm, prompt):
    raw = llm.generate(
        prompt,
        grammar=GRAMMAR,
        max_tokens=2048,
        temperature=0.1,          # 代码任务压低随机性
    )
    return json.loads(raw)        # 语法保证下，解析几乎不会失败

# agent/verify_loop.py
def run_patch(patch, repo):
    for edit in patch["edits"]:
        path = repo / edit["file"]
        src = path.read_text(encoding="utf-8")
        if edit["old"] not in src:
            return False, f"anchor not found in {edit['file']}"
        path.write_text(src.replace(edit["old"], edit["new"], 1), encoding="utf-8")
    return True, ""

def verify_loop(llm, task, repo, max_rounds=3):
    context = build_context(index_repo(repo), task)
    history = []

    for r in range(max_rounds):
        prompt = f"任务:{task}\n上下文:\n{context}\n历史:\n{history}\n输出补丁 JSON。"
        patch = request_patch(llm, prompt)

        ok, err = run_patch(patch, repo)
        if not ok:
            history.append(f"应用失败:{err}")
            continue

        test = subprocess.run(["pytest", "-q"], cwd=repo,
                              capture_output=True, text=True)
        if test.returncode == 0:
            return {"status": "pass", "rounds": r + 1, "patch": patch}

        history.append("测试失败:\n" + test.stdout[-2000:])  # 只回传尾部
        git_checkout(repo)                                     # 失败即回滚

    return {"status": "fail", "history": history}

# eval/bench.py
def evaluate(agent, cases):
    res = {"pass@1": 0, "avg_rounds": 0, "regression": 0}
    for c in cases:
        repo = clone(c["repo"], c["commit"])
        out = agent.verify_loop(c["task"], repo)
        res["pass@1"] += out["status"] == "pass"
        res["avg_rounds"] += out.get("rounds", 0)
        # 回归：原有测试是否被破坏
        res["regression"] += subprocess.run(
            ["pytest", c["legacy"]], cwd=repo).returncode != 0
    n = len(cases)
    return {k: v / n for k, v in res.items()}