跳到主要内容

规划与任务分解

规划（Planning）是 AI Agent 的核心能力之一。它让 Agent 能够将复杂的目标分解为一系列可执行的步骤，并按合理的顺序完成这些步骤。没有规划能力的 Agent 只能处理简单的单步任务，面对复杂需求时会陷入混乱。

为什么需要规划​

当用户给 Agent 一个复杂的请求时，如果 Agent 直接开始执行，往往会遇到以下问题：

  • 遗漏任务：复杂请求可能包含多个子任务，Agent 可能遗漏其中的某些部分
  • 顺序混乱：某些步骤有依赖关系，必须按特定顺序执行
  • 资源浪费：没有合理规划可能导致重复工作或无效操作
  • 难以追踪：缺少清晰的计划使得任务进度难以监控

举个例子，假设用户请求："帮我准备下周五的客户演示，我需要研究客户行业、制作幻灯片、练习演讲，还要给团队发会议邀请。"

这个请求实际上包含了四个子任务。如果 Agent 尝试一次性处理，可能会遗漏某些步骤，或者以错误的顺序执行（比如在制作幻灯片之前就练习演讲）。

规划模块的核心作用就是将这种模糊、复杂的目标转化为清晰、有序的行动计划。

规划的核心组件​

根据 Lilian Weng 的经典框架，AI Agent 的规划模块包含四个子模块：

组件作用说明
子目标分解将大目标拆分为小目标降低任务复杂度，便于逐步执行
思维链逐步思考推理过程帮助模型理清思路，提高推理质量
反思回顾和评估执行结果发现问题，改进策略
自我批评检验和修正决策提高决策的可靠性

这四个组件协同工作，让 Agent 能够系统地处理复杂任务。

任务分解策略​

顺序任务分解​

顺序任务是指各步骤之间存在依赖关系，必须按特定顺序执行。前一个步骤的输出可能是后一个步骤的输入。

这种分解方式的关键是识别步骤之间的依赖关系。例如，在制作幻灯片之前必须先完成研究，因为研究结果是幻灯片内容的基础。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

class SubTask(BaseModel):
    """子任务结构"""
    id: int = Field(description="任务编号")
    description: str = Field(description="任务描述")
    depends_on: List[int] = Field(default=[], description="依赖的任务编号")

class TaskPlan(BaseModel):
    """任务计划"""
    goal: str = Field(description="原始目标")
    subtasks: List[SubTask] = Field(description="子任务列表")

def decompose_sequential_task(goal: str) -> TaskPlan:
    """分解顺序任务"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个任务规划专家。请将用户的复杂目标分解为有序的子任务。

要求：
1. 每个子任务必须具体、可执行
2. 明确标注任务之间的依赖关系
3. 按合理的执行顺序排列

返回 JSON 格式的任务计划。"""),
        ("user", "{goal}")
    ])
    
    chain = prompt | llm.with_structured_output(TaskPlan)
    return chain.invoke({"goal": goal})

# 示例
goal = "帮我准备下周五的客户演示，包括研究客户行业、制作幻灯片、练习演讲和发会议邀请"
plan = decompose_sequential_task(goal)

for task in plan.subtasks:
    deps = f" (依赖: {task.depends_on})" if task.depends_on else ""
    print(f"{task.id}. {task.description}{deps}")

输出示例：

1. 研究客户行业，了解其业务模式、竞争格局和最新动态
2. 根据研究结果，制作演示幻灯片 (依赖: [1])
3. 练习演讲，确保流畅表达 (依赖: [2])
4. 给团队发送会议邀请 (依赖: [2])

并行任务分解​

并行任务是指各步骤之间没有依赖关系，可以同时执行或以任意顺序执行。并行分解可以提高效率，充分利用资源。

from typing import List, Literal
from pydantic import BaseModel, Field

