从零开发AI编程智能体：主循环、上下文压缩与Hook设计

Harness = Tools + Knowledge + Observation + Action + Permissions

    Tools:         文件读写、Shell、网络、数据库、浏览器
    Knowledge:     产品文档、API 规范、风格指南
    Observation:   git diff、错误日志、浏览器状态
    Action:        CLI 命令、API 调用、UI 交互
    Permissions:   沙箱隔离、审批流程、信任边界

def agent_loop(messages: list):
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return

        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = TOOL_HANDLERS[block.name](**block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

messages[]
    ↓
L3 tool_result_budget  （大结果写磁盘，0次API调用）
    ↓
L1 snip_compact        （消息数>50时截断中间，0次API调用）
    ↓
L2 micro_compact       （旧工具结果替换为占位符，0次API调用）
    ↓
[token > 50000 阈值？]
    ├─ No  → 直接调LLM
    └─ Yes → L4 auto compact（LLM摘要，1次API调用）
                    ↓
              [API 返回 prompt_too_long?]
                    └─ Yes → reactive_compact（应急，1次API调用）

def snip_compact(messages, max_messages=50):
    if len(messages) <= max_messages:
        return messages
    keep_head, keep_tail = 3, max_messages - 3
    snipped = len(messages) - keep_head - keep_tail
    return (
        messages[:keep_head] +
        [{"role": "user", "content": f"[snipped {snipped} messages]"}] +
        messages[-keep_tail:]
    )

def micro_compact(messages):
    tool_results = collect_tool_results(messages)
    if len(tool_results) <= KEEP_RECENT:
        return messages
    for _, _, block in tool_results[:-KEEP_RECENT]:
        if len(block.get("content", "")) > 120:
            block["content"] = "[Earlier tool result compacted. Re-run if needed.]"
    return messages

def persist_large_output(tool_use_id, output):
    if len(output) <= PERSIST_THRESHOLD:  # 30000
        return output
    path = TOOL_RESULTS_DIR / f"{tool_use_id}.txt"
    path.write_text(output)
    return (
        f"<persisted-output>\n"
        f"Full output: {path}\n"
        f"Preview:\n{output[:2000]}\n"
        f"</persisted-output>"
    )

def compact_history(messages):
    transcript_path = write_transcript(messages)
    summary = summarize_history(messages)
    return [{"role": "user", "content": f"[Compacted]\n\n{summary}"}]

def reactive_compact(messages):
    transcript = write_transcript(messages)
    summary = summarize_history(messages)
    return [
        {"role": "user", "content": f"[Reactive compact]\n\n{summary}"},
        *messages[-5:],
    ]

HOOKS = {
    "UserPromptSubmit": [],  # 用户输入前触发
    "PreToolUse": [],        # 工具执行前触发
    "PostToolUse": [],       # 工具执行后触发
    "Stop": [],              # Agent 停止时触发
}

def register_hook(event: str, callback):
    HOOKS[event].append(callback)

def trigger_hooks(event: str, *args):
    for callback in HOOKS[event]:
        result = callback(*args)
        if result is not None:  # 非None = 阻断
            return result
    return None

def permission_hook(block):
    if block.name == "bash":
        for pattern in DENY_LIST:
            if pattern in block.input.get("command", ""):
                return "Permission denied by deny list"
        for kw in DESTRUCTIVE:
            if kw in block.input.get("command", ""):
                choice = input("Allow? [y/N] ")
                if choice not in ("y", "yes"):
                    return "Permission denied by user"

# s03: 硬编码
if not check_permission(block):
    results.append({"error": "blocked"})
    continue

# s04: Hook
blocked = trigger_hooks("PreToolUse", block)
if blocked:
    results.append({"type": "tool_result", "content": str(blocked)})
    continue