Codex AI 工程实践：从上下文打包到代理循环的工程化实现

from pathlib import Path
from dataclasses import dataclass

IGNORE_DIRS = {".git", "node_modules", "__pycache__", ".venv", "dist", "build"}
ALLOW_EXT = {".py", ".ts", ".tsx", ".go", ".java", ".md", ".toml", ".yaml", ".yml"}

@dataclass
class FileMeta:
    path: str
    size: int
    tokens: int

def scan_repo(root: Path, max_files: int = 2000) -> list[FileMeta]:
    files: list[FileMeta] = []
    for p in root.rglob("*"):
        if not p.is_file() or p.suffix not in ALLOW_EXT:
            continue
        if any(part in IGNORE_DIRS for part in p.parts):
            continue
        text = p.read_text(encoding="utf-8", errors="ignore")
        tokens = len(text) // 4  # 粗略估算
        files.append(FileMeta(str(p.relative_to(root)), p.stat().st_size, tokens))
        if len(files) >= max_files:
            break
    return files

def select_context(files: list[FileMeta], keywords: list[str], budget: int = 60000) -> list[str]:
    scored = []
    for f in files:
        score = sum(1 for k in keywords if k.lower() in f.path.lower())
        scored.append((score, f))
    scored.sort(key=lambda x: (-x[0], x[1].tokens))
    chosen, used = [], 0
    for _, f in scored:
        if used + f.tokens > budget:
            continue
        chosen.append(f.path)
        used += f.tokens
    return chosen

import subprocess, json
from dataclasses import dataclass, field

@dataclass
class Step:
    goal: str
    files: list[str] = field(default_factory=list)
    status: str = "pending"
    result: str = ""

def run_cmd(cmd: list[str], timeout: int = 60) -> dict:
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        return {"code": p.returncode, "out": p.stdout[-4000:], "err": p.stderr[-2000:]}
    except subprocess.TimeoutExpired:
        return {"code": -1, "out": "", "err": "timeout"}

def verify(repo: Path) -> dict:
    checks = {
        "lint": run_cmd(["ruff", "check", "."]),
        "type": run_cmd(["mypy", "."]),
        "test": run_cmd(["pytest", "-q"]),
    }
    checks["ok"] = all(c["code"] == 0 for c in checks.values())
    return checks

def agent_loop(plan: list[Step], repo: Path, apply_fn, max_rounds: int = 5):
    for rnd in range(max_rounds):
        for step in plan:
            if step.status == "done":
                continue
            patch = apply_fn(step)          # 调用模型生成补丁
            step.result = patch
            check = verify(repo)
            if check["ok"]:
                step.status = "done"
            else:
                step.status = "retry"
                step.goal += f"\n上次失败：{check['test']['err'][:500]}"
        if all(s.status == "done" for s in plan):
            return plan
    return plan

import difflib
from pathlib import Path

def make_diff(old: str, new: str, path: str) -> str:
    return "".join(difflib.unified_diff(
        old.splitlines(keepends=True),
        new.splitlines(keepends=True),
        fromfile=f"a/{path}",
        tofile=f"b/{path}",
    ))

def summarize(diff: str) -> dict:
    added = sum(1 for l in diff.splitlines() if l.startswith("+") and not l.startswith("+++"))
    removed = sum(1 for l in diff.splitlines() if l.startswith("-") and not l.startswith("---"))
    return {"added": added, "removed": removed, "files": diff.count("--- ")}

name: ci
on: [push, pull_request]
jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install ruff mypy pytest pytest-cov bandit pip-audit
      - run: ruff check .
      - run: mypy .
      - run: pytest --cov=app --cov-fail-under=80
      - run: bandit -r app
      - run: pip-audit