class ParallelSubTask(BaseModel):
    """并行子任务"""
    id: int
    description: str
    can_parallelize: bool = Field(default=True, description="是否可并行执行")
    group: str = Field(description="任务分组，同组可并行")

class ParallelPlan(BaseModel):
    """并行任务计划"""
    goal: str
    phases: List[List[ParallelSubTask]] = Field(description="分阶段的任务列表")

def decompose_parallel_task(goal: str) -> ParallelPlan:
    """分解并行任务"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个任务规划专家。请分析用户目标并识别可以并行执行的任务。

要求：
1. 识别哪些任务可以同时执行
2. 将任务分为不同的执行阶段
3. 同一阶段的任务没有相互依赖

返回分阶段的任务计划。"""),
        ("user", "{goal}")
    ])
    
    chain = prompt | llm.with_structured_output(ParallelPlan)
    return chain.invoke({"goal": goal})

# 示例
goal = "帮我整理一下：检查邮件、查看日程、获取今日天气，然后汇总报告"
plan = decompose_parallel_task(goal)

for i, phase in enumerate(plan.phases, 1):
    print(f"\n阶段 {i}:")
    for task in phase:
        print(f"  - {task.description}")

输出示例：

阶段 1:
  - 检查邮件收件箱
  - 查看今日日程安排
  - 获取今日天气信息

阶段 2:
  - 汇总以上信息生成报告

嵌套任务分解​

有些任务需要多层分解。一个大任务分解为子任务后，某个子任务本身可能仍然很复杂，需要进一步分解。这形成了任务的层次结构。

from typing import List, Optional
from pydantic import BaseModel, Field

class NestedTask(BaseModel):
    """嵌套任务"""
    id: str = Field(description="任务ID，如 1, 1.1, 1.1.1")
    description: str
    subtasks: Optional[List["NestedTask"]] = Field(default=None, description="子任务")

def decompose_nested_task(goal: str, max_depth: int = 2) -> NestedTask:
    """分解嵌套任务"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""你是一个任务规划专家。请将用户目标分解为层次化的任务结构。

要求：
1. 最多分解到 {max_depth} 层深度
2. 每个任务应该足够具体，或者需要进一步分解
3. 使用层级ID标识任务（如 1, 1.1, 1.2）

返回嵌套的任务结构。"""),
        ("user", "{goal}")
    ])
    
    chain = prompt | llm.with_structured_output(NestedTask)
    return chain.invoke({"goal": goal})

def print_nested_task(task: NestedTask, indent: int = 0):
    """打印嵌套任务"""
    prefix = "  " * indent
    print(f"{prefix}{task.id}. {task.description}")
    if task.subtasks:
        for subtask in task.subtasks:
            print_nested_task(subtask, indent + 1)

# 示例
goal = "帮我规划一次三天两夜的旅行"
plan = decompose_nested_task(goal)
print_nested_task(plan)

输出示例：

1. 规划三天两夜旅行
  1.1. 选择目的地
  1.2. 预订交通
    1.2.1. 比较航班价格
    1.2.2. 预订往返机票
  1.3. 预订住宿
    1.3.1. 搜索酒店
    1.3.2. 比较价格和评价
    1.3.3. 完成预订
  1.4. 规划行程
    1.4.1. 规划第一天活动
    1.4.2. 规划第二天活动
    1.4.3. 规划第三天活动

规划模式​

单步规划 vs 动态规划​

单步规划是指在任务开始时一次性生成完整的执行计划，然后按计划执行。这种方式适合结构清晰、变化较少的任务。

def single_step_planning(goal: str) -> List[str]:
    """单步规划：一次性生成完整计划"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = f"""请为以下目标制定一个完整的执行计划：
    
目标：{goal}

要求：
1. 列出所有需要执行的步骤
2. 步骤之间要有清晰的顺序
3. 每个步骤要具体可执行

