Agent 循环机制设计：从 ReAct 到 Plan-and-Execute 的演进

Thought: 用户想了解今天的天气，我需要搜索
Action: search_weather("北京 今天")
Observation: 北京今天晴，最高温32°C
Thought: 搜索结果已获得，可以回答用户
Answer: 北京今天晴天，最高温度32°C。

def react_loop(query, max_iterations=5):
    messages = [{"role": "user", "content": query}]
    
    for i in range(max_iterations):
        response = llm.invoke(messages + [
            {"role": "system", "content": REACT_PROMPT}
        ])
        
        if "Final Answer:" in response:
            return parse_answer(response)
        
        action = parse_action(response)
        observation = execute_tool(action)
        messages.append({"role": "assistant", "content": response})
        messages.append({"role": "user", "content": f"Observation: {observation}"})
    
    return "达到最大迭代次数"

def plan(query):
    prompt = f"""将以下任务分解为具体步骤：
    任务：{query}
    
    输出JSON格式的步骤列表。"""
    
    plan = llm.invoke(prompt)
    return json.loads(plan)

# 输出示例
{
  "steps": [
    {"id": 1, "action": "search", "params": {"query": "2026年AI Agent市场规模"}},
    {"id": 2, "action": "search", "params": {"query": "主要Agent框架市场份额"}},
    {"id": 3, "action": "analyze", "params": {"source": "step1+step2"}},
    {"id": 4, "action": "write_report", "params": {"source": "step3"}}
  ]
}

def execute_plan(steps):
    results = {}
    for step in steps:
        result = execute_step(step, results)
        results[step["id"]] = result
        
        # 动态重规划
        if step.get("replan"):
            new_steps = replan(query, results)
            steps = merge_plan(steps, new_steps)
    
    return results

class AdaptiveAgent:
    def __init__(self):
        self.strategies = {
            "simple": ReActStrategy(),
            "complex": PlanExecuteStrategy(),
            "research": TreeOfThoughtStrategy(),
        }
    
    def run(self, query):
        complexity = self.assess_complexity(query)
        strategy = self.select_strategy(complexity)
        
        result = strategy.execute(query)
        
        # 如果失败，升级策略
        if not result.success and strategy != "research":
            return self.run_with_higher_strategy(query)
        
        return result
    
    def assess_complexity(self, query):
        factors = {
            "steps_required": self.estimate_steps(query),
            "tools_needed": self.identify_tools(query),
            "domain_breadth": self.assess_breadth(query),
        }
        score = sum(factors.values())
        if score < 5: return "simple"
        if score < 15: return "complex"
        return "research"

class CheckpointedAgent:
    def __init__(self, checkpoint_store):
        self.store = checkpoint_store
    
    def run(self, task_id, query):
        # 从检查点恢复
        state = self.store.load(task_id) or {"step": 0, "results": {}}
        
        while state["step"] < len(state["plan"]):
            step = state["plan"][state["step"]]
            
            try:
                result = self.execute_step(step)
                state["results"][step["id"]] = result
                state["step"] += 1
                
                # 保存检查点
                self.store.save(task_id, state)
                
            except Exception as e:
                state["error"] = str(e)
                self.store.save(task_id, state)
                raise
        
        return state["results"]