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[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。让我们一起进化吧。