请直接输出步骤列表，每行一个步骤。"""
    
    response = llm.invoke(prompt)
    steps = [line.strip() for line in response.content.split("\n") if line.strip()]
    return steps

# 执行计划
def execute_plan(steps: List[str]):
    """按计划执行"""
    results = []
    for i, step in enumerate(steps, 1):
        print(f"执行步骤 {i}/{len(steps)}: {step}")
        # 实际执行逻辑...
        results.append(f"步骤 {i} 完成")
    return results

动态规划是指在执行过程中根据实际情况调整计划。当环境变化或遇到意外情况时，Agent 可以重新规划。这种方式更灵活，适合不确定性较高的任务。

from typing import Optional
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

class DynamicPlanner:
    """动态规划器"""
    
    def __init__(self, goal: str):
        self.goal = goal
        self.tasks: List[dict] = []
        self.current_task_idx = 0
        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
        
        # 初始化计划
        self._create_initial_plan()
    
    def _create_initial_plan(self):
        """创建初始计划"""
        prompt = f"""请为以下目标创建执行计划：
        
目标：{self.goal}

返回 JSON 数组格式的任务列表，每个任务包含：
- id: 任务编号
- description: 任务描述
- status: 初始状态为 pending"""
        
        response = self.llm.invoke(prompt)
        import json
        self.tasks = json.loads(response.content)
    
    def get_next_task(self) -> Optional[dict]:
        """获取下一个待执行的任务"""
        for task in self.tasks:
            if task["status"] == TaskStatus.PENDING.value:
                return task
        return None
    
    def update_task_status(self, task_id: int, status: TaskStatus, result: str = ""):
        """更新任务状态"""
        for task in self.tasks:
            if task["id"] == task_id:
                task["status"] = status.value
                task["result"] = result
                break
    
    def replan(self, reason: str):
        """根据情况重新规划"""
        completed = [t for t in self.tasks if t["status"] == TaskStatus.COMPLETED.value]
        failed = [t for t in self.tasks if t["status"] == TaskStatus.FAILED.value]
        
        prompt = f"""原目标：{self.goal}

已完成的任务：
{completed}

失败的任务：
{failed}

重新规划原因：{reason}

请根据当前情况调整计划，返回新的任务列表。"""
        
        response = self.llm.invoke(prompt)
        import json
        new_tasks = json.loads(response.content)
        
        # 合并已完成的任务和新任务
        self.tasks = completed + [t for t in new_tasks if t["id"] not in [c["id"] for c in completed]]
    
    def execute(self):
        """执行计划"""
        while True:
            task = self.get_next_task()
            if not task:
                break
            
            print(f"执行任务: {task['description']}")
            
            try:
                # 执行任务
                result = self._execute_task(task)
                self.update_task_status(task["id"], TaskStatus.COMPLETED, result)
                print(f"任务完成: {task['description']}")
            except Exception as e:
                self.update_task_status(task["id"], TaskStatus.FAILED, str(e))
                print(f"任务失败: {task['description']}, 原因: {e}")
                
                # 动态重新规划
                self.replan(f"任务 {task['description']} 失败: {e}")
        
        print("所有任务执行完毕")

# 使用示例
planner = DynamicPlanner("帮我组织一次团队建设活动")
planner.execute()

Plan-and-Execute 模式​

Plan-and-Execute 模式将规划和执行明确分离：先制定完整的计划，然后按计划执行。这种方式结构清晰，便于调试和监控。

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

class PlanExecuteState(TypedDict):
    """Plan-and-Execute 状态"""
    goal: str
    plan: List[str]
    current_step: int
    results: Annotated[List[str], operator.add]
    final_answer: str

def plan_step(state: PlanExecuteState) -> dict:
    """规划步骤"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = f"""请为以下目标制定详细的执行计划：

目标：{state['goal']}

要求：
1. 将目标分解为具体步骤
2. 步骤要按执行顺序排列
3. 每个步骤应该清晰、可执行

