Agent规划核心概念

1.1 什么是任务规划

当用户提出复合型复杂问题(多步骤、多工具、多知识库查询),大模型无法一次性给出答案,需要先拆解成多个可执行子任务,按顺序分步执行,最后汇总结果。
例:“帮我计算(125+36)*8,同时查询RAG定义,最后汇总成一段总结”
拆解子任务:

  1. 调用计算器计算125+36
  2. 调用计算器计算结果×8
  3. RAG检索RAG相关知识
  4. 整合全部结果输出总结
1.2 两大主流推理框架(面试必考)
(1)ReAct 推理+行动框架(最简单、入门首选)

核心逻辑:Thought→Action→Observation循环

  1. Thought:模型思考当前需要做什么、下一步执行什么工具/检索
  2. Action:输出标准化工具调用/检索指令
  3. Observation:拿到工具/知识库返回结果,作为观察输入下一轮思考
    循环往复直到任务全部完成,输出最终答案。
    优势:实现简单、天然适配Function Calling;缺点:复杂多步骤任务容易跳步、漏任务。
(2)Plan-Solve 先规划后执行框架

两步流程:

  1. Plan阶段:模型一次性输出完整任务清单(结构化JSON子任务列表),包含任务类型、执行顺序、依赖关系
  2. Solve阶段:程序按顺序逐个执行子任务,缓存每个子任务结果,全部完成后统一汇总
    优势:任务清晰可控,便于监控、断点续跑、失败重试;缺点:一次性规划消耗更多Token,复杂依赖场景规划易出错。
1.3 子任务类型分类(统一抽象)

统一封装三类任务,一套调度器兼容所有任务:

  1. calc:数学计算工具调用(复用Day2计算器)
  2. rag_search:知识库检索任务(复用Day3 RAG)
  3. llm_summary:纯文本推理总结任务(无需外部工具)
1.4 任务依赖规则
  • 无依赖任务:可并行执行(多个独立RAG查询)
  • 强依赖任务:必须等待前置子任务完成才能执行(先求和再相乘)

2. 任务调度与状态管理

2.1 任务状态枚举
pending:待执行
running:执行中
success:执行成功
failed:执行失败(支持重试)
finished:全部完成
2.2 核心调度能力
  1. 任务缓存:存储每个子任务ID、类型、入参、执行结果、状态
  2. 重试机制:单个子任务失败最多重试2次,仍失败标记任务异常
  3. 执行顺序控制:区分串行依赖、并行独立任务
  4. 终止条件:所有子任务success,或达到最大规划轮次强制汇总

3. 规划Agent上下文与Token优化

  1. 规划阶段仅传入用户原始问题,不携带冗余历史,减少规划开销
  2. 每个子任务执行结果精简压缩,避免大量文本累积超限
  3. 设置最大规划轮次(默认5轮),防止无限循环拆解任务

4. 完整全能Agent链路(Day1~Day4全能力整合)

用户复杂提问
→ Plan:模型拆解结构化子任务列表
→ 调度器循环执行每个子任务
- 子任务=计算:调用Function Calling计算器
- 子任务=知识库查询:执行RAG检索
- 子任务=文本推理:直接LLM生成
→ 缓存所有子任务执行结果
→ LLM汇总全部子任务输出,生成最终完整回答

二、今日学习重点

  1. 掌握ReAct与Plan-Solve两种Agent推理框架区别与适用场景
  2. 定义标准化子任务JSON Schema,强制模型输出任务清单
  3. 实现通用任务调度器,支持任务状态、重试、串行执行
  4. 整合LLM、RAG、Function Calling、任务规划一体化Agent
  5. 处理规划异常:任务格式错乱、子任务重复、依赖顺序错误

三、今日难点 & 解决方案

难点1:模型拆解任务格式混乱,无法解析子任务列表

解决方案:

  1. 使用Pydantic定义任务列表Schema,Prompt强制输出纯JSON
  2. temperature设为0,消除随机性
  3. 正则提取JSON、解析失败自动重新规划一次
  4. Few-shot给出标准任务拆解示例

难点2:子任务执行失败导致整体流程中断

解决方案:

  1. 单个任务捕获全部异常,标记failed,记录错误信息继续执行剩余任务
  2. 配置单任务最大重试次数,重试失败后在最终汇总中提示异常
  3. 汇总阶段告知用户哪些任务执行失败,给出失败原因

难点3:模型无限拆解、产生大量冗余子任务

解决方案:

  1. 全局限制最大规划轮次,到达上限停止拆解直接汇总已有结果
  2. Prompt约束:无需拆分的简单问题直接回答,禁止多余子任务
  3. 任务去重:调度器过滤重复类型、重复入参的子任务

难点4:多子任务结果过长,Token超限

解决方案:

  1. 每个子任务执行后自动精简输出文本,剔除换行、冗余描述
  2. 汇总前做滑动窗口裁剪,只保留关键任务结果
  3. 长结果分段摘要压缩

四、完整练习代码(基于Day1/2/3扩展)

前置依赖

沿用前几日所有依赖:aiohttppydanticfaiss-cpunumpyfastapiuvicorn,最后总结提示词有bug需要提示词里加前置任务的结果

