尧图网站建设 
最近在给产品做一些“智能辅助”能力时我一直在思考一个问题单一的大模型对话接口到底能不能真正满足产品侧的复杂需求比如产品经理扔进来一段模糊的反馈系统既要判断这是功能建议还是 Bug 报告又要给出影响面分析还要自动生成测试用例。如果只靠一个 Prompt 把所有事情都塞给大模型效果往往不稳定而且一旦需求拆解粒度变粗后续维护成本也很高。后来我换了一个思路不再追求“一个大模型回答所有问题”而是把不同职责拆成多个智能节点每个节点负责一个专业方向节点之间通过统一的服务总线协作最终汇总成一个完整的结论。这套机制非常像蜂群——单个工蜂只做一件简单的事但整个蜂群能完成非常复杂的筑巢、采蜜、防御任务。放在产品工程里这个设计就是“The hive mind for your product”为产品构建一个多智能体协作体系让不同 agent 各司其职互相配合最终形成比单一大模型更强的整体智能。这篇文章会从概念讲起然后带大家设计一个可落地的产品智能体架构最后给出完整的可运行工程示例。无论你是后端工程师、全栈开发者还是对 AI Agent 应用感兴趣的产品技术负责人这篇文章都能提供一套可以照着实现的闭环方案。1. 什么是“The hive mind for your product”先解释一下标题里的核心词 Hive Mind。Hive Mind 直译是“蜂群思维”它描述的是多个独立个体通过信息共享、分工协作从而表现出远超单体智慧的群体智能。在社会性昆虫中单个蚂蚁或者蜜蜂的决策能力极其有限但是整个群体却可以通过简单的规则涌现出非常复杂的集体行为。放在产品工程语境下“The hive mind for your product” 可以理解成为你的产品构建一组 AI Agent每个 Agent 只负责特定领域比如需求分析、用户反馈分类、测试用例生成、知识库检索然后由一个协调中枢统一调度让这些 Agent 并行工作、互相校验最终输出产品团队可以直接使用的结果。1.1 为什么产品需要“蜂群思维”传统的 AI 功能大多是一个“问答式”闭环。用户输入问题模型输出答案。在产品内部这种方式的局限性很明显单一上下文很难覆盖复杂的业务场景。一个大 Prompt 里堆叠多个任务模型容易“偏科”。任务链路不透明产品团队无法定位问题出在哪个环节。扩展性差每新增一个能力都要改原来的 Prompt牵一发动全身。蜂群思维的多智能体架构则把一个大任务拆解成多个小任务每个小任务由专门的 Agent 处理。这样做的好处是任务边界清晰每个 Agent 的输入输出可预期。可以并行处理多个子任务提升整体响应速度。单个 Agent 升级不影响其他 Agent。每个 Agent 都能被独立测试、独立观测、独立运维。1.2 与 RAG、Prompt Engineering 的关系这里顺便做几个概念区分方便新手理解。Prompt Engineering 解决的是“如何让模型更好地理解需求”。RAG检索增强生成解决的是“如何让模型获得外部知识”。Multi-Agent System 解决的是“如何把任务合理拆解并组织多个模型协同工作”。三者不是替代关系而是互补关系。实际工程中往往先用 RAG 给 Agent 提供知识再用 Prompt Engineering 限定输出格式最后用多智能体架构把它们组织成完整的产品能力。1.3 典型应用场景“The hive mind for your product” 能落地的场景非常多下面列几个比较有代表性的方向场景说明涉及 Agent需求分析根据原始需求描述提取功能点、优先级和依赖关系需求解析 Agent、依赖分析 Agent用户反馈处理对客服工单、应用商店评论、社群反馈进行分类和情感分析分类 Agent、情感分析 Agent、聚合 Agent自动测试根据需求文档生成接口测试用例和边界条件需求 Agent、测试生成 Agent知识库问答结合企业文档做多轮问答和溯源检索 Agent、问答 Agent、引用核对 Agent缺陷分诊根据 Bug 描述判断模块归属、严重级别和推荐处理人文本分析 Agent、路由 Agent看到这里你应该已经感受到多智能体体系的核心并不是“多个模型”而是一套工程化组织方式。下面我们进入架构设计部分。2. 整体架构设计在动手编码之前架构设计很重要。没有一个清晰的架构Agent 一多就容易变成“蜘蛛网”互相调用混乱后期没法维护。2.1 分层架构我推荐把整个系统拆成四层输入层 → 协调层 → Agent 执行层 → 存储与输出层输入层输入层负责接收来自不同渠道的请求比如 Web 表单、API 调用、消息队列消息等。输入层不直接调用 Agent而是把原始数据标准化包装成统一的内部消息结构。协调层协调层是整个蜂群大脑。它负责任务拆解把用户请求拆成多个子任务。Agent 路由根据子任务类型找到合适的 Agent。执行编排决定多个 Agent 是串行执行还是并行执行。结果聚合收集所有 Agent 的返回结果做去重、合并、冲突仲裁。Agent 执行层执行层是具体的“工蜂”。每个 Agent 是一个独立的功能单元内部可以调用大模型、查询数据库、调用外部 API或者执行 Python 代码。对外只暴露统一的接口。存储与输出层存储层保存任务记录、Agent 执行结果、日志和缓存。输出层把最终结果格式化成用户需要的结构比如 JSON、Markdown 文档或者前端可渲染的卡片。2.2 任务拆解与编排策略在实际项目中任务拆解有两种主要方式固定流程拆解适合流程比较稳定的业务。比如需求分析一定是先解析需求再提取功能点最后生成优先级建议。这种拆解方式通过代码写死编排逻辑稳定性高容易调试。动态规划拆解适合比较开放的场景比如“帮我看看这个产品的体验问题”系统需要模型自主决定要调用哪些 Agent。动态拆解灵活但可控性差往往需要引入模型自省和人工审核。对于大多数产品团队我建议先做固定流程拆解跑通之后再考虑动态规划。先稳定再智能。2.3 Agent 通信机制Agent 之间怎么通信我见过几种方案直接 HTTP 调用适合微服务化部署但链路复杂时不好排查。消息队列适合异步任务和削峰但会增加运维复杂度。共享数据库每个 Agent 从数据库读取任务、写入结果简单可靠适合中小型系统。进程内调度所有 Agent 在同一个进程中运行通过函数调用传递数据最简单适合本文这种教学工程。本文的示例会采用“进程内调度 共享数据库”的方式既能讲清楚完整链路又不需要引入额外中间件读者复制到本地就能跑起来。3. 环境准备与技术选型3.1 技术栈说明为了让大家能快速复现本文选用一套轻量级技术栈语言Python 3.9Web 框架FastAPI提供 REST API数据库SQLite保存任务和结果大模型接入使用 OpenAI 兼容接口代码中通过抽象层隔离不直接绑定具体厂商测试工具pytest用于验证核心逻辑版本说明一下Python 建议使用 3.9 以上版本FastAPI 使用当前主流版本即可。由于大模型 SDK 更新较快本文不会把某个 SDK 的版本写死而是提供一个 Mock 模型实现方便没有 API Key 的读者也能跑通整个流程。3.2 项目结构规划建议按下面的目录结构组织项目hive-mind/ ├── main.py # FastAPI 入口 ├── config.py # 配置管理 ├── coordinator.py # 协调器负责任务拆解和结果聚合 ├── agents/ │ ├── __init__.py │ ├── base.py # Agent 基类 │ ├── requirement_agent.py# 需求解析 Agent │ ├── feedback_agent.py # 用户反馈分类 Agent │ ├── testcase_agent.py # 测试用例生成 Agent │ └── mock_llm.py # 大模型调用抽象含 Mock 实现 ├── models/ │ ├── __init__.py │ └── schema.py # 数据模型 ├── data_store.py # SQLite 存储层 ├── requirements.txt └── README.md3.3 为什么用这种选型选择 FastAPI 是因为它支持异步同时自带 OpenAPI 文档方便调试。SQLite 则是零配置文件型数据库适合做原型后续换成 MySQL 或者 PostgreSQL 只需要改存储层实现。最重要的设计决策是抽象出 mock_llm这样即使没有大模型 API Key也可以先开发完整个业务流程之后无缝切换到真实模型。下面进入核心代码实现。4. 核心实现构建你的产品蜂群这一节是全文重点我会按模块逐步讲解。4.1 定义数据结构首先定义整个系统的核心数据结构。我们把任务、Agent 结果、最终聚合结果都抽象成模型。创建文件models/schema.py 数据模型定义 from enum import Enum from typing import List, Optional from pydantic import BaseModel, Field class TaskType(str, Enum): 任务类型枚举 REQUIREMENT_ANALYSIS requirement_analysis FEEDBACK_CLASSIFICATION feedback_classification TESTCASE_GENERATION testcase_generation class AgentResult(BaseModel): 单个 Agent 的执行结果 agent_name: str Field(descriptionAgent 名称) task_type: TaskType Field(description任务类型) status: str Field(defaultsuccess, description执行状态) output: dict Field(descriptionAgent 输出的结构化数据) elapsed_ms: int Field(description耗时(毫秒)) class TaskRequest(BaseModel): 外部传入的任务请求 task_type: TaskType Field(description任务类型) raw_input: str Field(description原始输入文本) options: dict Field(default_factorydict, description附加参数) class TaskResponse(BaseModel): 最终聚合响应 task_id: str Field(description任务 ID) status: str Field(description任务状态) results: List[AgentResult] Field(description所有 Agent 的执行结果) final_output: dict Field(description聚合后的最终输出)说明几个设计要点TaskType使用枚举类型避免字符串满天飞后续新增任务类型只需要扩展枚举。AgentResult里面记录了agent_name和elapsed_ms方便做链路追踪和性能分析。每个 Agent 的output是字典类型可以灵活承载结构化输出不会因为字段变化导致模型类频繁修改。4.2 大模型调用抽象为了避免把代码写死在某一家大模型提供商上我们做一个抽象层。这样后续接 GPT、通义千问、文心或者本地模型只需要替换这一个模块。创建文件agents/mock_llm.py 大模型调用抽象层 在未配置真实 API 时使用 Mock 实现方便本地开发测试 import os import time from typing import List, Dict, Any class BaseLLM: LLM 抽象基类 def complete(self, system_prompt: str, user_prompt: str) - str: raise NotImplementedError class MockLLM(BaseLLM): Mock 实现返回固定格式的数据 方便没有 API Key 的读者运行完整流程 def complete(self, system_prompt: str, user_prompt: str) - str: time.sleep(0.3) # 模拟模型推理耗时 if 需求 in user_prompt or requirement in user_prompt.lower(): return { functional_points: [用户登录, 订单查询, 订单支付], priority: high, dependencies: [用户模块, 订单模块] } if 反馈 in user_prompt or feedback in user_prompt.lower(): return { category: bug, sentiment: negative, module: 支付模块, suggestion: 建议优先修复支付超时问题 } if 测试 in user_prompt or test in user_prompt.lower(): return { test_cases: [ {title: 正常登录, steps: 输入正确账号密码, expected: 登录成功}, {title: 密码错误, steps: 输入错误密码, expected: 提示密码错误}, {title: 账号不存在, steps: 输入未注册账号, expected: 提示账号不存在} ] } return {result: unknown task} class OpenAIConpatLLM(BaseLLM): OpenAI 兼容接口实现 通过环境变量配置 base_url 和 api_key 依赖 openai 库需要 pip install openai def __init__(self): # 这里只做说明实际使用时需要安装 openai 并配置客户端 pass def complete(self, system_prompt: str, user_prompt: str) - str: raise NotImplementedError(真实模型接入请配置 openai 库) def get_llm() - BaseLLM: 工厂方法根据环境变量返回对应的 LLM 实例 if os.getenv(USE_MOCK_LLM, true).lower() true: return MockLLM() return OpenAIConpatLLM()这里的设计意图是应用启动时通过get_llm()决定使用哪个模型实现。默认走 Mock新手可以零成本跑通全流程。有真实 API Key 后继承BaseLLM实现complete方法即可业务代码不用动。需要说明的是OpenAIConpatLLM我故意没有写死具体的调用参数因为各家兼容接口的细节差异比较大。你如果要用真实模型只需要补全__init__和complete方法即可网上有大量参考。4.3 Agent 基类和具体实现先定义 Agent 基类后续所有 Agent 都继承这个基类。创建文件agents/base.py Agent 基类定义 from abc import ABC, abstractmethod from typing import Dict, Any from models.schema import AgentResult, TaskType from agents.mock_llm import BaseLLM class BaseAgent(ABC): 所有 Agent 的抽象基类 def __init__(self, llm: BaseLLM): self.llm llm self.name self.__class__.__name__ abstractmethod def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: 处理输入并返回结构化输出 pass def execute(self, task_type: TaskType, raw_input: str, options: Dict[str, Any]) - AgentResult: 执行入口封装耗时统计和异常处理 子类不需要重复写 try-except import time start time.time() try: output self.process(raw_input, options) status success except Exception as e: output {error: str(e)} status failed elapsed_ms int((time.time() - start) * 1000) return AgentResult( agent_nameself.name, task_typetask_type, statusstatus, outputoutput, elapsed_mselapsed_ms )基类中做了一层封装把异常捕获和耗时统计从业务代码中剥离出来。这样每个具体 Agent 只需要关心“怎么处理输入”不需要关心“怎么上报状态”代码会干净很多。接下来创建需求解析 Agent。创建文件agents/requirement_agent.py 需求解析 Agent 职责从原始需求描述中提取功能点、优先级和依赖关系 import json from typing import Dict, Any from agents.base import BaseAgent class RequirementAgent(BaseAgent): 需求解析 Agent def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: system_prompt 你是一名资深产品经理请从用户需求中提取结构化信息。 输出 JSON包含三个字段 functional_points: 功能点列表 priority: 优先级可选 high/medium/low dependencies: 依赖的模块列表 user_prompt f用户需求如下{raw_input} response self.llm.complete(system_prompt, user_prompt) try: result json.loads(response) except json.JSONDecodeError: result { functional_points: [], priority: medium, dependencies: [], parse_error: 模型输出无法解析为 JSON } result[original_input] raw_input return result这里有个细节解析模型输出时一定要做 JSON 解析异常兜底不能假设模型每次都返回合法 JSON。返回值带上original_input也是必要的方便追溯数据来源。创建文件agents/feedback_agent.py 用户反馈分类 Agent 职责对用户反馈进行分类、情感分析并给出模块归属建议 import json from typing import Dict, Any from agents.base import BaseAgent class FeedbackAgent(BaseAgent): 用户反馈分类 Agent def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: system_prompt 你是一名用户研究工程师请对用户反馈进行分类分析。 输出 JSON包含四个字段 category: 反馈类型可选 feature_request / bug / complaint / praise sentiment: 情感倾向可选 positive / neutral / negative module: 涉及的产品模块 suggestion: 一句话处理建议 user_prompt f用户反馈如下{raw_input} response self.llm.complete(system_prompt, user_prompt) try: result json.loads(response) except json.JSONDecodeError: result { category: unknown, sentiment: neutral, module: unknown, suggestion: 无法自动解析建议人工处理 } result[original_input] raw_input return result创建文件agents/testcase_agent.py 测试用例生成 Agent 职责根据需求描述生成基础测试用例 import json from typing import Dict, Any from agents.base import BaseAgent class TestcaseAgent(BaseAgent): 测试用例生成 Agent def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: system_prompt 你是一名测试开发工程师请根据需求生成测试用例。 输出 JSON包含一个字段 test_cases: 测试用例列表每个用例包含 title/steps/expected user_prompt f需求描述如下{raw_input} response self.llm.complete(system_prompt, user_prompt) try: result json.loads(response) except json.JSONDecodeError: result { test_cases: [], parse_error: 模型输出无法解析为 JSON } result[original_input] raw_input return result4.4 协调器设计协调器是整个蜂群的核心负责把任务分发给对应的 Agent并对结果做聚合。创建文件coordinator.py 协调器负责任务路由、Agent 调度和结果聚合 from typing import Dict, List, Any from uuid import uuid4 from models.schema import TaskRequest, TaskResponse, AgentResult, TaskType from agents.requirement_agent import RequirementAgent from agents.feedback_agent import FeedbackAgent from agents.testcase_agent import TestcaseAgent from agents.mock_llm import BaseLLM class Coordinator: 任务协调器 def __init__(self, llm: BaseLLM): self.llm llm self.registry { TaskType.REQUIREMENT_ANALYSIS: RequirementAgent(llm), TaskType.FEEDBACK_CLASSIFICATION: FeedbackAgent(llm), TaskType.TESTCASE_GENERATION: TestcaseAgent(llm), } def dispatch(self, request: TaskRequest) - TaskResponse: 派发任务并聚合结果 简单场景单 Agent 执行 复杂场景根据 options 决定是否联动其他 Agent task_id uuid4().hex[:12] agent self.registry.get(request.task_type) if not agent: return TaskResponse( task_idtask_id, statusfailed, results[], final_output{error: fUnsupported task type: {request.task_type}} ) # 执行主任务 Agent main_result agent.execute(request.task_type, request.raw_input, request.options) # 聚合逻辑如果主任务成功尝试获取补充分析 final_output main_result.output.copy() if request.task_type TaskType.REQUIREMENT_ANALYSIS: # 需求分析联动测试用例生成 testcase_agent self.registry[TaskType.TESTCASE_GENERATION] test_result testcase_agent.execute( TaskType.TESTCASE_GENERATION, request.raw_input, request.options ) final_output[test_cases_suggestion] test_result.output.get(test_cases, []) return TaskResponse( task_idtask_id, statussuccess if main_result.status success else partial, results[main_result, test_result], final_outputfinal_output ) if request.task_type TaskType.FEEDBACK_CLASSIFICATION: # 反馈分类联动处理建议 # 实际项目中可以在这里接入工单系统或通知服务 final_output[handling_note] 建议在 24 小时内人工复核该反馈 return TaskResponse( task_idtask_id, statusmain_result.status, results[main_result], final_outputfinal_output ) return TaskResponse( task_idtask_id, statusmain_result.status, results[main_result], final_outputfinal_output )这段代码的核心是dispatch方法。当前实现里需求分析任务会自动触发测试用例生成 Agent这就体现了“蜂群”的协作特性一个 Agent 的输出可以作为另一个 Agent 的补充而不是所有逻辑都堆在一个函数里。在实际项目中你可以把这种联动关系配置在数据库里而不是写在代码里比如维护一张agent_linkage表这样新增联动关系就不用改代码了。4.5 存储层实现为了让任务结果可追溯我们需要把每次请求保存到数据库。创建文件data_store.py SQLite 存储层 import sqlite3 import json from typing import Dict, Any class DataStore: 基于 SQLite 的简单存储实现 def __init__(self, db_path: str hive_mind.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self._create_table() def _create_table(self): cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS task_records ( task_id TEXT PRIMARY KEY, task_type TEXT, raw_input TEXT, result TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def save_task(self, task_id: str, task_type: str, raw_input: str, result: Dict[str, Any]): 保存任务记录 cursor self.conn.cursor() cursor.execute( INSERT OR REPLACE INTO task_records (task_id, task_type, raw_input, result) VALUES (?, ?, ?, ?), (task_id, task_type, raw_input, json.dumps(result, ensure_asciiFalse)) ) self.conn.commit() def get_task(self, task_id: str) - Dict[str, Any] | None: 查询任务记录 cursor self.conn.cursor() cursor.execute( SELECT task_id, task_type, raw_input, result FROM task_records WHERE task_id ?, (task_id,) ) row cursor.fetchone() if not row: return None return { task_id: row[0], task_type: row[1], raw_input: row[2], result: json.loads(row[3]) } def close(self): self.conn.close()SQLite 的check_same_threadFalse在 FastAPI 多线程场景下能避免不必要的报错不过这是简单写法。生产环境建议改用连接池或者直接用 SQLAlchemy。4.6 FastAPI 入口最后是 FastAPI 入口文件把前面所有模块串起来。创建文件main.py FastAPI 入口 from fastapi import FastAPI from models.schema import TaskRequest, TaskResponse from coordinator import Coordinator from agents.mock_llm import get_llm from data_store import DataStore app FastAPI( titleHive Mind API, descriptionThe hive mind for your product - 多智能体产品协作系统, version0.1.0 ) # 初始化协调器和存储层 llm get_llm() coordinator Coordinator(llm) data_store DataStore() app.get(/health) def health(): 健康检查 return {status: ok} app.post(/api/v1/task, response_modelTaskResponse) def submit_task(request: TaskRequest): 提交任务 response coordinator.dispatch(request) data_store.save_task( task_idresponse.task_id, task_typerequest.task_type.value, raw_inputrequest.raw_input, resultresponse.model_dump() ) return response app.get(/api/v1/task/{task_id}) def get_task(task_id: str): 查询任务记录 record data_store.get_task(task_id) if not record: return {error: task not found} return record app.on_event(shutdown) def shutdown(): data_store.close()有一个地方需要提一下response.model_dump()是 Pydantic v2 的写法如果你用的是 Pydantic v1需要改成response.dict()。FastAPI 新版默认使用 Pydantic v2但如果你的环境比较旧要注意这个兼容性问题。4.7 依赖文件与启动命令创建文件requirements.txtfastapi uvicorn pydantic注意openai库只有在接入真实大模型时才需要Mock 模式下不需要安装。启动服务cd hive-mind pip install -r requirements.txt uvicorn main:app --reload --port 8000看到日志输出Uvicorn running on http://127.0.0.1:8000就说明服务启动成功了。5. 运行与验证5.1 测试需求分析任务打开终端执行下面的 curl 命令提交一个需求分析任务curl -X POST http://127.0.0.1:8000/api/v1/task \ -H Content-Type: application/json \ -d { task_type: requirement_analysis, raw_input: 我们希望在 App 内增加一个订单管理功能用户可以查看历史订单、取消未发货订单、申请售后。 }预期返回结果大致如下{ task_id: a1b2c3d4e5f6, status: success, results: [ { agent_name: RequirementAgent, task_type: requirement_analysis, status: success, output: { functional_points: [用户登录, 订单查询, 订单支付], priority: high, dependencies: [用户模块, 订单模块], original_input: 我们希望在 App 内增加... }, elapsed_ms: 310 }, { agent_name: TestcaseAgent, task_type: testcase_generation, status: success, output: { test_cases: [ {title: 正常登录, steps: 输入正确账号密码, expected: 登录成功}, {title: 密码错误, steps: 输入错误密码, expected: 提示密码错误} ], original_input: 我们希望在 App 内增加... }, elapsed_ms: 305 } ], final_output: { functional_points: [用户登录, 订单查询, 订单支付], priority: high, dependencies: [用户模块, 订单模块], test_cases_suggestion: [ {title: 正常登录, steps: 输入正确账号密码, expected: 登录成功}, {title: 密码错误, steps: 输入错误密码, expected: 提示密码错误} ] } }这个返回结果里可以清楚看到需求分析任务自动触发了测试用例生成 Agent最终返回的final_output同时包含需求解析内容和测试用例建议。这就是蜂群协作的效果。5.2 测试用户反馈分类任务再测试一个用户反馈分类场景curl -X POST http://127.0.0.1:8000/api/v1/task \ -H Content-Type: application/json \ -d { task_type: feedback_classification, raw_input: 最近在支付页面经常超时用户反馈支付失败客户投诉率大幅上升需要尽快处理 }返回结果的关键部分如下{ final_output: { category: bug, sentiment: negative, module: 支付模块, suggestion: 建议优先修复支付超时问题, original_input: 最近在支付页面经常超时..., handling_note: 建议在 24 小时内人工复核该反馈 } }系统能识别出这是 Bug、情感为负面、涉及支付模块并给出处理建议和人工复核提醒。这套能力放到真实产品里可以对接工单系统实现自动分诊。5.3 查询历史任务每次提交的任务都会持久化到 SQLite可以通过下面的接口查询curl http://127.0.0.1:8000/api/v1/task/a1b2c3d4e5f6返回历史任务的完整记录包括原始输入和处理结果。这为后续做数据分析、模型效果评估保留了真实的样本数据。6. 常见问题与排查思路在开发和运行这套系统的过程中比较容易遇到下面几类问题。问题现象常见原因解决思路启动报错ModuleNotFoundError: No module named fastapi未安装依赖执行pip install -r requirements.txt启动报错Pydantic model_dump not foundPydantic 版本过旧升级到 Pydantic v2或改用.dict()需求分析结果为空Mock 模型未命中关键字检查agent输入文本是否包含“需求”等触发词返回 JSON 解析失败大模型输出非法 JSON在 Agent 中增加解析兜底逻辑并对结果做人工审核多线程访问 SQLite 报错默认连接不支持多线程初始化时设置check_same_threadFalse真实模型接不上base_url 或 api_key 配置错误先确认网络连通性再检查认证参数任务执行超时大模型推理耗时过长为每个 Agent 设置超时时间超时走降级策略6.1 排查清单如果你提交任务后返回了异常结果按下面顺序排查先看results中每个 Agent 的status字段定位是哪个 Agent 失败了。再看output中的error信息确认是模型解析问题还是逻辑异常。查看服务端控制台日志确认请求是否到达 Agent。如果使用真实模型检查 API 调用耗时和 token 消耗确认是否触发限流。检查 SQLite 是否有历史记录确认存储层是否正常持久化。核心原则是先定位是“模型问题”还是“工程问题”再对症下药。很多新手一看到错误就怀疑大模型实际上很多问题是代码边界条件没处理好。7. 工程化最佳实践把这套系统从 Demo 推向生产环境有几个关键点值得注意7.1 Agent 粒度设计Agent 不是越细越好。粒度过粗每个 Agent 承担太多职责又退化成“一个大 Prompt 解决所有问题”。粒度过细Agent 之间通信成本增加链路变长维护困难。我的建议是遵循“单一职责原则”一个 Agent 只做一件可命名的事情。如果这个 Agent 的 Prompt 已经超过几百字考虑拆分。如果多个 Agent 经常一起被调用考虑合并成一个流程型 Agent。7.2 上下文管理与记忆真实业务中Agent 之间往往需要共享信息。比如需求分析 Agent 识别出“订单模块”测试用例 Agent 需要知道这个信息来生成更有针对性的用例。目前示例中每个 Agent 都独立处理原始输入没有共享上下文。在实际工程中建议引入“上下文总线”机制每个 Agent 执行前从总线拉取需要的信息。执行后把输出写入总线。协调器负责控制总线的读写权限避免数据污染。7.3 可观测性Agent 系统最大的问题就是不可控。要让系统可维护必须有完整的可观测性方案每个 Agent 记录输入摘要、输出摘要、耗时、token 消耗。使用链路追踪 ID 串起整个调用链。对模型输出做版本快照方便回放和对比效果。核心场景建立自动化测试模型升级后自动回归验证。7.4 安全与权限边界多智能体系统如果接入企业管理后台必须考虑权限边界不同 Agent 使用独立的 API Key避免一个 Key 泄露导致全部接口暴露。对 Agent 可访问的数据做最小权限控制比如财务数据只有财务 Agent 能读取。外部输入要经过过滤和脱敏防止提示注入攻击。涉及删除、修改类操作前必须有人工审批环节。7.5 模型降级策略大模型服务不可用时系统不能直接崩溃。建议设计降级链路一级降级从高精度大模型切换到轻量模型。二级降级返回规则匹配的兜底答案。三级降级返回人工处理提示并创建工单通知相关负责人。示例代码中的MockLLM实际上就是一种降级实现保证本地无网络也能跑通流程。7.6 性能优化思路当前框架是同步执行多个 Agent 串行耗时累加。对于响应时间要求高的场景可以改成并发执行使用 FastAPI 的异步接口。通过asyncio.gather并行调用多个 Agent。对耗时长的 Agent 做缓存相同输入直接返回历史结果。引入消息队列把耗时任务异步化前端先返回“处理中”处理完成再通过 WebSocket 推送结果。8. 总结与下一步学习路线这篇文章围绕“The hive mind for your product”这个主题从概念到落地实现了一个多智能体产品协作系统。核心内容包括什么是蜂群思维以及它在产品工程中的含义。一个四层架构输入层、协调层、Agent 执行层、存储与输出层。完整可运行的多 Agent 工程包含需求解析 Agent、用户反馈分类 Agent、测试用例生成 Agent。协调器如何做任务路由、联动编排和结果聚合。基于 SQLite 的任务持久化方案。生产环境的多项关键设计可观测性、安全边界、降级策略、性能优化。如果你想继续深入我建议按下面路径学习先跑通本文示例理解每个模块的作用。为某个 Agent 接入真实大模型感受 Mock 与真实模型的差异。把固定流程拆解改成配置化让调度逻辑可以通过页面或接口动态调整。引入向量数据库给 Agent 增加长期记忆能力实现更复杂的业务场景。研究 LangGraph、AutoGen 等成熟的多智能体框架对比它们的编排思路与本文手写方案的异同。最后给一个实操建议不要一开始就追求复杂的动态规划先把一个垂直场景做扎实。你可以从“用户反馈自动分诊”这个场景开始用三个 Agent 跑通全链路再逐步扩展到需求、测试、知识库等方向。蜂群的力量来自分工与协作而你负责做的是给每只“工蜂”划清边界、定好规则。这套系统上手之后你会明显感受到单一的问答式 AI 交互和真正的产品级智能体体系之间差距不是一点半点。 多智能体协作架构实战：构建产品的蜂群思维系统
最近在给产品做一些“智能辅助”能力时我一直在思考一个问题单一的大模型对话接口到底能不能真正满足产品侧的复杂需求比如产品经理扔进来一段模糊的反馈系统既要判断这是功能建议还是 Bug 报告又要给出影响面分析还要自动生成测试用例。如果只靠一个 Prompt 把所有事情都塞给大模型效果往往不稳定而且一旦需求拆解粒度变粗后续维护成本也很高。后来我换了一个思路不再追求“一个大模型回答所有问题”而是把不同职责拆成多个智能节点每个节点负责一个专业方向节点之间通过统一的服务总线协作最终汇总成一个完整的结论。这套机制非常像蜂群——单个工蜂只做一件简单的事但整个蜂群能完成非常复杂的筑巢、采蜜、防御任务。放在产品工程里这个设计就是“The hive mind for your product”为产品构建一个多智能体协作体系让不同 agent 各司其职互相配合最终形成比单一大模型更强的整体智能。这篇文章会从概念讲起然后带大家设计一个可落地的产品智能体架构最后给出完整的可运行工程示例。无论你是后端工程师、全栈开发者还是对 AI Agent 应用感兴趣的产品技术负责人这篇文章都能提供一套可以照着实现的闭环方案。1. 什么是“The hive mind for your product”先解释一下标题里的核心词 Hive Mind。Hive Mind 直译是“蜂群思维”它描述的是多个独立个体通过信息共享、分工协作从而表现出远超单体智慧的群体智能。在社会性昆虫中单个蚂蚁或者蜜蜂的决策能力极其有限但是整个群体却可以通过简单的规则涌现出非常复杂的集体行为。放在产品工程语境下“The hive mind for your product” 可以理解成为你的产品构建一组 AI Agent每个 Agent 只负责特定领域比如需求分析、用户反馈分类、测试用例生成、知识库检索然后由一个协调中枢统一调度让这些 Agent 并行工作、互相校验最终输出产品团队可以直接使用的结果。1.1 为什么产品需要“蜂群思维”传统的 AI 功能大多是一个“问答式”闭环。用户输入问题模型输出答案。在产品内部这种方式的局限性很明显单一上下文很难覆盖复杂的业务场景。一个大 Prompt 里堆叠多个任务模型容易“偏科”。任务链路不透明产品团队无法定位问题出在哪个环节。扩展性差每新增一个能力都要改原来的 Prompt牵一发动全身。蜂群思维的多智能体架构则把一个大任务拆解成多个小任务每个小任务由专门的 Agent 处理。这样做的好处是任务边界清晰每个 Agent 的输入输出可预期。可以并行处理多个子任务提升整体响应速度。单个 Agent 升级不影响其他 Agent。每个 Agent 都能被独立测试、独立观测、独立运维。1.2 与 RAG、Prompt Engineering 的关系这里顺便做几个概念区分方便新手理解。Prompt Engineering 解决的是“如何让模型更好地理解需求”。RAG检索增强生成解决的是“如何让模型获得外部知识”。Multi-Agent System 解决的是“如何把任务合理拆解并组织多个模型协同工作”。三者不是替代关系而是互补关系。实际工程中往往先用 RAG 给 Agent 提供知识再用 Prompt Engineering 限定输出格式最后用多智能体架构把它们组织成完整的产品能力。1.3 典型应用场景“The hive mind for your product” 能落地的场景非常多下面列几个比较有代表性的方向场景说明涉及 Agent需求分析根据原始需求描述提取功能点、优先级和依赖关系需求解析 Agent、依赖分析 Agent用户反馈处理对客服工单、应用商店评论、社群反馈进行分类和情感分析分类 Agent、情感分析 Agent、聚合 Agent自动测试根据需求文档生成接口测试用例和边界条件需求 Agent、测试生成 Agent知识库问答结合企业文档做多轮问答和溯源检索 Agent、问答 Agent、引用核对 Agent缺陷分诊根据 Bug 描述判断模块归属、严重级别和推荐处理人文本分析 Agent、路由 Agent看到这里你应该已经感受到多智能体体系的核心并不是“多个模型”而是一套工程化组织方式。下面我们进入架构设计部分。2. 整体架构设计在动手编码之前架构设计很重要。没有一个清晰的架构Agent 一多就容易变成“蜘蛛网”互相调用混乱后期没法维护。2.1 分层架构我推荐把整个系统拆成四层输入层 → 协调层 → Agent 执行层 → 存储与输出层输入层输入层负责接收来自不同渠道的请求比如 Web 表单、API 调用、消息队列消息等。输入层不直接调用 Agent而是把原始数据标准化包装成统一的内部消息结构。协调层协调层是整个蜂群大脑。它负责任务拆解把用户请求拆成多个子任务。Agent 路由根据子任务类型找到合适的 Agent。执行编排决定多个 Agent 是串行执行还是并行执行。结果聚合收集所有 Agent 的返回结果做去重、合并、冲突仲裁。Agent 执行层执行层是具体的“工蜂”。每个 Agent 是一个独立的功能单元内部可以调用大模型、查询数据库、调用外部 API或者执行 Python 代码。对外只暴露统一的接口。存储与输出层存储层保存任务记录、Agent 执行结果、日志和缓存。输出层把最终结果格式化成用户需要的结构比如 JSON、Markdown 文档或者前端可渲染的卡片。2.2 任务拆解与编排策略在实际项目中任务拆解有两种主要方式固定流程拆解适合流程比较稳定的业务。比如需求分析一定是先解析需求再提取功能点最后生成优先级建议。这种拆解方式通过代码写死编排逻辑稳定性高容易调试。动态规划拆解适合比较开放的场景比如“帮我看看这个产品的体验问题”系统需要模型自主决定要调用哪些 Agent。动态拆解灵活但可控性差往往需要引入模型自省和人工审核。对于大多数产品团队我建议先做固定流程拆解跑通之后再考虑动态规划。先稳定再智能。2.3 Agent 通信机制Agent 之间怎么通信我见过几种方案直接 HTTP 调用适合微服务化部署但链路复杂时不好排查。消息队列适合异步任务和削峰但会增加运维复杂度。共享数据库每个 Agent 从数据库读取任务、写入结果简单可靠适合中小型系统。进程内调度所有 Agent 在同一个进程中运行通过函数调用传递数据最简单适合本文这种教学工程。本文的示例会采用“进程内调度 共享数据库”的方式既能讲清楚完整链路又不需要引入额外中间件读者复制到本地就能跑起来。3. 环境准备与技术选型3.1 技术栈说明为了让大家能快速复现本文选用一套轻量级技术栈语言Python 3.9Web 框架FastAPI提供 REST API数据库SQLite保存任务和结果大模型接入使用 OpenAI 兼容接口代码中通过抽象层隔离不直接绑定具体厂商测试工具pytest用于验证核心逻辑版本说明一下Python 建议使用 3.9 以上版本FastAPI 使用当前主流版本即可。由于大模型 SDK 更新较快本文不会把某个 SDK 的版本写死而是提供一个 Mock 模型实现方便没有 API Key 的读者也能跑通整个流程。3.2 项目结构规划建议按下面的目录结构组织项目hive-mind/ ├── main.py # FastAPI 入口 ├── config.py # 配置管理 ├── coordinator.py # 协调器负责任务拆解和结果聚合 ├── agents/ │ ├── __init__.py │ ├── base.py # Agent 基类 │ ├── requirement_agent.py# 需求解析 Agent │ ├── feedback_agent.py # 用户反馈分类 Agent │ ├── testcase_agent.py # 测试用例生成 Agent │ └── mock_llm.py # 大模型调用抽象含 Mock 实现 ├── models/ │ ├── __init__.py │ └── schema.py # 数据模型 ├── data_store.py # SQLite 存储层 ├── requirements.txt └── README.md3.3 为什么用这种选型选择 FastAPI 是因为它支持异步同时自带 OpenAPI 文档方便调试。SQLite 则是零配置文件型数据库适合做原型后续换成 MySQL 或者 PostgreSQL 只需要改存储层实现。最重要的设计决策是抽象出 mock_llm这样即使没有大模型 API Key也可以先开发完整个业务流程之后无缝切换到真实模型。下面进入核心代码实现。4. 核心实现构建你的产品蜂群这一节是全文重点我会按模块逐步讲解。4.1 定义数据结构首先定义整个系统的核心数据结构。我们把任务、Agent 结果、最终聚合结果都抽象成模型。创建文件models/schema.py 数据模型定义 from enum import Enum from typing import List, Optional from pydantic import BaseModel, Field class TaskType(str, Enum): 任务类型枚举 REQUIREMENT_ANALYSIS requirement_analysis FEEDBACK_CLASSIFICATION feedback_classification TESTCASE_GENERATION testcase_generation class AgentResult(BaseModel): 单个 Agent 的执行结果 agent_name: str Field(descriptionAgent 名称) task_type: TaskType Field(description任务类型) status: str Field(defaultsuccess, description执行状态) output: dict Field(descriptionAgent 输出的结构化数据) elapsed_ms: int Field(description耗时(毫秒)) class TaskRequest(BaseModel): 外部传入的任务请求 task_type: TaskType Field(description任务类型) raw_input: str Field(description原始输入文本) options: dict Field(default_factorydict, description附加参数) class TaskResponse(BaseModel): 最终聚合响应 task_id: str Field(description任务 ID) status: str Field(description任务状态) results: List[AgentResult] Field(description所有 Agent 的执行结果) final_output: dict Field(description聚合后的最终输出)说明几个设计要点TaskType使用枚举类型避免字符串满天飞后续新增任务类型只需要扩展枚举。AgentResult里面记录了agent_name和elapsed_ms方便做链路追踪和性能分析。每个 Agent 的output是字典类型可以灵活承载结构化输出不会因为字段变化导致模型类频繁修改。4.2 大模型调用抽象为了避免把代码写死在某一家大模型提供商上我们做一个抽象层。这样后续接 GPT、通义千问、文心或者本地模型只需要替换这一个模块。创建文件agents/mock_llm.py 大模型调用抽象层 在未配置真实 API 时使用 Mock 实现方便本地开发测试 import os import time from typing import List, Dict, Any class BaseLLM: LLM 抽象基类 def complete(self, system_prompt: str, user_prompt: str) - str: raise NotImplementedError class MockLLM(BaseLLM): Mock 实现返回固定格式的数据 方便没有 API Key 的读者运行完整流程 def complete(self, system_prompt: str, user_prompt: str) - str: time.sleep(0.3) # 模拟模型推理耗时 if 需求 in user_prompt or requirement in user_prompt.lower(): return { functional_points: [用户登录, 订单查询, 订单支付], priority: high, dependencies: [用户模块, 订单模块] } if 反馈 in user_prompt or feedback in user_prompt.lower(): return { category: bug, sentiment: negative, module: 支付模块, suggestion: 建议优先修复支付超时问题 } if 测试 in user_prompt or test in user_prompt.lower(): return { test_cases: [ {title: 正常登录, steps: 输入正确账号密码, expected: 登录成功}, {title: 密码错误, steps: 输入错误密码, expected: 提示密码错误}, {title: 账号不存在, steps: 输入未注册账号, expected: 提示账号不存在} ] } return {result: unknown task} class OpenAIConpatLLM(BaseLLM): OpenAI 兼容接口实现 通过环境变量配置 base_url 和 api_key 依赖 openai 库需要 pip install openai def __init__(self): # 这里只做说明实际使用时需要安装 openai 并配置客户端 pass def complete(self, system_prompt: str, user_prompt: str) - str: raise NotImplementedError(真实模型接入请配置 openai 库) def get_llm() - BaseLLM: 工厂方法根据环境变量返回对应的 LLM 实例 if os.getenv(USE_MOCK_LLM, true).lower() true: return MockLLM() return OpenAIConpatLLM()这里的设计意图是应用启动时通过get_llm()决定使用哪个模型实现。默认走 Mock新手可以零成本跑通全流程。有真实 API Key 后继承BaseLLM实现complete方法即可业务代码不用动。需要说明的是OpenAIConpatLLM我故意没有写死具体的调用参数因为各家兼容接口的细节差异比较大。你如果要用真实模型只需要补全__init__和complete方法即可网上有大量参考。4.3 Agent 基类和具体实现先定义 Agent 基类后续所有 Agent 都继承这个基类。创建文件agents/base.py Agent 基类定义 from abc import ABC, abstractmethod from typing import Dict, Any from models.schema import AgentResult, TaskType from agents.mock_llm import BaseLLM class BaseAgent(ABC): 所有 Agent 的抽象基类 def __init__(self, llm: BaseLLM): self.llm llm self.name self.__class__.__name__ abstractmethod def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: 处理输入并返回结构化输出 pass def execute(self, task_type: TaskType, raw_input: str, options: Dict[str, Any]) - AgentResult: 执行入口封装耗时统计和异常处理 子类不需要重复写 try-except import time start time.time() try: output self.process(raw_input, options) status success except Exception as e: output {error: str(e)} status failed elapsed_ms int((time.time() - start) * 1000) return AgentResult( agent_nameself.name, task_typetask_type, statusstatus, outputoutput, elapsed_mselapsed_ms )基类中做了一层封装把异常捕获和耗时统计从业务代码中剥离出来。这样每个具体 Agent 只需要关心“怎么处理输入”不需要关心“怎么上报状态”代码会干净很多。接下来创建需求解析 Agent。创建文件agents/requirement_agent.py 需求解析 Agent 职责从原始需求描述中提取功能点、优先级和依赖关系 import json from typing import Dict, Any from agents.base import BaseAgent class RequirementAgent(BaseAgent): 需求解析 Agent def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: system_prompt 你是一名资深产品经理请从用户需求中提取结构化信息。 输出 JSON包含三个字段 functional_points: 功能点列表 priority: 优先级可选 high/medium/low dependencies: 依赖的模块列表 user_prompt f用户需求如下{raw_input} response self.llm.complete(system_prompt, user_prompt) try: result json.loads(response) except json.JSONDecodeError: result { functional_points: [], priority: medium, dependencies: [], parse_error: 模型输出无法解析为 JSON } result[original_input] raw_input return result这里有个细节解析模型输出时一定要做 JSON 解析异常兜底不能假设模型每次都返回合法 JSON。返回值带上original_input也是必要的方便追溯数据来源。创建文件agents/feedback_agent.py 用户反馈分类 Agent 职责对用户反馈进行分类、情感分析并给出模块归属建议 import json from typing import Dict, Any from agents.base import BaseAgent class FeedbackAgent(BaseAgent): 用户反馈分类 Agent def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: system_prompt 你是一名用户研究工程师请对用户反馈进行分类分析。 输出 JSON包含四个字段 category: 反馈类型可选 feature_request / bug / complaint / praise sentiment: 情感倾向可选 positive / neutral / negative module: 涉及的产品模块 suggestion: 一句话处理建议 user_prompt f用户反馈如下{raw_input} response self.llm.complete(system_prompt, user_prompt) try: result json.loads(response) except json.JSONDecodeError: result { category: unknown, sentiment: neutral, module: unknown, suggestion: 无法自动解析建议人工处理 } result[original_input] raw_input return result创建文件agents/testcase_agent.py 测试用例生成 Agent 职责根据需求描述生成基础测试用例 import json from typing import Dict, Any from agents.base import BaseAgent class TestcaseAgent(BaseAgent): 测试用例生成 Agent def process(self, raw_input: str, options: Dict[str, Any]) - Dict[str, Any]: system_prompt 你是一名测试开发工程师请根据需求生成测试用例。 输出 JSON包含一个字段 test_cases: 测试用例列表每个用例包含 title/steps/expected user_prompt f需求描述如下{raw_input} response self.llm.complete(system_prompt, user_prompt) try: result json.loads(response) except json.JSONDecodeError: result { test_cases: [], parse_error: 模型输出无法解析为 JSON } result[original_input] raw_input return result4.4 协调器设计协调器是整个蜂群的核心负责把任务分发给对应的 Agent并对结果做聚合。创建文件coordinator.py 协调器负责任务路由、Agent 调度和结果聚合 from typing import Dict, List, Any from uuid import uuid4 from models.schema import TaskRequest, TaskResponse, AgentResult, TaskType from agents.requirement_agent import RequirementAgent from agents.feedback_agent import FeedbackAgent from agents.testcase_agent import TestcaseAgent from agents.mock_llm import BaseLLM class Coordinator: 任务协调器 def __init__(self, llm: BaseLLM): self.llm llm self.registry { TaskType.REQUIREMENT_ANALYSIS: RequirementAgent(llm), TaskType.FEEDBACK_CLASSIFICATION: FeedbackAgent(llm), TaskType.TESTCASE_GENERATION: TestcaseAgent(llm), } def dispatch(self, request: TaskRequest) - TaskResponse: 派发任务并聚合结果 简单场景单 Agent 执行 复杂场景根据 options 决定是否联动其他 Agent task_id uuid4().hex[:12] agent self.registry.get(request.task_type) if not agent: return TaskResponse( task_idtask_id, statusfailed, results[], final_output{error: fUnsupported task type: {request.task_type}} ) # 执行主任务 Agent main_result agent.execute(request.task_type, request.raw_input, request.options) # 聚合逻辑如果主任务成功尝试获取补充分析 final_output main_result.output.copy() if request.task_type TaskType.REQUIREMENT_ANALYSIS: # 需求分析联动测试用例生成 testcase_agent self.registry[TaskType.TESTCASE_GENERATION] test_result testcase_agent.execute( TaskType.TESTCASE_GENERATION, request.raw_input, request.options ) final_output[test_cases_suggestion] test_result.output.get(test_cases, []) return TaskResponse( task_idtask_id, statussuccess if main_result.status success else partial, results[main_result, test_result], final_outputfinal_output ) if request.task_type TaskType.FEEDBACK_CLASSIFICATION: # 反馈分类联动处理建议 # 实际项目中可以在这里接入工单系统或通知服务 final_output[handling_note] 建议在 24 小时内人工复核该反馈 return TaskResponse( task_idtask_id, statusmain_result.status, results[main_result], final_outputfinal_output ) return TaskResponse( task_idtask_id, statusmain_result.status, results[main_result], final_outputfinal_output )这段代码的核心是dispatch方法。当前实现里需求分析任务会自动触发测试用例生成 Agent这就体现了“蜂群”的协作特性一个 Agent 的输出可以作为另一个 Agent 的补充而不是所有逻辑都堆在一个函数里。在实际项目中你可以把这种联动关系配置在数据库里而不是写在代码里比如维护一张agent_linkage表这样新增联动关系就不用改代码了。4.5 存储层实现为了让任务结果可追溯我们需要把每次请求保存到数据库。创建文件data_store.py SQLite 存储层 import sqlite3 import json from typing import Dict, Any class DataStore: 基于 SQLite 的简单存储实现 def __init__(self, db_path: str hive_mind.db): self.conn sqlite3.connect(db_path, check_same_threadFalse) self._create_table() def _create_table(self): cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS task_records ( task_id TEXT PRIMARY KEY, task_type TEXT, raw_input TEXT, result TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) self.conn.commit() def save_task(self, task_id: str, task_type: str, raw_input: str, result: Dict[str, Any]): 保存任务记录 cursor self.conn.cursor() cursor.execute( INSERT OR REPLACE INTO task_records (task_id, task_type, raw_input, result) VALUES (?, ?, ?, ?), (task_id, task_type, raw_input, json.dumps(result, ensure_asciiFalse)) ) self.conn.commit() def get_task(self, task_id: str) - Dict[str, Any] | None: 查询任务记录 cursor self.conn.cursor() cursor.execute( SELECT task_id, task_type, raw_input, result FROM task_records WHERE task_id ?, (task_id,) ) row cursor.fetchone() if not row: return None return { task_id: row[0], task_type: row[1], raw_input: row[2], result: json.loads(row[3]) } def close(self): self.conn.close()SQLite 的check_same_threadFalse在 FastAPI 多线程场景下能避免不必要的报错不过这是简单写法。生产环境建议改用连接池或者直接用 SQLAlchemy。4.6 FastAPI 入口最后是 FastAPI 入口文件把前面所有模块串起来。创建文件main.py FastAPI 入口 from fastapi import FastAPI from models.schema import TaskRequest, TaskResponse from coordinator import Coordinator from agents.mock_llm import get_llm from data_store import DataStore app FastAPI( titleHive Mind API, descriptionThe hive mind for your product - 多智能体产品协作系统, version0.1.0 ) # 初始化协调器和存储层 llm get_llm() coordinator Coordinator(llm) data_store DataStore() app.get(/health) def health(): 健康检查 return {status: ok} app.post(/api/v1/task, response_modelTaskResponse) def submit_task(request: TaskRequest): 提交任务 response coordinator.dispatch(request) data_store.save_task( task_idresponse.task_id, task_typerequest.task_type.value, raw_inputrequest.raw_input, resultresponse.model_dump() ) return response app.get(/api/v1/task/{task_id}) def get_task(task_id: str): 查询任务记录 record data_store.get_task(task_id) if not record: return {error: task not found} return record app.on_event(shutdown) def shutdown(): data_store.close()有一个地方需要提一下response.model_dump()是 Pydantic v2 的写法如果你用的是 Pydantic v1需要改成response.dict()。FastAPI 新版默认使用 Pydantic v2但如果你的环境比较旧要注意这个兼容性问题。4.7 依赖文件与启动命令创建文件requirements.txtfastapi uvicorn pydantic注意openai库只有在接入真实大模型时才需要Mock 模式下不需要安装。启动服务cd hive-mind pip install -r requirements.txt uvicorn main:app --reload --port 8000看到日志输出Uvicorn running on http://127.0.0.1:8000就说明服务启动成功了。5. 运行与验证5.1 测试需求分析任务打开终端执行下面的 curl 命令提交一个需求分析任务curl -X POST http://127.0.0.1:8000/api/v1/task \ -H Content-Type: application/json \ -d { task_type: requirement_analysis, raw_input: 我们希望在 App 内增加一个订单管理功能用户可以查看历史订单、取消未发货订单、申请售后。 }预期返回结果大致如下{ task_id: a1b2c3d4e5f6, status: success, results: [ { agent_name: RequirementAgent, task_type: requirement_analysis, status: success, output: { functional_points: [用户登录, 订单查询, 订单支付], priority: high, dependencies: [用户模块, 订单模块], original_input: 我们希望在 App 内增加... }, elapsed_ms: 310 }, { agent_name: TestcaseAgent, task_type: testcase_generation, status: success, output: { test_cases: [ {title: 正常登录, steps: 输入正确账号密码, expected: 登录成功}, {title: 密码错误, steps: 输入错误密码, expected: 提示密码错误} ], original_input: 我们希望在 App 内增加... }, elapsed_ms: 305 } ], final_output: { functional_points: [用户登录, 订单查询, 订单支付], priority: high, dependencies: [用户模块, 订单模块], test_cases_suggestion: [ {title: 正常登录, steps: 输入正确账号密码, expected: 登录成功}, {title: 密码错误, steps: 输入错误密码, expected: 提示密码错误} ] } }这个返回结果里可以清楚看到需求分析任务自动触发了测试用例生成 Agent最终返回的final_output同时包含需求解析内容和测试用例建议。这就是蜂群协作的效果。5.2 测试用户反馈分类任务再测试一个用户反馈分类场景curl -X POST http://127.0.0.1:8000/api/v1/task \ -H Content-Type: application/json \ -d { task_type: feedback_classification, raw_input: 最近在支付页面经常超时用户反馈支付失败客户投诉率大幅上升需要尽快处理 }返回结果的关键部分如下{ final_output: { category: bug, sentiment: negative, module: 支付模块, suggestion: 建议优先修复支付超时问题, original_input: 最近在支付页面经常超时..., handling_note: 建议在 24 小时内人工复核该反馈 } }系统能识别出这是 Bug、情感为负面、涉及支付模块并给出处理建议和人工复核提醒。这套能力放到真实产品里可以对接工单系统实现自动分诊。5.3 查询历史任务每次提交的任务都会持久化到 SQLite可以通过下面的接口查询curl http://127.0.0.1:8000/api/v1/task/a1b2c3d4e5f6返回历史任务的完整记录包括原始输入和处理结果。这为后续做数据分析、模型效果评估保留了真实的样本数据。6. 常见问题与排查思路在开发和运行这套系统的过程中比较容易遇到下面几类问题。问题现象常见原因解决思路启动报错ModuleNotFoundError: No module named fastapi未安装依赖执行pip install -r requirements.txt启动报错Pydantic model_dump not foundPydantic 版本过旧升级到 Pydantic v2或改用.dict()需求分析结果为空Mock 模型未命中关键字检查agent输入文本是否包含“需求”等触发词返回 JSON 解析失败大模型输出非法 JSON在 Agent 中增加解析兜底逻辑并对结果做人工审核多线程访问 SQLite 报错默认连接不支持多线程初始化时设置check_same_threadFalse真实模型接不上base_url 或 api_key 配置错误先确认网络连通性再检查认证参数任务执行超时大模型推理耗时过长为每个 Agent 设置超时时间超时走降级策略6.1 排查清单如果你提交任务后返回了异常结果按下面顺序排查先看results中每个 Agent 的status字段定位是哪个 Agent 失败了。再看output中的error信息确认是模型解析问题还是逻辑异常。查看服务端控制台日志确认请求是否到达 Agent。如果使用真实模型检查 API 调用耗时和 token 消耗确认是否触发限流。检查 SQLite 是否有历史记录确认存储层是否正常持久化。核心原则是先定位是“模型问题”还是“工程问题”再对症下药。很多新手一看到错误就怀疑大模型实际上很多问题是代码边界条件没处理好。7. 工程化最佳实践把这套系统从 Demo 推向生产环境有几个关键点值得注意7.1 Agent 粒度设计Agent 不是越细越好。粒度过粗每个 Agent 承担太多职责又退化成“一个大 Prompt 解决所有问题”。粒度过细Agent 之间通信成本增加链路变长维护困难。我的建议是遵循“单一职责原则”一个 Agent 只做一件可命名的事情。如果这个 Agent 的 Prompt 已经超过几百字考虑拆分。如果多个 Agent 经常一起被调用考虑合并成一个流程型 Agent。7.2 上下文管理与记忆真实业务中Agent 之间往往需要共享信息。比如需求分析 Agent 识别出“订单模块”测试用例 Agent 需要知道这个信息来生成更有针对性的用例。目前示例中每个 Agent 都独立处理原始输入没有共享上下文。在实际工程中建议引入“上下文总线”机制每个 Agent 执行前从总线拉取需要的信息。执行后把输出写入总线。协调器负责控制总线的读写权限避免数据污染。7.3 可观测性Agent 系统最大的问题就是不可控。要让系统可维护必须有完整的可观测性方案每个 Agent 记录输入摘要、输出摘要、耗时、token 消耗。使用链路追踪 ID 串起整个调用链。对模型输出做版本快照方便回放和对比效果。核心场景建立自动化测试模型升级后自动回归验证。7.4 安全与权限边界多智能体系统如果接入企业管理后台必须考虑权限边界不同 Agent 使用独立的 API Key避免一个 Key 泄露导致全部接口暴露。对 Agent 可访问的数据做最小权限控制比如财务数据只有财务 Agent 能读取。外部输入要经过过滤和脱敏防止提示注入攻击。涉及删除、修改类操作前必须有人工审批环节。7.5 模型降级策略大模型服务不可用时系统不能直接崩溃。建议设计降级链路一级降级从高精度大模型切换到轻量模型。二级降级返回规则匹配的兜底答案。三级降级返回人工处理提示并创建工单通知相关负责人。示例代码中的MockLLM实际上就是一种降级实现保证本地无网络也能跑通流程。7.6 性能优化思路当前框架是同步执行多个 Agent 串行耗时累加。对于响应时间要求高的场景可以改成并发执行使用 FastAPI 的异步接口。通过asyncio.gather并行调用多个 Agent。对耗时长的 Agent 做缓存相同输入直接返回历史结果。引入消息队列把耗时任务异步化前端先返回“处理中”处理完成再通过 WebSocket 推送结果。8. 总结与下一步学习路线这篇文章围绕“The hive mind for your product”这个主题从概念到落地实现了一个多智能体产品协作系统。核心内容包括什么是蜂群思维以及它在产品工程中的含义。一个四层架构输入层、协调层、Agent 执行层、存储与输出层。完整可运行的多 Agent 工程包含需求解析 Agent、用户反馈分类 Agent、测试用例生成 Agent。协调器如何做任务路由、联动编排和结果聚合。基于 SQLite 的任务持久化方案。生产环境的多项关键设计可观测性、安全边界、降级策略、性能优化。如果你想继续深入我建议按下面路径学习先跑通本文示例理解每个模块的作用。为某个 Agent 接入真实大模型感受 Mock 与真实模型的差异。把固定流程拆解改成配置化让调度逻辑可以通过页面或接口动态调整。引入向量数据库给 Agent 增加长期记忆能力实现更复杂的业务场景。研究 LangGraph、AutoGen 等成熟的多智能体框架对比它们的编排思路与本文手写方案的异同。最后给一个实操建议不要一开始就追求复杂的动态规划先把一个垂直场景做扎实。你可以从“用户反馈自动分诊”这个场景开始用三个 Agent 跑通全链路再逐步扩展到需求、测试、知识库等方向。蜂群的力量来自分工与协作而你负责做的是给每只“工蜂”划清边界、定好规则。这套系统上手之后你会明显感受到单一的问答式 AI 交互和真正的产品级智能体体系之间差距不是一点半点。