直接输出步骤列表，每行一个步骤。"""
    
    response = llm.invoke(prompt)
    plan = [line.strip() for line in response.content.split("\n") 
            if line.strip() and not line.strip().startswith("#")]
    
    return {"plan": plan, "current_step": 0}

def execute_step(state: PlanExecuteState) -> dict:
    """执行步骤"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    current_idx = state["current_step"]
    current_task = state["plan"][current_idx]
    
    prompt = f"""请执行以下任务：

任务：{current_task}

原始目标：{state['goal']}

已完成的工作：{state.get('results', [])}

请输出执行结果。"""
    
    response = llm.invoke(prompt)
    
    return {
        "current_step": current_idx + 1,
        "results": [f"步骤 {current_idx + 1}: {response.content}"]
    }

def should_continue(state: PlanExecuteState) -> str:
    """判断是否继续执行"""
    if state["current_step"] < len(state["plan"]):
        return "execute"
    return "finalize"

def finalize(state: PlanExecuteState) -> dict:
    """生成最终结果"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = f"""原始目标：{state['goal']}

执行结果：
{chr(10).join(state['results'])}

请根据以上结果，给出最终的总结回答。"""
    
    response = llm.invoke(prompt)
    return {"final_answer": response.content}

# 构建工作流
workflow = StateGraph(PlanExecuteState)

workflow.add_node("plan", plan_step)
workflow.add_node("execute", execute_step)
workflow.add_node("finalize", finalize)

workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_conditional_edges("execute", should_continue, {
    "execute": "execute",
    "finalize": "finalize"
})
workflow.add_edge("finalize", END)

app = workflow.compile()

# 执行
result = app.invoke({
    "goal": "帮我准备一份关于人工智能发展趋势的报告",
    "plan": [],
    "current_step": 0,
    "results": [],
    "final_answer": ""
})

print(result["final_answer"])

自我反思与迭代改进​

自我反思是规划模块的重要能力。Agent 在执行过程中或执行完成后，可以回顾自己的决策和结果，发现问题并改进策略。

反思机制​

from pydantic import BaseModel, Field
from typing import List

class Reflection(BaseModel):
    """反思结果"""
    success: bool = Field(description="任务是否成功")
    analysis: str = Field(description="分析执行过程")
    issues: List[str] = Field(description="发现的问题")
    improvements: List[str] = Field(description="改进建议")
    next_action: str = Field(description="下一步行动建议")

def reflect_on_execution(goal: str, actions: List[str], results: List[str]) -> Reflection:
    """对执行结果进行反思"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = f"""请对以下任务执行过程进行反思：

目标：{goal}

执行的动作：
{chr(10).join(actions)}

得到的结果：
{chr(10).join(results)}

请分析：
1. 任务是否成功完成？
2. 执行过程中有什么问题？
3. 有哪些可以改进的地方？
4. 如果任务未完成，下一步应该做什么？

