AI 搜索会是什么样?——百度 AI 搜索新范式(4)
作者：微信文章“运筹如织千机密，擘理似裁万缕匀。智绘蓝图分经纬，巧梳脉络见本真。”
上回书说到，AI 搜索中多智能体的 Master Agent，它是整个 AI 搜索系统的"大脑中枢"和"总指挥"。
AI 搜索范式概览
接下来为你详细解析 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: """ 基于 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) -> 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) -> str: """构建第一阶段的思考提示""" tool_descriptions = "\n".join() 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`: 一个列表，每个元素是一个二元组 ``，表示依赖关系。
# 请输出严格的JSON对象: """Planner 与其他 Agent 的协同机制
Planner 在整个工作流中扮演着承上启下的关键角色。
graph TD subgraph Master's Role A -- "1. Query & Available Tools" --> B end
subgraph Planner's Role B(Planner Agent) B -- "2. Retrieve Toolset" --> C C --> B B -- "3. Generate DAG" --> D D --> B end
subgraph Executor's Role B -- "4. Executable Plan (DAG)" --> E end
subgraph Master's Feedback Loop F --> G 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) -> 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: {}")
# 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。让我们一起进化吧。
页:
[1]