今日总学习目标全天8h分配时长拆分理论学习2.5h分文件代码编写、单元调试4h面试题复盘、流程梳理背诵1.5h学习目标清晰认知单一体Agent的短板掌握中控调度垂直子Agent分治设计思想独立分层实现全套底层基础设施每模块解耦、可单独复用实现消息总线完成多Agent并行/串行任务调度、结果统一存储搭建完整业务链路会话记忆→RAG知识库检索→工具函数调用→ReAct反思校验→多智能体汇总输出掌握工程分层规范学会拆分项目文件提升代码可维护性二、核心理论教学笔记单一体Agent的致命缺陷提示词臃肿计算、检索、规划、记忆全部塞入同一个系统prompt模型极易混淆能力边界推理不稳定复杂多步骤任务容易漏执行步骤、选错工具、大量生成幻觉上下文持续膨胀无分层隔离长对话快速Token超限出现对话失忆维护成本极高新增工具/知识库需要大面积修改主Agent代码耦合严重多智能体分层架构标准工业分层四层完整结构自上而下接口层main.pyFastAPI提供SSE流式、多智能体对话接口接收用户请求调度总线层multi_agent.py消息总线统一存储各子Agent执行结果通过flow_id隔离单次工作流Dispatcher中控调度Agent解析用户需求自动拆分任务清单ReAct反思模块循环校验当前信息是否充足不足自动补充检索/计算垂直子AgentRetrieve检索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格式错乱调度器无法分发任务解决方案任务规划强制temperature0消除随机性Prompt严格约束输出格式正则提取JSON片段Pydantic模型二次校验解析失败自动重试一次。难点4多轮长对话持续Token溢出出现上下文截断、失忆解决方案双层压缩策略第一层滑动窗口永久保留system人设仅留存最近N轮原始对话第二层Token仍超限时将早期历史生成精简摘要替换原始长对话难点5ReAct无限循环重复调用工具浪费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 numpyllm_client.pyimport asyncioimport aiohttpimport reimport jsonfrom typing import List, Dict, AsyncGeneratorfrom 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):passclass AsyncLLM:definit(self, model_name: str “qwen-turbo”):self.conf LLM_CONFIG[model_name]self.semaphore asyncio.Semaphore(5)self.timeout aiohttp.ClientTimeout(total60)# 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[idx5:end_idx].strip() buffer buffer[end_idx2:] 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], temperature0.1) - str: payload { model: qwen-turbo, messages: messages, temperature: temperature, stream: False } headers { Authorization: fBearer {self.conf[api_key]}, Content-Type: application/json } async with self.semaphore: async with aiohttp.ClientSession(timeoutself.timeout) as session: async with session.post(self.conf[base_url], jsonpayload, headersheaders) as resp: res await resp.json() return res[choices][0][message][content] # 流式打字机输出 async def chat_stream(self, messages: List[Dict], temperature0.1): payload { model: qwen-turbo, messages: messages, temperature: temperature, stream: True } headers { Authorization: fBearer {self.conf[api_key]}, Content-Type: application/json } async with self.semaphore: async with aiohttp.ClientSession(timeoutself.timeout) as session: async with session.post(self.conf[base_url], jsonpayload, headersheaders) 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禁止任何解释、markdownJSON规范{schema.model_json_schema()} new_msg messages.copy() new_msg[-1][content] extend_prompt raw await self.chat_sync(new_msg, temperature0.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()