返回结构化的反思结果。"""
    
    chain = llm.with_structured_output(Reflection)
    return chain.invoke(prompt)

# 示例
goal = "查询北京天气并发送邮件通知"
actions = ["调用天气API", "发送邮件"]
results = ["北京：晴，25°C", "邮件发送失败：SMTP 连接超时"]

reflection = reflect_on_execution(goal, actions, results)
print(f"是否成功: {reflection.success}")
print(f"分析: {reflection.analysis}")
print(f"问题: {reflection.issues}")
print(f"改进建议: {reflection.improvements}")
print(f"下一步: {reflection.next_action}")

ReAct 模式中的反思​

ReAct（Reasoning and Acting）模式天然包含了反思机制。Agent 在每一步都会先"思考"，然后选择"行动"，观察结果后再进行下一轮思考。这个过程本身就是一个持续反思的过程。

问题：找出目前最受欢迎的无线耳机并确认是否有货

思考 1：流行度是时间敏感的，我需要使用搜索工具获取最新信息。
行动 1：search_products("wireless headphones")
观察 1：搜索结果显示 Sony WH-1000XM5 最受欢迎...

思考 2：搜索成功获取了热门产品信息。现在需要确认库存。
行动 2：check_inventory("WH-1000XM5")
观察 2：库存充足，有 50 件...

思考 3：我已经获得了完整信息，可以回答用户的问题。
行动 3：输出最终答案

在 LangChain 中实现带反思的 Agent：

from langchain.agents import create_agent, Middleware
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage

@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果：{query}"

@tool
def check_inventory(product: str) -> str:
    """检查库存"""
    return f"{product} 库存充足"

class ReflectionMiddleware(AgentMiddleware):
    """反思中间件"""
    
    def after_tool_call(self, tool_call, result):
        """工具调用后反思"""
        print(f"[反思] 工具 {tool_call.name} 返回: {result[:100]}...")
        return result
    
    def on_tool_error(self, error, tool_call):
        """工具错误时反思"""
        print(f"[反思] 工具 {tool_call.name} 失败: {error}")
        print("[反思] 尝试其他方法...")
        return f"工具 {tool_call.name} 执行失败，请尝试其他方法。"

llm = ChatOpenAI(model="gpt-4", temperature=0)

agent = create_agent(
    model=llm,
    tools=[search, check_inventory],
    system_prompt="""你是一个智能助手。请使用 ReAct 模式工作：
    
1. 思考：分析当前情况，决定下一步行动
2. 行动：调用工具或输出答案
3. 观察：分析工具返回的结果
4. 重复以上步骤直到完成任务

如果某个工具调用失败，请反思原因并尝试其他方法。""",
    middleware=[ReflectionMiddleware()]
)

# 执行
result = agent.invoke({
    "messages": [HumanMessage(content="找出最受欢迎的无线耳机并确认是否有货")]
})

思维链与推理增强​

Chain of Thought (CoT)​

思维链（Chain of Thought）是一种提示技术，让模型将复杂问题分解为一系列中间推理步骤，而不是直接给出答案。这有助于模型更好地处理需要多步推理的问题。

def chain_of_thought_reasoning(question: str) -> str:
    """使用思维链进行推理"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = f"""请使用思维链方法解决以下问题。

问题：{question}

要求：
1. 一步步思考，展示推理过程
2. 每个推理步骤要清晰明确
3. 最后给出答案

让我们一步步思考："""
    
    response = llm.invoke(prompt)
    return response.content

# 示例
question = "一个篮子里有 15 个苹果，小明拿走了 3 个，小红又放进去了 5 个，请问篮子里现在有多少个苹果？"
answer = chain_of_thought_reasoning(question)
print(answer)

输出示例：

让我们一步步思考：

步骤 1：篮子里最初有 15 个苹果。

步骤 2：小明拿走了 3 个苹果。
       计算：15 - 3 = 12
       现在篮子里有 12 个苹果。

步骤 3：小红又放进去了 5 个苹果。
       计算：12 + 5 = 17
       现在篮子里有 17 个苹果。

答案：篮子里现在有 17 个苹果。

Tree of Thoughts (ToT)​

思维树（Tree of Thoughts）是思维链的扩展，它让模型生成多个可能的推理路径，然后评估每个路径的价值，选择最优路径继续探索。这种方式更适合需要探索和决策的问题。

from typing import List, Tuple
import json

class TreeOfThoughts:
    """思维树推理"""
    
    def __init__(self, max_depth: int = 3, num_thoughts: int = 3):
        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
        self.max_depth = max_depth
        self.num_thoughts = num_thoughts
    
    def generate_thoughts(self, problem: str, current_path: List[str] = None) -> List[str]:
        """生成可能的思考方向"""
        context = "\n".join(current_path) if current_path else "这是开始"
        
        prompt = f"""问题：{problem}

当前思考路径：
{context}

请生成 {self.num_thoughts} 个可能的下一步思考方向。
每个方向应该是一个具体的推理步骤。

