AI 搜索范式（4）
“运筹如织千机密，擘理似裁万缕匀。智绘蓝图分经纬，巧梳脉络见本真。”
上回书说到，AI 搜索中多智能体的 Master Agent，它是整个 AI 搜索系统的"大脑中枢"和"总指挥"。
image-3.png
接下来为你详细解析 Planner Agent。它是 AI 搜索范式中的“总设计师”和“首席策略师”，负责将用户模糊的“想法”转变为一个清晰、可执行的“行动蓝图”。
Planner Agent
职责概览
如果说 Master 是决定“做不做”和“谁来做”的决策者，那么 Planner 就是决定“怎么做”的规划者。它的核心职责可以分解为三大步：
class PlannerAgent:
"""
Planner 智能体的三大核心职责:
1. 工具集检索 (Toolset Retrieval):
- 根据查询意图，从海量工具中检索出一个完整、高效的工具组合。
2. 任务分解 (Task Decomposition):
- 将复杂的用户查询分解成一系列原子化的、可执行的子任务。
3. 依赖关系建模 (Dependency Modeling):
- 识别子任务间的逻辑依赖，并构建成一个DAG（有向无环图）执行计划。
"""
def __init__(self):
self.tool_retriever = CollaborativeToolRetriever()
self.task_decomposer = TaskDecomposerLLM()
self.dag_builder = DAGBuilder()
职责1: 工具集检索 (Toolset Retrieval)**
原理与实现
Planner 的第一步不是直接规划，而是先“清点装备”。它需要根据用户查询，从庞大的 MCP 工具平台中，选取一小撮但功能完备的工具集，为后续规划做准备。
class CollaborativeToolRetriever:
"""
协作式工具检索器 - Planner 的关键组件
"""
def __init__(self):
self.semantic_encoder = BERTEncoder()
# 预先构建好的工具协作图
self.collaboration_graphs = self.load_collaboration_graphs()
def get_tools_for_query(self, query: str) -> List[Tool]:
"""
基于 COL T方法，检索一个功能完备的工具集
"""
# 步骤1: 语义召回 (Semantic Recall)
# 根据查询与工具描述的语义相似度，初步召回一批相关工具
semantic_candidates = self.semantic_match(query)
# 步骤2: 场景识别 (Scene Identification)
# 判断查询属于哪个任务场景（如：旅行规划、金融分析、健康咨询）
scene = self.identify_task_scene(query)
# 步骤3: 协作扩展 (Collaborative Expansion)
# 基于场景，从协作图中找到必须协同工作的其他工具，补齐工具集
# 例子: 如果识别为"旅行规划"场景，即使只匹配到"机票"工具，
# 也会自动补齐"酒店"和"天气"工具。
complete_toolset = self.expand_based_on_scene(semantic_candidates, scene)
return complete_toolset
职责2 & 3: 任务分解与依赖建模 (生成DAG)
这是 Planner 最核心的功能：创建执行蓝图。
原理: Chain-of-Thought → Structured-Sketch
Planner 通过一个两阶段的提示过程来生成高质量的 DAG，这极大地提升了规划的稳定性和准确性。
class TaskDecomposerLLM:
"""
任务分解器 - 使用两阶段提示策略生成 DAG
"""
def generate_plan_dag(self, query: str, available_tools: List[Tool]) -> DAG:
"""
生成 DAG 的完整过程
"""
# 阶段1: 链式思考 (Chain-of-Thought) - 生成自然语言的思考过程
cot_prompt = self._build_cot_prompt(query, available_tools)
reasoning_text = self.llm_generate(cot_prompt)
# 阶段2: 结构化草图 (Structured-Sketch) - 将思考过程转换为JSON格式的DAG
sketch_prompt = self._build_sketch_prompt(reasoning_text)
dag_json_str = self.llm_generate(sketch_prompt)
try:
dag_data = json.loads(dag_json_str)
return DAG.from_dict(dag_data)
except json.JSONDecodeError:
# 如果 LLM 输出的 JSON 格式错误，可以进行重试或错误处理
raise PlanGenerationError("Failed to generate a valid DAG JSON.")
def _build_cot_prompt(self, query: str, tools: List[Tool]) -> str:
"""构建第一阶段的思考提示"""
tool_descriptions = "\n".join([f"- {t.name}: {t.description}" for t in tools])
return f"""
# 任务: 为以下用户查询制定一个详细的、分步骤的行动计划。
# 用户查询:
"{query}"
# 可用工具:
{tool_descriptions}
# 思考过程 (请按以下逻辑思考):
1. **理解目标**: 用户的最终目的是什么？
2. **分解步骤**: 为了达成 这个目标，需要哪些独立的步骤？
3. **识别依赖**: 这些步骤之间是否存在先后顺序？后一步是否需要前一步的结果作为输入？
4. **匹配工具**: 每个步骤应该使用哪个工具来完成？
5. **确定输入输出**: 每个步骤需要什么输入，会产出什么结果？
# 请输出你的自然语言思考过程:
"""
def _build_sketch_prompt(self, reasoning_text: str) -> str:
"""构建第二阶段的结构化转换提示"""
return f"""
# 任务: 将以下自然语言的行动计划，转换为严格的JSON格式的DAG。
# 行动计划:
{reasoning_text}
# JSON格式要求:
- `vertices`: 一个列表，每个元素代表一个子任务(节点)。
- `id`: 任务的唯一标识符 (e.g., "task_1")。
- `description`: 对该任务的简短描述。
- `tool_binding`: 完成该任务所需工具的名称。
- `args`: 一个列表，包含该任务所需的输入参数。如果参数来自其他任务的输出，请使用`"{{task_id.ret}}"`的格式。
- `ret`: 该任务输出结果的变量名。
- `edges`: 一个列表，每个元素是一个二元组 `[source_task_id, target_task_id]`，表示依赖关系。
# 请输出严格的JSON对象:
"""
Planner 与其他 Agent 的协同机制
Planner 在整个工作流中扮演着承上启下的关键角色。
graph TD
subgraph Master's Role
A[Master Agent] -- "1. Query & Available Tools" --> B
end
subgraph Planner's Role
B(Planner Agent)
B -- "2. Retrieve Toolset" --> C[CollaborativeToolRetriever]
C --> B
B -- "3. Generate DAG" --> D[TaskDecomposerLLM]
D --> B
end
subgraph Executor's Role
B -- "4. Executable Plan (DAG)" --> E[Executor Agent]
end
subgraph Master's Feedback Loop
F[Executor Fails] --> G[Master Agent]
G -- "5. Re-plan Request & Failure Info" --> B
end
-
与Master的协同 (输入):
-
Master在判断查询为“复杂”后，会唤醒Planner。 -
Master向Planner传递两个关键信息：① 原始用户查询；② 一个初步筛选过的、与查询可能相关的可用工具列表。
-
-
与 Executor 的协同 (输出):
-
Planner的最终产出是一个结构化的 DAG 计划。 - 这个 DAG 是
Executor的“工作指令”，Executor会严格按照 DAG 的拓扑顺序和工具绑定来执行任务。
-
-
与 Master 的协同 (重规划):
- 如果
Executor在执行过程中遇到失败（如工具 API 超时、返回结果无效），它会将失败信息上报给Master。 -
Master进行反思后，会向Planner发出“重规划”请求，并附上失败的上下文信息（哪个任务失败了，失败原因是什么）。 -
Planner接收到这些信息后，可以进行局部重规划，例如只修改 DAG 中的一个节点（换一个工具）或增加一个补充步骤，而无需从头开始。
- 如果
完整实例代码演示
# 假设已有 Tool, DAG, LLM 等基础类
# ...
class PlannerAgent:
def __init__(self):
self.tool_retriever = CollaborativeToolRetriever()
self.task_decomposer = TaskDecomposerLLM()
self.plan_validator = PlanValidator()
def create_plan(self, query: str, available_tools: List[Tool]) -> DAG:
print(f"--- Planner received query: '{query}' ---")
# 1. 检索完成任务所需的工具集
print("Step 1: Retrieving collaborative toolset...")
toolset = self.tool_retriever.get_tools_for_query(query)
print(f" > Found tools: {[t.name for t in toolset]}")
# 2. 生成 DAG 计划
print("Step 2: Generating DAG plan using CoT -> Sketch...")
try:
dag = self.task_decomposer.generate_plan_dag(query, toolset)
print(" > DAG generated successfully.")
except PlanGenerationError as e:
print(f" > ERROR: {e}")
# 可以加入重试逻辑
return None
# 3. 验证计划的健全性
print("Step 3: Validating the generated plan...")
try:
self.plan_validator.validate_plan(dag, toolset)
print(" > Plan is valid and executable.")
except PlanValidationError as e:
print(f" > ERROR: Plan is invalid. Reason: {e}")
# 可以触发带错误信息的重规划
return None
print("--- Planner finished. Returning executable DAG. ---")
return dag
# --- 模拟执行 ---
if __name__ == "__main__":
# 模拟环境
mock_tools = [
Tool(name="Web_Search_Tool", description="Search the web for information."),
Tool(name="Calculator_Tool", description="Perform mathematical calculations.")
]
planner = PlannerAgent()
complex_query = "Who was older, Emperor Wu of Han or Julius Caesar, and by how many years?"
# Master 调用 Planner
final_plan = planner.create_plan(complex_query, mock_tools)
if final_plan:
print("\n--- Final Executable DAG Plan ---")
print(final_plan.to_pretty_json())
总结
通过以上设计，Planner Agent 实现了：
- 整个 AI 搜索范式从“理解问题”到“解决问题”的桥梁。
- 它通过强大的逻辑推理和规划能力，将用户的自然语言查询，系统化、结构化地转变为一个由机器组成的“项目执行团队”可以理解并遵循的行动路线图，是整个系统智能性的核心体现。
预知 Executor Agent 如何工作，我们下回分解。
大家好，我是自在哪吒的创始人、首席服务官 Kafka。让我们一起进化吧。
