一、核心理论教学笔记

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

调用计算器计算125+36
调用计算器计算结果×8
RAG检索RAG相关知识
整合全部结果输出总结
1.2 两大主流推理框架（面试必考）
（1）ReAct 推理+行动框架（最简单、入门首选）
核心逻辑：Thought→Action→Observation循环

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

Plan阶段：模型一次性输出完整任务清单（结构化JSON子任务列表），包含任务类型、执行顺序、依赖关系
Solve阶段：程序按顺序逐个执行子任务，缓存每个子任务结果，全部完成后统一汇总
优势：任务清晰可控，便于监控、断点续跑、失败重试；缺点：一次性规划消耗更多Token，复杂依赖场景规划易出错。
1.3 子任务类型分类（统一抽象）
统一封装三类任务，一套调度器兼容所有任务：

calc：数学计算工具调用（复用Day2计算器）
rag_search：知识库检索任务（复用Day3 RAG）
llm_summary：纯文本推理总结任务（无需外部工具）
1.4 任务依赖规则
无依赖任务：可并行执行（多个独立RAG查询）
强依赖任务：必须等待前置子任务完成才能执行（先求和再相乘）
2. 任务调度与状态管理
2.1 任务状态枚举
pending：待执行
running：执行中
success：执行成功
failed：执行失败（支持重试）
finished：全部完成
2.2 核心调度能力
任务缓存：存储每个子任务ID、类型、入参、执行结果、状态
重试机制：单个子任务失败最多重试2次，仍失败标记任务异常
执行顺序控制：区分串行依赖、并行独立任务
终止条件：所有子任务success，或达到最大规划轮次强制汇总
3. 规划Agent上下文与Token优化
规划阶段仅传入用户原始问题，不携带冗余历史，减少规划开销
每个子任务执行结果精简压缩，避免大量文本累积超限
设置最大规划轮次（默认5轮），防止无限循环拆解任务
4. 完整全能Agent链路（Day1~Day4全能力整合）
用户复杂提问
→ Plan：模型拆解结构化子任务列表
→ 调度器循环执行每个子任务

• 子任务=计算：调用Function Calling计算器
• 子任务=知识库查询：执行RAG检索
• 子任务=文本推理：直接LLM生成
  → 缓存所有子任务执行结果
  → LLM汇总全部子任务输出，生成最终完整回答

二、今日学习重点
掌握ReAct与Plan-Solve两种Agent推理框架区别与适用场景
定义标准化子任务JSON Schema，强制模型输出任务清单
实现通用任务调度器，支持任务状态、重试、串行执行
整合LLM、RAG、Function Calling、任务规划一体化Agent
处理规划异常：任务格式错乱、子任务重复、依赖顺序错误
三、今日难点 & 解决方案
难点1：模型拆解任务格式混乱，无法解析子任务列表
解决方案：

使用Pydantic定义任务列表Schema，Prompt强制输出纯JSON
temperature设为0，消除随机性
正则提取JSON、解析失败自动重新规划一次
Few-shot给出标准任务拆解示例
难点2：子任务执行失败导致整体流程中断
解决方案：

单个任务捕获全部异常，标记failed，记录错误信息继续执行剩余任务
配置单任务最大重试次数，重试失败后在最终汇总中提示异常
汇总阶段告知用户哪些任务执行失败，给出失败原因
难点3：模型无限拆解、产生大量冗余子任务
解决方案：