以 JSON 数组格式返回，如：["思考1", "思考2", "思考3"]"""
        
        response = self.llm.invoke(prompt)
        try:
            return json.loads(response.content)
        except:
            return response.content.split("\n")
    
    def evaluate_thoughts(self, problem: str, thoughts: List[str], current_path: List[str]) -> List[Tuple[str, float]]:
        """评估每个思考方向的价值"""
        context = "\n".join(current_path) if current_path else ""
        
        prompt = f"""问题：{problem}

当前路径：{context}

候选思考方向：
{chr(10).join([f"{i+1}. {t}" for i, t in enumerate(thoughts)])}

请为每个思考方向评分（0-10分），评估其对解决问题的价值。
返回 JSON 数组，如：[8, 5, 7]"""
        
        response = self.llm.invoke(prompt)
        try:
            scores = json.loads(response.content)
        except:
            scores = [5.0] * len(thoughts)
        
        return list(zip(thoughts, scores))
    
    def solve(self, problem: str) -> str:
        """使用思维树解决问题"""
        best_path = []
        
        def explore(path: List[str], depth: int):
            if depth >= self.max_depth:
                return path
            
            thoughts = self.generate_thoughts(problem, path)
            evaluated = self.evaluate_thoughts(problem, thoughts, path)
            
            # 选择得分最高的方向
            best_thought, best_score = max(evaluated, key=lambda x: x[1])
            new_path = path + [best_thought]
            
            return explore(new_path, depth + 1)
        
        best_path = explore([], 0)
        
        # 基于最佳路径生成最终答案
        prompt = f"""问题：{problem}

思考路径：
{chr(10).join(best_path)}

请根据以上思考路径，给出最终答案。"""
        
        return self.llm.invoke(prompt).content

# 使用示例
tot = TreeOfThoughts(max_depth=3, num_thoughts=3)
problem = "如何在一个月内提高英语口语能力？"
answer = tot.solve(problem)
print(answer)

规划在 LangGraph 中的实现​

使用 LangGraph 可以更灵活地实现各种规划策略。以下是一个完整的规划-执行-反思循环：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Literal
from langchain_openai import ChatOpenAI
import operator

class AgentState(TypedDict):
    """Agent 状态"""
    goal: str
    plan: List[str]
    current_step: int
    execution_results: Annotated[List[dict], operator.add]
    reflection: dict
    iterations: int
    final_answer: str

llm = ChatOpenAI(model="gpt-4", temperature=0)

def plan_node(state: AgentState) -> dict:
    """规划节点"""
    prompt = f"""请为以下目标制定执行计划：

目标：{state['goal']}

{f"之前的尝试失败，原因：{state['reflection'].get('issues', [])}" if state.get('reflection') else ""}

请输出具体的执行步骤列表。"""
    
    response = llm.invoke(prompt)
    plan = [line.strip() for line in response.content.split("\n") 
            if line.strip() and not line.strip().startswith("#")]
    
    return {"plan": plan, "current_step": 0}

def execute_node(state: AgentState) -> dict:
    """执行节点"""
    current_idx = state["current_step"]
    current_task = state["plan"][current_idx]
    
    # 执行任务（这里用 LLM 模拟）
    prompt = f"请执行任务：{current_task}"
    result = llm.invoke(prompt).content
    
    return {
        "current_step": current_idx + 1,
        "execution_results": [{
            "step": current_idx + 1,
            "task": current_task,
            "result": result
        }]
    }

def reflect_node(state: AgentState) -> dict:
    """反思节点"""
    results = state["execution_results"]
    
    prompt = f"""目标：{state['goal']}

执行结果：
{chr(10).join([f"步骤{r['step']}: {r['task']} -> {r['result'][:100]}..." for r in results])}

请反思：
1. 任务是否成功？
2. 有什么问题？
3. 需要重新规划吗？

