当用户提出复合型复杂问题（多步骤、多工具、多知识库查询），大模型无法一次性给出答案，需要先拆解成多个可执行子任务，按顺序分步执行，最后汇总结果。
例：“帮我计算(125+36)*8，同时查询RAG定义，最后汇总成一段总结”
拆解子任务：
核心逻辑：Thought→Action→Observation循环
两步流程：
统一封装三类任务，一套调度器兼容所有任务：
calc：数学计算工具调用（复用Day2计算器）rag_search：知识库检索任务（复用Day3 RAG）llm_summary：纯文本推理总结任务（无需外部工具）pending：待执行
running：执行中
success：执行成功
failed：执行失败（支持重试）
finished：全部完成
用户复杂提问
→ Plan：模型拆解结构化子任务列表
→ 调度器循环执行每个子任务
- 子任务=计算：调用Function Calling计算器
- 子任务=知识库查询：执行RAG检索
- 子任务=文本推理：直接LLM生成
→ 缓存所有子任务执行结果
→ LLM汇总全部子任务输出，生成最终完整回答
解决方案：
解决方案：
解决方案：
解决方案：
沿用前几日所有依赖：aiohttp、pydantic、faiss-cpu、numpy、fastapi、uvicorn,最后总结提示词有bug需要提示词里加前置任务的结果
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)
# 完整规划+执行+汇总流程
async def run_full_agent(self, user_query: str):
# 1. 生成任务清单
plan = await self.create_plan(user_query)
# 2. 调度器初始化任务
self.scheduler.init_tasks(plan.task_list)
# 3. 执行所有子任务，拿到汇总原始结果
task_summary = await self.scheduler.run_all_tasks()
# 4. LLM整合所有子任务输出最终回答
final_prompt = f"""
用户原始问题：{user_query}
各子任务执行结果：
{task_summary}
基于上面所有任务结果，整合输出完整通顺的最终答案，不要遗漏计算与知识库信息。
"""
final_ans = await self.llm.chat([{"role": "user", "content": final_prompt}], temperature=0.1)
return {
"user_query": user_query,
"task_list": [t.model_dump() for t in plan.task_list],
"task_detail": self.scheduler.task_cache,
"task_raw_summary": task_summary,
"final_answer": final_ans
}
# 测试入口
async def test_planner():
planner = TaskPlanner()
# 预先导入知识库
await planner.rag.add_document("RAG是检索增强生成，用于读取私有知识库减少模型幻觉", source="RAG文档")
# 复合型复杂问题
question = "先计算(100+25)*4，再查询什么是RAG，最后把计算结果和RAG介绍整合一段总结"
res = await planner.run_full_agent(question)
print("=== 拆解任务列表 ===")
for t in res["task_list"]:
print(t)
print("\n=== 最终回答 ===")
print(res["final_answer"])
if __name__ == "__main__":
asyncio.run(test_planner())
from fastapi import FastAPI, Query
import asyncio
from task_planner import TaskPlanner
app = FastAPI(title="Day4 任务规划Agent｜Plan-Solve全能力整合")
planner = TaskPlanner("qwen-turbo")
# 启动加载知识库
@app.on_event("startup")
async def load_kb():
doc = """
1. Function Calling：大模型自主调用外部工具，支持计算、接口查询
2. RAG检索增强：私有文档向量化存储，相似度召回参考资料
3. Agent任务规划：复杂问题拆分为多步骤子任务分步执行
"""
await planner.rag.add_document(doc, source="Day4知识库")
@app.get("/agent/plan")
async def agent_plan(prompt: str = Query(..., description="复杂复合型提问")):
result = await planner.run_full_agent(prompt)
return result
if __name__ == "__main__":
import uvicorn
uvicorn.run("main_plan.py", reload=True)
task_planner.py，观察复杂问题自动拆解多子任务任务依赖rely_task_ids
部分计算、推理任务需要前置任务结果作为输入，依赖ID标记前置任务，调度器等待依赖全部完成再执行当前任务，保证执行顺序正确。
规划解决的痛点
用户复合型多步骤问题，模型无法一次性完成；拆分后分步调用工具、检索知识库，降低单次推理压力，提升结果准确率，便于错误局部重试。
调度器基础能力
任务状态管理、依赖判断、串行执行、失败重试、结果缓存、任务汇总。
规划temperature=0原因
任务清单是结构化JSON，要求字段、类型、任务ID完全规范；高随机性会导致字段缺失、格式错乱，0温度保证输出稳定合规。
Day4完成AI Agent核心能力——任务自主规划，整合前三天全部底层能力：异步LLM、SSE流式、Function Calling、多轮对话、RAG知识库、任务调度。
Plan-Solve框架是工业级简易Agent主流实现方案，任务拆解、调度、重试、状态管理是工程落地高频考点。
至此基础全能Agent链路全部闭环，后续课程将进阶记忆持久化、ReAct动态智能体、工具网关、多模态RAG、Agent应用部署上线。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。