1. 任务规划核心模块 task_planner.py

import asyncio
import re
import json
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
# 复用已有能力
from llm_client_v2 import AsyncLLMClientV2, TOOLS, CalcToolParams
from rag_store import RAGService

# ========== 1. 子任务结构化Schema ==========
class SubTask(BaseModel):
    task_id: str = Field(description="任务唯一id,如t1/t2")
    task_type: str = Field(description="任务类型:calc/rag_search/llm_summary")
    content: str = Field(description="任务执行入参,计算填表达式,rag填查询词")
    rely_task_ids: List[str] = Field(default=[], description="依赖前置任务id,无依赖为空")

class TaskPlan(BaseModel):
    task_list: List[SubTask] = Field(description="完整子任务列表")

# ========== 2. 任务执行缓存与调度器 ==========
class TaskScheduler:
    def __init__(self, llm_client: AsyncLLMClientV2, rag: RAGService):
        self.llm = llm_client
        self.rag = rag
        self.task_cache: Dict[str, Dict] = {}  # 存储任务状态、结果
        self.max_retry = 2  # 单任务最大重试次数

    # 初始化任务
    def init_tasks(self, task_list: List[SubTask]):
        for task in task_list:
            self.task_cache[task.task_id] = {
                "info": task,
                "status": "pending",
                "result": "",
                "retry_count": 0
            }

    # 判断依赖任务是否全部完成
    def is_rely_finish(self, task: SubTask) -> bool:
        for tid in task.rely_task_ids:
            item = self.task_cache.get(tid)
            if not item or item["status"] not in ("success", "failed"):
                return False
        return True

    # 执行单个子任务
    async def run_single_task(self, task: SubTask) -> str:
        if task.task_type == "calc":
            # 调用计算器工具
            messages = [{"role": "user", "content": f"计算:{task.content}"}]
            return await self.llm.chat_with_tools(messages, TOOLS)
        elif task.task_type == "rag_search":
            # RAG知识库检索
            res = await self.rag.retrieve(task.content, top_k=2)
            return "\n".join([i["text"] for i in res])
        elif task.task_type == "llm_summary":
            # 纯文本推理
            messages = [{"role": "user", "content": task.content}]
            return await self.llm.chat(messages)
        else:
            return f"不支持的任务类型:{task.task_type}"

    # 批量调度执行所有任务(串行依赖)
    async def run_all_tasks(self):
        all_task_ids = list(self.task_cache.keys())
        finished = set()
        while len(finished) < len(all_task_ids):
            has_new_run = False
            for tid in all_task_ids:
                item = self.task_cache[tid]
                if item["status"] != "pending":
                    continue
                task_info = item["info"]
                # 依赖未完成跳过
                if not self.is_rely_finish(task_info):
                    continue
                has_new_run = True
                item["status"] = "running"
                try:
                    result = await self.run_single_task(task_info)
                    item["result"] = result
                    item["status"] = "success"
                except Exception as e:
                    item["retry_count"] += 1
                    if item["retry_count"] < self.max_retry:
                        item["status"] = "pending"
                    else:
                        item["status"] = "failed"
                        item["result"] = f"任务执行失败:{str(e)}"
                finished.add(tid)
            if not has_new_run:
                break
        # 汇总所有任务结果
        summary_text = ""
        for tid in all_task_ids:
            t = self.task_cache[tid]
            summary_text += f"【{tid}】类型{t['info'].task_type} 状态{t['status']}:\n{t['result']}\n\n"
        return summary_text

# ========== 3. Plan-Solve规划器 ==========
class TaskPlanner:
    def __init__(self, model_name="qwen-turbo"):
        self.llm = AsyncLLMClientV2(model_name)
        self.rag = RAGService(model_name)
        self.scheduler = TaskScheduler(self.llm, self.rag)
        self.max_plan_round = 5
        # 规划提示词
        self.plan_prompt = """
你是任务拆解专家,将用户复杂问题拆分为子任务,严格输出JSON,禁止多余文字。
支持3种任务类型:
1. calc:数学计算,content填需要计算的表达式
2. rag_search:查询知识库,content填检索关键词
3. llm_summary:纯文本推理总结

规则:
1. 有先后计算关系必须填写rely_task_ids依赖;无依赖填空数组
2. 简单无需拆分的问题,只生成1条llm_summary任务
3. 输出格式严格遵循下面JSON Schema:
{TaskPlanSchema}
用户问题:{query}
"""

    async def create_plan(self, user_query: str) -> TaskPlan:
        schema = TaskPlan.model_json_schema()
        prompt = self.plan_prompt.format(TaskPlanSchema=schema, query=user_query)
        raw = await self.llm.chat([{"role": "user", "content": prompt}], temperature=0.0)
        # 提取JSON
        match = re.search(r"\{.*\}", raw, re.S)
        if not match:
            raw = await self.llm.chat([{"role": "user", "content": prompt}], temperature=0.0)
            match = re.search(r"\{.*\}", raw, re.S)
            if not match:
                raise Exception("任务规划JSON解析失败")
        json_str = match.group()
        return TaskPlan.model_validate_json(json_str)

更多推荐