Loop Engineering：把 AI Agent 从对话助手变成可运行的工程系统

目标 + 当前状态
       ↓
   模型决策
       ↓
工具调用 → 结构化结果 → 状态更新
       ↑                    ↓
       └──── 未完成则继续 ──┘

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path
from typing import Any

WORKSPACE = Path.cwd().resolve()
MAX_STEPS = 6
LOG_FILE = WORKSPACE / "agent-events.jsonl"


def check_file(path: str) -> dict[str, Any]:
    target = (WORKSPACE / path).resolve()
    if WORKSPACE not in target.parents and target != WORKSPACE:
        return {"ok": False, "error": "path escapes workspace"}
    if not target.is_file():
        return {"ok": False, "error": f"missing file: {path}"}
    return {"ok": True, "path": path, "size": target.stat().st_size}


def run_tests() -> dict[str, Any]:
    command = [sys.executable, "-m", "unittest", "discover"]
    try:
        result = subprocess.run(
            command,
            cwd=WORKSPACE,
            capture_output=True,
            text=True,
            timeout=30,
        )
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": "tests timed out"}

    output = (result.stdout + result.stderr)[-4000:]
    return {
        "ok": result.returncode == 0,
        "returncode": result.returncode,
        "output": output,
    }


TOOLS = {
    "check_file": lambda args: check_file(args["path"]),
    "run_tests": lambda args: run_tests(),
}


def decide(history: list[dict[str, Any]]) -> dict[str, Any]:
    """确定性模型替身；生产环境可替换为返回 JSON 的 LLM 调用。"""
    if not history:
        return {
            "action": "tool",
            "tool": "check_file",
            "args": {"path": "README.md"},
            "reason": "确认项目说明文件存在",
        }
    if len(history) == 1:
        return {
            "action": "tool",
            "tool": "run_tests",
            "args": {},
            "reason": "收集可验证的测试结果",
        }

    tests = history[-1]["result"]
    return {
        "action": "finish",
        "status": "completed" if tests.get("ok") else "needs_human",
        "summary": "测试通过" if tests.get("ok") else "测试失败，需要人工检查",
    }


def append_event(event: dict[str, Any]) -> None:
    with LOG_FILE.open("a", encoding="utf-8") as file:
        file.write(json.dumps(event, ensure_ascii=False) + "\n")


def main() -> None:
    history: list[dict[str, Any]] = []

    for step in range(1, MAX_STEPS + 1):
        decision = decide(history)
        append_event({"step": step, "type": "decision", "data": decision})

        if decision.get("action") == "finish":
            print(json.dumps(decision, ensure_ascii=False, indent=2))
            return

        tool_name = decision.get("tool")
        if tool_name not in TOOLS:
            raise RuntimeError(f"tool is not allowed: {tool_name}")

        result = TOOLS[tool_name](decision.get("args", {}))
        observation = {"decision": decision, "result": result}
        history.append(observation)
        append_event({"step": step, "type": "observation", "data": result})

    print(json.dumps({
        "action": "finish",
        "status": "budget_exhausted",
        "summary": f"超过最大步数 {MAX_STEPS}",
    }, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

你是项目检查 Agent。根据目标、历史动作和工具结果选择下一步。

只允许输出 JSON，不要输出解释文字。
可用动作：
1. {"action":"tool","tool":"check_file","args":{"path":"..."},"reason":"..."}
2. {"action":"tool","tool":"run_tests","args":{},"reason":"..."}
3. {"action":"finish","status":"completed|needs_human","summary":"..."}

规则：
- 不得虚构工具执行结果。
- 完成状态必须有测试结果作为证据。
- 同一个失败动作最多重试一次。
- 信息不足或风险不可判断时返回 needs_human。