一、今日总学习目标（全天8h分配）
时长拆分
理论学习：2.5h
分文件代码编写、单元调试：4h
面试题复盘、流程梳理背诵：1.5h
学习目标
清晰认知单一体Agent的短板，掌握中控调度+垂直子Agent分治设计思想
独立分层实现全套底层基础设施，每模块解耦、可单独复用
实现消息总线，完成多Agent并行/串行任务调度、结果统一存储
搭建完整业务链路：会话记忆→RAG知识库检索→工具函数调用→ReAct反思校验→多智能体汇总输出
掌握工程分层规范，学会拆分项目文件，提升代码可维护性
二、核心理论教学笔记
- 单一体Agent的致命缺陷
提示词臃肿：计算、检索、规划、记忆全部塞入同一个系统prompt，模型极易混淆能力边界
推理不稳定：复杂多步骤任务容易漏执行步骤、选错工具、大量生成幻觉
上下文持续膨胀：无分层隔离，长对话快速Token超限，出现对话失忆
维护成本极高：新增工具/知识库需要大面积修改主Agent代码，耦合严重 - 多智能体分层架构（标准工业分层）
四层完整结构，自上而下：
接口层（main.py）：FastAPI提供SSE流式、多智能体对话接口，接收用户请求
调度总线层（multi_agent.py）
消息总线：统一存储各子Agent执行结果，通过flow_id隔离单次工作流
Dispatcher中控调度Agent：解析用户需求，自动拆分任务清单
ReAct反思模块：循环校验当前信息是否充足，不足自动补充检索/计算
垂直子Agent：Retrieve检索Agent、Calc计算Agent，职责单一互不干扰
基础能力层
LLM客户端（llm_client.py）：兼容OpenAI标准，同步/流式输出、结构化JSON强解析
RAG向量检索（rag_store.py）：文本重叠分块、简易余弦相似度向量检索
工具函数（tool_func.py）：标准化计算器工具，Pydantic参数校验
持久记忆层（memory_store.py）：双层压缩会话记忆（滑动窗口+摘要压缩），多会话隔离
3. 任务执行两种模式
并行执行：无依赖任务同时运行（同时检索知识库+数学计算），大幅提升响应速度
串行执行：存在数据依赖时分步执行（必须先拿到检索结果，再整合输出回答）
4. 全链路标准执行流程
用户输入提问 → 接口接收
读取独立会话历史，自动执行Token预估、双层压缩裁剪
中控Dispatcher拆解任务，生成检索/计算任务列表
消息总线分发任务，多子Agent并行执行
收集所有工具/检索结果存入总线
ReAct反思判断信息是否充足，不足自动补充多轮检索
汇总全部结果交给大模型生成通顺完整回答
本轮问答持久存入会话内存，自动裁剪防止Token爆炸
5. 各模块核心能力简介
模块文件 核心功能
llm_client.py 异步并发LLM请求、SSE流式分片解析、强制结构化JSON输出、失败重试
rag_store.py 文本重叠分块、简易字符向量、余弦相似度检索、知识库批量入库
memory_store.py 多会话隔离、Token估算、滑动窗口裁剪、历史对话摘要压缩
tool_func.py 标准化计算器工具、Pydantic参数校验、运算异常捕获
multi_agent.py 消息总线、中控任务拆解、检索/计算子Agent、ReAct反思循环
main.py FastAPI服务、SSE流式接口、多智能体对话业务接口
三、今日开发难点 & 落地解决方案
难点1：整套底层组件全部重新实现，容易遗漏流式、JSON校验等细节
解决方案：模块化分步开发，写完一个文件单独测试运行，全部单元调试通过后再串联整体流程。
难点2：多Agent并行执行，任务结果错乱、工作流数据互相污染
解决方案：每个对话流程分配唯一flow_id，消息总线以flow_id为分区存储结果，不同对话数据完全隔离。
难点3：模型拆解任务时JSON格式错乱，调度器无法分发任务
解决方案：任务规划强制temperature=0消除随机性；Prompt严格约束输出格式；正则提取JSON片段；Pydantic模型二次校验，解析失败自动重试一次。
难点4：多轮长对话持续Token溢出，出现上下文截断、失忆
解决方案：双层压缩策略
第一层：滑动窗口，永久保留system人设，仅留存最近N轮原始对话
第二层：Token仍超限时，将早期历史生成精简摘要，替换原始长对话
难点5：ReAct无限循环重复调用工具，浪费API额度
解决方案：设置最大反思轮次（默认3轮），达到上限强制停止工具调用，基于已有信息汇总回答。
四、分文件完整项目代码
项目目录
day8_multi_agent/
├── llm_client.py # 异步LLM、SSE流式、结构化输出
├── rag_store.py # 文本分块、简易余弦向量RAG检索
├── memory_store.py # 分层会话记忆、双层压缩逻辑
├── tool_func.py # 计算器工具、参数校验
├── multi_agent.py # 消息总线、中控调度、ReAct、子Agent
└── main.py # FastAPI接口入口
依赖安装命令
pip install aiohttp pydantic fastapi uvicorn numpy
- llm_client.py
import asyncio
import aiohttp
import re
import json
from typing import List, Dict, AsyncGenerator
from pydantic import BaseModel
模型密钥配置
LLM_CONFIG = {
“qwen-turbo”: {
“base_url”: “https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions”,
“api_key”: “你的通义千问API Key”
}
}
结构化输出通用父类
class LLMBaseSchema(BaseModel):
pass
class AsyncLLM:
definit(self, model_name: str = “qwen-turbo”):
self.conf = LLM_CONFIG[model_name]
self.semaphore = asyncio.Semaphore(5)
self.timeout = aiohttp.ClientTimeout(total=60)
# SSE流式分片解析缓冲区 async def _stream_parse(self, resp) -> AsyncGenerator[str, None]: buffer = "" async for chunk in resp.content.iter_chunked(1024): buffer += chunk.decode("utf-8") while "data:" in buffer: idx = buffer.find("data:") end_idx = buffer.find("\n\n", idx) if end_idx == -1: break block = buffer[idx+5:end_idx].strip() buffer = buffer[end_idx+2:] if block == "[DONE]": return try: data = json.loads(block) delta = data["choices"][0]["delta"].get("content", "") if delta: yield delta except Exception: continue # 一次性同步完整返回 async def chat_sync(self, messages: List[Dict], temperature=0.1) -> str: payload = { "model": "qwen-turbo", "messages": messages, "temperature": temperature, "stream": False } headers = { "Authorization": f"Bearer {self.conf['api_key']}", "Content-Type": "application/json" } async with self.semaphore: async with aiohttp.ClientSession(timeout=self.timeout) as session: async with session.post(self.conf["base_url"], json=payload, headers=headers) as resp: res = await resp.json() return res["choices"][0]["message"]["content"] # 流式打字机输出 async def chat_stream(self, messages: List[Dict], temperature=0.1): payload = { "model": "qwen-turbo", "messages": messages, "temperature": temperature, "stream": True } headers = { "Authorization": f"Bearer {self.conf['api_key']}", "Content-Type": "application/json" } async with self.semaphore: async with aiohttp.ClientSession(timeout=self.timeout) as session: async with session.post(self.conf["base_url"], json=payload, headers=headers) as resp: async for text in self._stream_parse(resp): yield text # 强制返回标准JSON，带重试兜底 async def chat_struct(self, messages: List[Dict], schema: type[LLMBaseSchema]) -> LLMBaseSchema: extend_prompt = f"仅输出标准JSON，禁止任何解释、markdown，JSON规范：{schema.model_json_schema()}" new_msg = messages.copy() new_msg[-1]["content"] += extend_prompt raw = await self.chat_sync(new_msg, temperature=0.0) match = re.search(r"\{.*\}", raw, re.S) if not match: raw = await self.chat_sync(new_msg, 0.0) match = re.search(r"\{.*\}", raw, re.S) return schema.model_validate_json(match.group())全局单例客户端
llm_client = AsyncLLM()
2. rag_store.py
import numpy as np
from typing import List, Dict
文本重叠分片
def split_chunk(text: str, chunk_size=400, overlap=80) -> List[str]:
chunks = []
start = 0
text_len = len(text)
while start < text_len:
end = min(start + chunk_size, text_len)
chunks.append(text[start:end].strip())
start += chunk_size - overlap
return chunks
简易文本向量化
def text_to_vec(text: str) -> np.ndarray:
vec = np.array([ord© for c in text[:256]], dtype=np.float32)
vec = np.pad(vec, (0, 256 - len(vec)), mode=“constant”)
norm = np.linalg.norm(vec)
if norm > 0:
vec = vec / norm
return vec
余弦相似度计算
def cosine_similarity(v1: np.ndarray, v2: np.ndarray) -> float:
return float(np.dot(v1, v2))
简易内存RAG库
class SimpleRAGStore:
definit(self):
self.doc_meta: List[Dict] = []
self.vec_cache: List[np.ndarray] = []
# 批量入库文档 async def add_document(self, text: str, source="知识库"): chunks = split_chunk(text) for chunk in chunks: vec = text_to_vec(chunk) self.vec_cache.append(vec) self.doc_meta.append({"text": chunk, "source": source}) # 相似度检索，过滤低相关片段 async def search(self, query: str, top_k=3, threshold=0.6) -> List[Dict]: q_vec = text_to_vec(query) score_list = [] for idx, vec in enumerate(self.vec_cache): score = cosine_similarity(q_vec, vec) score_list.append((score, idx)) score_list.sort(reverse=True) result = [] for score, idx in score_list[:top_k]: if score >= threshold: result.append(self.doc_meta[idx]) return result全局向量库实例
rag_store = SimpleRAGStore()
预加载测试知识库
import asyncio
asyncio.run(rag_store.add_document(“多智能体使用中控调度拆分任务，RAG检索私有知识库，Function Calling执行数学工具计算”))
3. memory_store.py
from typing import List, Dict
from llm_client import llm_client
class ChatMemory:
definit(self):
self.session_map: Dict[str, List[Dict]] = {}
self.token_threshold = 1800
self.keep_raw_round = 3
# 简易Token估算 def estimate_token(self, msg_list: List[Dict]) -> int: total = 0 for msg in msg_list: total += len(msg.get("content", "")) * 2 return total # 滑动窗口裁剪，保留system与最近N轮对话 def slide_trim(self, msg_list: List[Dict]) -> List[Dict]: sys_msg = None other_msg = [] for m in msg_list: if m["role"] == "system": sys_msg = m else: other_msg.append(m) new_other = other_msg[-self.keep_raw_round:] if sys_msg: return [sys_msg] + new_other return new_other # 旧对话生成摘要压缩 async def compress_summary(self, msg_list: List[Dict]) -> List[Dict]: sys_msg = None history = [] for m in msg_list: if m["role"] == "system": sys_msg = m else: history.append(m) if len(history) <= self.keep_raw_round: return msg_list old_part = history[:-self.keep_raw_round] recent_part = history[-self.keep_raw_round:] old_text = "\n".join([f"{m['role']}:{m['content']}" for m in old_part]) sum_prompt = [{"role": "user", "content": f"精简对话摘要，保留数字与关键业务信息：{old_text}"}] summary = await llm_client.chat_sync(sum_prompt, temperature=0.0) new_msg = [] if sys_msg: new_msg.append(sys_msg) new_msg.append({"role": "system", "content": f"历史对话摘要：{summary}"}) new_msg.extend(recent_part) return new_msg # 自动双层压缩入口 async def auto_compress(self, msg_list: List[Dict]) -> List[Dict]: if self.estimate_token(msg_list) < self.token_threshold: return msg_list trim_res = self.slide_trim(msg_list) if self.estimate_token(trim_res) < self.token_threshold: return trim_res return await self.compress_summary(msg_list) # 读取会话历史 def load_history(self, session_id: str) -> List[Dict]: return self.session_map.get(session_id, []) # 追加单条对话 def append_msg(self, session_id: str, role: str, content: str): if session_id not in self.session_map: self.session_map[session_id] = [] self.session_map.append({"role": role, "content": content})全局内存会话实例
memory = ChatMemory()
4. tool_func.py
from pydantic import BaseModel, Field
计算器入参约束模型
class CalcParams(BaseModel):
num1: float = Field(description=“第一个运算数字”)
num2: float = Field(description=“第二个运算数字”)
op: str = Field(description=“运算符，仅支持 + - * /”)
异步计算工具
async def calculator_tool(params: CalcParams) -> str:
try:
match params.op:
case “+”:
res = params.num1 + params.num2
case “-”:
res = params.num1 - params.num2
case “*”:
res = params.num1 * params.num2
case “/”:
if params.num2 == 0:
return “工具错误：除数不能为0”
res = params.num1 / params.num2
case _:
return f"不支持运算符：{params.op}"
return f"计算结果：{params.num1}{params.op}{params.num2}={res}"
except Exception as e:
return f"计算异常：{str(e)}"
5. multi_agent.py
import re
from typing import List, Dict
from pydantic import BaseModel, Field
from llm_client import llm_client, LLMBaseSchema
from rag_store import rag_store
from memory_store import memory
from tool_func import calculator_tool, CalcParams
单任务结构
class TaskItem(BaseModel):
agent_name: str = Field(description=“子Agent名称 retrieve/calc”)
task_content: str = Field(description=“任务执行入参”)
任务列表Schema
class TaskPlan(LLMBaseSchema):
task_list: List[TaskItem]
消息总线：隔离工作流，存储各子Agent输出
class MessageBus:
definit(self):
self.flow_storage: Dict[str, Dict[str, str]] = {}
def save_result(self, flow_id: str, agent: str, content: str): if flow_id not in self.flow_storage: self.flow_storage[flow_id] = {} self.flow_storage[flow_id][agent] = content def get_all_results(self, flow_id: str) -> Dict[str, str]: return self.flow_storage.get(flow_id, {})bus = MessageBus()
知识库检索专用子Agent
class RetrieveAgent:
async def run(self, query: str) -> str:
docs = await rag_store.search(query)
if not docs:
return “知识库未查询到相关内容”
text = “\n”.join([item[“text”] for item in docs])
return f"【知识库检索】{text}"
数学计算专用子Agent
class CalcAgent:
async def run(self, expr: str) -> str:
prompt = [{“role”: “user”, “content”: f"提取算式数字与运算符，仅输出JSON{{num1,num2,op}}，算式：{expr}“}]
raw = await llm_client.chat_sync(prompt, temperature=0.0)
match = re.search(r”{.*}", raw, re.S)
params = CalcParams.model_validate_json(match.group())
return await calculator_tool(params)
ReAct反思判断：信息是否充足
async def reflect_judge(question: str, info: Dict[str, str]) -> bool:
info_text = “\n”.join([f"{k}:{v}" for k, v in info.items()])
prompt = f"根据已有信息判断是否足够回答用户问题，仅输出true/false。问题：{question} 已有资料：{info_text}"
res = await llm_client.chat_sync([{“role”: “user”, “content”: prompt}], 0.0)
return “true” in res.lower()
中控调度核心Agent
class DispatcherAgent:
# 拆解用户问题，生成任务清单
async def parse_task_plan(self, user_query: str) -> TaskPlan:
prompt_text = f"“”
拆解用户问题生成任务，仅输出JSON，task_list可选agent_name：retrieve/calc
用户问题：{user_query}
“”"
msg = [{“role”: “user”, “content”: prompt_text}]
return await llm_client.chat_struct(msg, TaskPlan)
# 完整工作流执行入口 async def run_workflow(self, session_id: str, user_query: str): flow_id = f"flow_{session_id}" # 读取历史并自动压缩 history = memory.load_history(session_id) base_msg = [{"role": "system", "content": "结合知识库与计算结果完整回答用户问题"}] + history base_msg.append({"role": "user", "content": user_query}) compress_msg = await memory.auto_compress(base_msg) # 第一步：并行执行所有拆解任务 plan = await self.parse_task_plan(user_query) retrieve_agent = RetrieveAgent() calc_agent = CalcAgent() task_coros = [] task_meta = [] for task in plan.task_list: task_meta.append(task) if task.agent_name == "retrieve": task_coros.append(retrieve_agent.run(task.task_content)) elif task.agent_name == "calc": task_coros.append(calc_agent.run(task.task_content)) task_outputs = await asyncio.gather(*task_coros) for idx, t in enumerate(task_meta): bus.save_result(flow_id, t.agent_name, task_outputs[idx]) # 第二步：ReAct反思循环，最多3轮补充检索 max_loop = 3 loop_cnt = 1 all_info = bus.get_all_results(flow_id) enough = await reflect_judge(user_query, all_info) while not enough and loop_cnt < max_loop: new_plan = await self.parse_task_plan(f"补充检索信息：{user_query}") new_coros = [] new_meta = [] for t in new_plan.task_list: if t.agent_name == "retrieve": new_meta.append(t) new_coros.append(retrieve_agent.run(t.task_content)) new_out = await asyncio.gather(*new_coros) for idx, t in enumerate(new_meta): bus.save_result(f"{flow}_loop{loop_cnt}", t.agent_name, new_out[idx]) all_info = bus.get_all_results(flow_id) enough = await reflect_judge(user_query, all_info) loop_cnt += 1 # 第三步：汇总全部信息生成最终回答 concat_info = "\n".join([f"{k}:{v}" for k, v in all_info.items()]) final_prompt = compress_msg + [{"role": "user", "content": f"结合下面资料完整回答：{concat_info}，用户问题：{user_query}"}] final_ans = await llm_client.chat_sync(final_prompt, temperature=0.1) # 保存本轮对话至内存 memory.append_msg(session_id, "user", user_query) memory.append_msg(session_id, "assistant", final_ans) return { "flow_id": flow_id, "task_info": all_info, "answer": final_ans }全局调度单例
dispatcher = DispatcherAgent()
6. main.py
from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse
from llm_client import llm_client
from multi_agent import dispatcher
app = FastAPI(title=“Day8 分文件多智能体全链路项目”)
SSE流式生成器
async def stream_generator(prompt: str):
yield “data: 开始生成\n\n”
async for chunk in llm_client.chat_stream([{“role”: “user”, “content”: prompt}]):
yield f"data: {chunk}\n\n"
yield “data: [DONE]\n\n”
基础流式对话接口
@app.get(“/chat/stream”)
async def stream_chat(prompt: str):
return StreamingResponse(stream_generator(prompt), media_type=“text/event-stream”)
完整多智能体业务接口
@app.get(“/agent/chat”)
async def agent_chat(
session_id: str = Query(…, description=“会话唯一标识”),
prompt: str = Query(…, description=“用户提问内容”)
):
return await dispatcher.run_workflow(session_id, prompt)
ifname== “main”:
import uvicorn
uvicorn.run(“main:app”, reload=True)
五、今日实操练习任务
新建独立文件夹，按目录创建6个py文件，填入对应代码，修改LLM配置中的API密钥
启动服务uvicorn main:app --reload，访问127.0.0.1:8000/docs
基础流式测试：调用/chat/stream，验证SSE打字机效果、分片解析无乱码
复合多智能体测试，接口/agent/chat，传入任意session_id，提问：计算(100+20)*6，介绍多智能体RAG架构
连续多轮同一session提问，验证滑动窗口+摘要双层压缩逻辑
修改rag_store知识库文本，验证检索结果同步更新
构造除零计算，测试工具异常捕获、错误信息正常返回
构造信息不足的提问，观察ReAct自动补充检索逻辑
六、配套高频面试题
基础问答
单一体Agent和多智能体分治架构优缺点、适用场景？
消息总线在多Agent系统中的作用是什么？flow_id设计意义？
并行任务、串行任务分别适用什么业务场景？
分层记忆两种压缩方案（滑动窗口/摘要）区别与取舍？
ReAct反思循环的实现逻辑，为什么要限制最大循环轮次？
工程实操题
如何保证模型稳定输出标准任务JSON，防止调度解析失败？
SSE流式分片出现截断、残缺如何解决？
多轮对话持续Token超限，双层压缩实现思路？
多Agent并行执行，如何统一收集、隔离各工作流结果？
拓展思考题
内存会话如何升级Redis分布式持久化，支持多服务实例？
简易内存向量库如何替换Chroma/Milvus实现海量持久知识库？
如何给多智能体增加并发限流、工具熔断、指数退避重试？
七、面试题标准答案
基础问答
单Agent开发简单、代码集中；缺点提示词臃肿、推理不稳定、维护困难，适合简单闲聊。多智能体各模块职责单一、推理稳定、可独立迭代扩容，适合多步骤工具+知识库复杂业务。
统一存储每个流程下所有子Agent执行结果；flow_id区分单次对话工作流，避免不同用户、不同轮次数据互相污染。
无依赖查询/计算使用并行，提升响应速度；前置结果作为输入的任务使用串行。
滑动窗口性能高，直接删除早期历史，但会丢失信息；摘要通过LLM压缩旧对话，保留关键数据，额外消耗一次模型调用。长对话采用双层兜底策略。
将当前全部检索、计算结果交给模型，结构化判断信息是否充足；限制最大轮次防止无限循环调用工具，节省API额度。
工程实操题
temperature设为0消除随机性；Prompt强制JSON规范；正则提取大括号内容；Pydantic模型校验，解析失败自动重新规划一次。
设置缓冲区拼接不完整分片，仅处理完整data:行；捕获单条JSON解析异常，不中断整条流式输出。
先滑动窗口保留system与最近N轮原始对话；Token仍超标时将早期历史生成摘要替换原始长文本，永久保留人设信息。
每条对话分配唯一flow_id，消息总线以flow_id作为key分区存储，调度完成后统一读取对应分区结果。
拓展思考题
将内存字典替换aioredis，session_id作为Redis键，列表存储对话消息，设置过期自动清理会话。
上层检索接口不变，底层向量计算、入库逻辑替换为Chroma持久化集合，做到业务代码无感知切换。
全局异步信号量控制最大并行任务数；工具连续多次失败触发熔断；捕获429限流异常，实现指数退避重试。
八、学习总结
Day8完整搭建分层多智能体项目，所有底层组件从零独立开发，无任何前序代码依赖，模块化拆分清晰，适合分步练习复盘。