返回 JSON 格式的反思结果。"""
    
    response = llm.invoke(prompt)
    import json
    try:
        reflection = json.loads(response.content)
    except:
        reflection = {"success": True, "needs_replanning": False}
    
    return {
        "reflection": reflection,
        "iterations": state["iterations"] + 1
    }

def route_after_execution(state: AgentState) -> Literal["execute", "reflect", "end"]:
    """执行后路由"""
    # 检查是否还有未执行的步骤
    if state["current_step"] < len(state["plan"]):
        return "execute"
    
    # 所有步骤完成，进入反思
    return "reflect"

def route_after_reflection(state: AgentState) -> Literal["plan", "finalize", "end"]:
    """反思后路由"""
    reflection = state.get("reflection", {})
    
    # 检查是否需要重新规划
    if reflection.get("needs_replanning", False) and state["iterations"] < 3:
        return "plan"
    
    # 检查是否成功
    if reflection.get("success", True):
        return "finalize"
    
    # 超过最大迭代次数，强制结束
    return "end"

def finalize_node(state: AgentState) -> dict:
    """生成最终结果"""
    results = state["execution_results"]
    
    prompt = f"""目标：{state['goal']}

执行过程：
{chr(10).join([f"步骤{r['step']}: {r['task']}" for r in results])}

执行结果：
{chr(10).join([r['result'] for r in results])}

请给出最终总结。"""
    
    return {"final_answer": llm.invoke(prompt).content}

# 构建图
workflow = StateGraph(AgentState)

workflow.add_node("plan", plan_node)
workflow.add_node("execute", execute_node)
workflow.add_node("reflect", reflect_node)
workflow.add_node("finalize", finalize_node)

workflow.set_entry_point("plan")
workflow.add_edge("plan", "execute")
workflow.add_conditional_edges("execute", route_after_execution, {
    "execute": "execute",
    "reflect": "reflect"
})
workflow.add_conditional_edges("reflect", route_after_reflection, {
    "plan": "plan",
    "finalize": "finalize",
    "end": END
})
workflow.add_edge("finalize", END)

app = workflow.compile()

# 执行
result = app.invoke({
    "goal": "帮我分析最近的科技新闻趋势",
    "plan": [],
    "current_step": 0,
    "execution_results": [],
    "reflection": {},
    "iterations": 0,
    "final_answer": ""
})

print(result["final_answer"])

规划最佳实践​

保持计划的可执行性​

每个子任务都应该足够具体，让 Agent 能够明确知道要做什么。避免过于抽象的计划。

# 不好的计划
bad_plan = [
    "研究市场",
    "分析数据",
    "写报告"
]

# 好的计划
good_plan = [
    "搜索最近一个月的市场分析报告",
    "提取关键数据和趋势信息",
    "对比三个主要竞争对手的市场表现",
    "撰写 500 字的市场趋势分析摘要"
]

设置合理的终止条件​

规划系统需要知道何时停止：

  • 达到目标状态
  • 执行完所有计划步骤
  • 超过最大迭代次数
  • 连续失败次数超过阈值

平衡规划深度与执行效率​

过深的计划分解会增加系统复杂度，过浅则可能导致任务不清晰。根据任务复杂度选择合适的分解深度：

  • 简单任务：1 层分解
  • 中等任务：2 层分解
  • 复杂任务：2-3 层分解，配合动态规划

小结​

规划是 AI Agent 处理复杂任务的核心能力：

  • 任务分解是规划的基础，包括顺序分解、并行分解和嵌套分解
  • Plan-and-Execute 模式将规划和执行分离，结构清晰
  • 自我反思让 Agent 能够评估执行结果，必要时重新规划
  • 思维链和思维树等推理增强技术可以提高规划质量
  • LangGraph 提供了灵活的状态图机制来实现复杂的规划逻辑

下一章我们将学习如何使用 LangChain 构建完整的 Agent 应用。

参考资料​