05 | Loop Engineering——Agent 的核心循环

[循环开始]

第 1 轮：
  思考：用户想知道北京明天的天气，我需要用天气工具查询
  行动：调用 weather_api("北京", "明天")
  观察：返回结果 {"temp_high": 12, "temp_low": 3, "weather": "多云"}

第 2 轮：
  思考：最高温 12°C，低于 15°C，需要提醒用户带外套
  行动：生成回复
  观察：回复已生成，任务完成

[循环结束]

Thought: 我需要搜索 Python 3.12 的新特性
Action: search
Action Input: "Python 3.12 new features"
Observation: [搜索结果返回了 5 条相关信息...]

Thought: 搜索结果显示 Python 3.12 主要有 3 个重要变化...
         但我还需要确认性能提升的具体数据
Action: search
Action Input: "Python 3.12 performance benchmarks"
Observation: [性能测试结果...]

Thought: 现在我有了足够的信息来回答用户的问题
Action: final_answer
Action Input: "Python 3.12 的主要新特性包括..."

[任务] 编写一个 Python 函数，计算列表中所有素数的和

第 1 轮：
  Thought: 我需要写一个判断素数的函数，然后对列表中的素数求和
  Action: code_exec
  Code:
    def sum_primes(lst):
        return sum(x for x in lst if x > 1 and all(x % i for i in range(2, x)))
  Observation: 测试失败——输入 [2,3,4,1,0,-1] 时返回了错误结果（0 和负数应该被排除）

  [Reflexion 步骤]
  反思: 我的素数判断函数对 0、1、负数处理不正确。
       x > 1 这个条件应该已经排除了 0 和 1，
       但 range(2, x) 在 x=2 时返回空列表，all() 对空列表返回 True，
       所以 2 会被正确判断为素数...
       问题可能在 -1：range(2, -1) 也是空列表，all() 返回 True，-1 被误判为素数！
  教训: 需要明确排除负数和 0、1

第 2 轮：
  Thought: 根据上次的反思，我需要先检查 x >= 2 再判断素数
  Action: code_exec
  Code:
    def sum_primes(lst):
        def is_prime(n):
            if n < 2: return False
            return all(n % i for i in range(2, int(n**0.5) + 1))
        return sum(x for x in lst if is_prime(x))
  Observation: 所有测试通过 ✓

# 停止条件 1: Agent 明确给出了最终答案
if action == "final_answer":
    return result

# 停止条件 2: 达到最大循环次数
if loop_count >= max_iterations:
    return "任务未完成：超过最大循环次数限制"

# 停止条件 3: 超时
if elapsed_time >= max_time:
    return "任务未完成：执行超时"

# 停止条件 4: 重复检测（连续 N 轮没有进展）
if is_stuck_in_loop(recent_observations, threshold=3):
    return "任务未完成：Agent 陷入循环"

try:
    result = execute_tool(action, action_input)
except ToolTimeoutError:
    result = "工具执行超时，请尝试其他方法或简化输入"
except ToolPermissionError:
    result = "没有权限执行此操作，请换一种方式"
except Exception as e:
    result = f"工具执行失败: {str(e)}。请分析原因并重试"

# 方法 1: 记录历史行动，检测重复
action_history = []

def detect_repetition(new_action):
    if new_action in action_history[-3:]:
        return "你已经在重复同一个操作。请换一种方法，或者承认当前无法完成。"
    return None

# 方法 2: 限制同一工具的使用次数
tool_use_count = defaultdict(int)
MAX_TOOL_USE = 5

def check_tool_limit(tool_name):
    if tool_use_count[tool_name] >= MAX_TOOL_USE:
        return f"{tool_name} 已使用 {MAX_TOOL_USE} 次，请换其他工具"
    return None

=== Loop #1 ===
[Thought] 用户想要查天气，我需要调用天气 API
[Action] weather_api(city="北京", date="tomorrow")
[Observation] {"temp": 12, "condition": "多云"}
[Latency] 1.2s | [Tokens] 150 in / 80 out

=== Loop #2 ===
[Thought] 温度 12°C < 15°C，需要提醒用户
[Action] final_answer("明天北京多云，最高温 12°C，建议带外套")
[Observation] 任务完成
[Latency] 0.8s | [Tokens] 200 in / 50 out

=== Summary ===
Total loops: 2 | Total time: 2.0s | Total tokens: 480

def run_agent(task, tools, max_iterations=15, timeout=120):
    """一个生产级的 Agent 执行循环"""
    history = []
    start_time = time.time()

    for i in range(max_iterations):
        # 超时检查
        if time.time() - start_time > timeout:
            return {"status": "timeout", "history": history}

        # 1. 推理：让 LLM 根据历史和任务决定下一步
        thought, action, action_input = llm_reason(task, history)

        # 2. 检查是否完成
        if action == "final_answer":
            return {"status": "success", "answer": action_input, "history": history}

        # 3. 检查重复
        if detect_repetition(action, action_input, history):
            history.append({"role": "system", "content": "检测到重复操作，请换方法"})
            continue

        # 4. 执行行动
        try:
            observation = execute_tool(action, action_input, tools)
        except Exception as e:
            observation = f"执行失败: {e}，请分析原因"

        # 5. 记录到历史
        history.append({
            "thought": thought,
            "action": action,
            "action_input": action_input,
            "observation": observation
        })

    return {"status": "max_iterations", "history": history}