Agent Loop 与终止条件

import hashlib, json, time
from dataclasses import dataclass, field
def sig(tool: str, args: dict) -> str:
    """工具调用签名：参数做稳定序列化，避免键序影响"""
    return f"{tool}::{json.dumps(args, sort_keys=True, ensure_ascii=False)}"
@dataclass
class LoopGuard:
    max_turns: int = 15
    max_seconds: float = 300
    stall_limit: int = 3           # 连续多少轮无进展就退出
    repeat_limit: int = 3          # 同一调用签名最多重复几次
    started: float = field(default_factory=time.time)
    turns: int = 0
    seen: dict = field(default_factory=dict)     # 签名 -> 次数
    last_obs: str = ""
    stall: int = 0
    def check(self) -> tuple[bool, str]:
        """返回 (是否应继续, 退出原因)"""
        if self.turns >= self.max_turns:
            return False, "BUDGET_TURNS"
        if time.time() - self.started > self.max_seconds:
            return False, "BUDGET_TIME"
        if self.stall >= self.stall_limit:
            return False, "NO_PROGRESS"
        return True, ""
    def note_call(self, tool: str, args: dict) -> tuple[bool, str]:
        """记录一次调用；返回是否允许执行"""
        s = sig(tool, args)
        self.seen[s] = self.seen.get(s, 0) + 1
        if self.seen[s] > self.repeat_limit:
            return False, f"同一调用已重复 {self.seen[s]} 次，请换策略或结束任务"
        return True, ""
    def note_observation(self, content: str) -> None:
        """观察结果没变化 → 记为一次停滞"""
        h = hashlib.sha256(content.encode()).hexdigest()
        if h == self.last_obs:
            self.stall += 1
        else:
            self.stall = 0
            self.last_obs = h
def run(task: str, call_model, exec_tool, ask_human):
    guard = LoopGuard()
    ctx = [{"role": "user", "content": task}]
    while True:
        ok, why = guard.check()
        if not ok:
            # 关键：退出时带上已完成的部分，而不是一句失败
            return {"status": why, "turns": guard.turns,
                    "partial": summarize(ctx), "context": ctx}
        guard.turns += 1
        plan = call_model(ctx, temperature=0)
        # ① 显式完成信号（不要靠「没调工具」隐式判断）
        if plan["action"] == "finish":
            return {"status": "DONE", "result": plan.get("answer"), "turns": guard.turns}
        # ④ 需要人工：暂停而不是硬闯
        if plan["action"] == "need_human":
            return {"status": "NEED_HUMAN", "question": plan.get("question", ""),
                    "context": ctx, "resumable": True}
        # ⑤ 输出无法解析：有限次重试后才放弃
        if plan["action"] == "invalid":
            ctx.append({"role": "user",
                        "content": "上一次输出无法解析为合法动作，请只输出 JSON。"})
            continue
        allowed, msg = guard.note_call(plan["tool"], plan.get("args", {}))
        if not allowed:
            ctx.append({"role": "user", "content": msg})
            continue
        result = exec_tool(plan["tool"], plan.get("args", {}))
        guard.note_observation(str(result))
        # 副作用操作必须确认（人工介入点）
        if result.get("side_effect") and not ask_human(plan):
            ctx.append({"role": "user", "content": "用户拒绝了该操作。"})
            continue
        ctx.append({"role": "tool", "content": str(result)[:2000]})