📰 来源: 博客园
每日时长分配(全天8h)
1. Agent规划核心概念
1.1 什么是任务规划
当用户提出复合型复杂问题(多步骤、多工具、多知识库查询),大模型无法一次性给出答案,需要先拆解成多个可执行子任务,按顺序分步执行,最后汇总结果。
例:“帮我计算(125+36)*8,同时查询RAG定义,最后汇总成一段总结”
拆解子任务:
1.2 两大主流推理框架(面试必考)
核心逻辑:Thought→Action→Observation循环
循环往复直到任务全部完成,输出最终答案。
优势:实现简单、天然适配Function Calling;缺点:复杂多步骤任务容易跳步、漏任务。
优势:任务清晰可控,便于监控、断点续跑、失败重试;缺点:一次性规划消耗更多Token,复杂依赖场景规划易出错。
1.3 子任务类型分类(统一抽象)
统一封装三类任务,一套调度器兼容所有任务:
calc:数学计算工具调用(复用Day2计算器)rag_search:知识库检索任务(复用Day3 RAG)llm_summary:纯文本推理总结任务(无需外部工具)2. 任务调度与状态管理
pending:待执行
running:执行中
success:执行成功
failed:执行失败(支持重试)
finished:全部完成
3. 规划Agent上下文与Token优化
4. 完整全能Agent链路(Day1~Day4全能力整合)
用户复杂提问
→ Plan:模型拆解结构化子任务列表
→ 调度器循环执行每个子任务
- 子任务=计算:调用Function Calling计算器
- 子任务=知识库查询:执行RAG检索
- 子任务=文本推理:直接LLM生成
→ 缓存所有子任务执行结果
→ LLM汇总全部子任务输出,生成最终完整回答
三、今日难点 & 解决方案
难点1:模型拆解任务格式混乱,无法解析子任务列表
难点2:子任务执行失败导致整体流程中断
难点3:模型无限拆解、产生大量冗余子任务
难点4:多子任务结果过长,Token超限
四、完整练习代码(基于Day1/2/3扩展)
沿用前几日所有依赖:aiohttp、pydantic、faiss-cpu、numpy、fastapi、uvicorn,最后总结提示词有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. 任务执行缓存与调度器 ==========
🔗 原文链接: 点击阅读原文
文章评论