Agent 怎样规划和停止：ReAct、任务拆解与执行预算

import json
import time


def signature(action: dict) -> str:
    return json.dumps(action, ensure_ascii=False, sort_keys=True)


def run_agent(goal: str, policy, execute, max_steps: int = 5, timeout: float = 2.0):
    started = time.monotonic()
    state = {"goal": goal, "trajectory": [], "observations": []}
    previous_signature, repeats = None, 0

    for step in range(1, max_steps + 1):
        if time.monotonic() - started >= timeout:
            return {**state, "status": "time_budget_exceeded"}

        action = policy(state)
        if action["name"] == "finish":
            return {**state, "status": "completed", "answer": action["answer"]}

        current = signature(action)
        repeats = repeats + 1 if current == previous_signature else 0
        if repeats >= 2:
            return {**state, "status": "repeated_action"}
        previous_signature = current

        observation = execute(action)
        state["trajectory"].append({"step": step, "action": action, "observation": observation})
        state["observations"].append(observation)

    return {**state, "status": "step_budget_exceeded"}


def demo_policy(state: dict) -> dict:
    if not state["observations"]:
        return {"name": "get_status", "device_id": "AX-3"}
    return {"name": "finish", "answer": "AX-3 当前离线，需要人工检查网络。"}


def execute(action: dict) -> dict:
    if action == {"name": "get_status", "device_id": "AX-3"}:
        return {"online": False}
    return {"error": "invalid_action"}


print(run_agent("判断 AX-3 是否需要处理", demo_policy, execute))