全局限制最大规划轮次，到达上限停止拆解直接汇总已有结果
Prompt约束：无需拆分的简单问题直接回答，禁止多余子任务
任务去重：调度器过滤重复类型、重复入参的子任务
难点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. 任务执行缓存与调度器 ==========

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())
2. FastAPI一体化接口 main_plan.py
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)
五、今日必做练习任务
填入API Key，运行task_planner.py，观察复杂问题自动拆解多子任务
构造带依赖计算问题（如(10+5)*6），验证rely_task_ids依赖执行顺序
构造纯知识库提问，测试仅生成rag_search任务
手动修改代码制造任务异常，验证重试机制、失败标记逻辑
启动FastAPI，访问/docs接口，连续提交复合型问题查看完整链路返回
修改规划Prompt，测试模型输出非法JSON，观察自动重试规划逻辑
六、今日配套面试题（Agent进阶核心）
基础问答
ReAct和Plan-Solve两种Agent推理框架分别是什么？各自优缺点？
什么是任务依赖？为什么需要rely_task_ids？
Agent任务规划解决了大模型什么痛点？
任务调度器需要具备哪些基础能力？
规划阶段为什么temperature必须设为0？
工程实操题
模型输出的任务列表JSON格式错乱，如何多层兜底处理？
子任务执行报错，如何保证整体流程不中断、不卡死？
用户问题简单无需拆分，如何约束模型不生成多余子任务？
多子任务结果文本过长，如何防止汇总阶段Token超限？
拓展思考题（高级Agent面试）
如何实现并行任务执行？适用于什么场景？
如何给Agent增加动态反思能力（ReAct循环自我修正）？
线上多用户场景，任务缓存如何持久化？
复杂多步骤业务流程，如何实现断点续跑、中途人工干预？
面试题标准答案
基础问答
ReAct vs Plan-Solve
ReAct：Thought思考→Action执行→Observation结果循环，边执行边思考；优点轻量、代码简单；缺点长流程容易遗漏步骤。
Plan-Solve：先一次性生成完整任务清单，再批量执行所有子任务；优点任务可控、便于监控重试；缺点一次性规划消耗Token，复杂依赖易规划出错。
任务依赖rely_task_ids
部分计算、推理任务需要前置任务结果作为输入，依赖ID标记前置任务，调度器等待依赖全部完成再执行当前任务，保证执行顺序正确。

规划解决的痛点
用户复合型多步骤问题，模型无法一次性完成；拆分后分步调用工具、检索知识库，降低单次推理压力，提升结果准确率，便于错误局部重试。

调度器基础能力
任务状态管理、依赖判断、串行执行、失败重试、结果缓存、任务汇总。

规划temperature=0原因
任务清单是结构化JSON，要求字段、类型、任务ID完全规范；高随机性会导致字段缺失、格式错乱，0温度保证输出稳定合规。

工程实操题
JSON格式兜底方案
① Prompt强约束输出纯JSON；② temperature=0；③ 正则提取大括号内容；④ 解析失败自动重新规划一次；⑤ Pydantic模型强制校验。
任务异常不中断流程
每个子任务独立try/except捕获异常；失败标记状态、记录错误信息；达到重试上限后跳过，继续执行剩余任务；汇总阶段展示失败任务提示用户。
禁止多余子任务
Prompt增加约束：简单单一问题仅生成一条llm_summary；Few-shot示例演示简单问题不拆分；调度器自动合并重复同类型任务。
避免汇总Token超限
子任务执行后精简结果文本；限制最大规划轮次；汇总前对长任务结果做摘要压缩；滑动窗口裁剪冗余内容。
拓展思考题
并行任务实现
筛选无依赖的任务，使用asyncio.gather并发执行；适合多个独立知识库查询、互不关联的外部接口调用，大幅提升执行效率。
ReAct反思修正
每轮执行完观察结果后，让模型反思当前信息是否足够、是否需要补充工具/检索，循环迭代直到信息充足再输出答案，适合动态多变的开放式问题。
任务缓存持久化
使用Redis存储每个用户会话的任务ID、状态、结果；设置过期时间，多请求之间共享任务进度。
断点续跑与人工干预
任务状态持久化存储；增加人工中断接口，可手动修改任务结果、新增/删除子任务；调度器读取最新状态继续执行未完成任务。
学习总结
Day4完成AI Agent核心能力——任务自主规划，整合前三天全部底层能力：异步LLM、SSE流式、Function Calling、多轮对话、RAG知识库、任务调度。