企业官网建设流程全解析

大模型应用做了一年多，我最大的感受是：模型换得再快，也填不上工程链路里挖的坑。从一个RAG知识库问答项目起步，后来演进成跨部门协作的多智能体架构，中间踩了无数坑才发现，真正拉开项目差距的，是能不能把“工程规范”这四个字落到每一行代码里。这篇把RAG检索链路、多智能体协作、评估回归、线上可观测性这几块的关键点串起来，附带一套可以复用的精简代码。如果你正在做知识库问答、客服助手、复杂任务编排，或者只是想把一个能跑的Demo变成能上线维护的系统，这篇文章值得你花半小时读完。

1. 先把“大模型工程规范”这个提法聊透

1.1 为什么规范比模型本身更能决定项目成败

很多团队有个默认观念：只要模型换得够新、够强，应用效果就会跟着变好。实际做过几个项目之后，我觉得这个观念害人。模型本身的能力当然重要，但大部分线上质量问题并不是“模型不够聪明”，而是“链路没法控制”。比如同一个问题，上周和这周答案不一样，你以为是模型幻觉，其实是分块参数被谁改过；换了一个新的Embedding模型之后，明明向量召回变准了，坏答案却变多了，问题往往出在重排序策略没有跟着调。

这类现象的本质，是工程链路缺少可约束、可验证的规范。我理解的工程规范不是一堆流程文档，而是围绕数据、检索、生成、评估、可观测性形成的一套可执行约定。它要回答几个具体问题：数据怎么清洗和分块？分块参数改动的回归标准是什么？提示词如何做版本管理？每个回答能否追溯到命中了哪些资料、调用了哪些工具？

没有这些约定，项目刚到线上就会陷入被动。我也见过一个团队，Demo演示时效果惊艳，全量上线后用户一问就崩，排查了三天，最后发现是文档格式多样化导致分块不均匀，检索结果大量截断。这个问题的根子不在模型，而在工程规范缺位。

1.2 工程规范的四层拆解

我把大模型工程规范拆成四层，每一层都有明确的产出物和检验方式：

第一层是数据规范。包括文档接入、去重、清洗、格式标准化、元数据设计、更新策略。元数据尤其容易被忽略，检索结果如果没有文档标题、来源链接、更新时间，模型拿到一堆裸正文，很难正确标注引用，更谈不上可追溯。

第二层是检索规范。包括Embedding选型、混合检索策略、相似度阈值、Top-K范围、重排序触发条件。检索层的改动对结果影响最大，也最容易出现“指标没变但效果忽上忽下”的诡异现象，所以必须有回归评估兜底。

第三层是生成规范。包括系统提示词结构、引用格式约束、禁止编造的表达方式、输出JSON化规则。生成层的核心目标是稳定，不是让模型自由发挥。多数应用场景下，我宁愿要一个“没找到就说没找到”的助手，也不想要一个“自信地胡编”的助手。

第四层是评估规范。包括离线金标准集、回归冒烟集、线上指标口径、坏例复盘机制。这一层是前面所有规范的保险丝。没有评估，任何改动都像蒙眼换零件。

这四层不是一开始就要做到完美，而是边跑边补。真正有效的规范通常来自故障驱动：线上出一次问题，就沉淀一条检查项。半年之后再回看，你会发现自己其实已经构建了一套轻量但完整的大模型应用治理体系。

2. RAG落地的关键取舍：分块、嵌入、重排序与召回质量

2.1 分块策略不是玄学，是有数据支撑的参数调优

RAG项目最先遇到的问题就是“文本怎么切”。网上常见说法是固定512或者1024个Token，但不同文档类型、不同语言、不同问答粒度，对分块的要求完全不一样。我整理过一组对比，这个表格后来成了内部培训材料：

我一般推荐的做法是：先按标题拆出大致层级，再对超长块做二次切分；块的上限可以设到800到1200个Token左右，具体数值需要拿真实问题集做一次召回实验，而不是拍脑袋。分块之后，给每个块打上元数据，包括文档标题、一级栏目、时间、来源URL、正文Key。这一步很关键，但很多人会跳过。设想一下，检索回来的块只有一段内容，没有任何可展示的来源信息，生成环节想给用户一个引用链接都做不到，最终只能靠模型硬编，幻觉率自然直线上升。

我踩过的比较典型的一次，是某类技术手册的图片说明和正文混在一起，按固定窗口切完后，很多块变成半句话加上一张图的描述，检索召回的准确率明显下降。后来改成按标题层级切块，再给图文说明单独成块，同一个测试集上的命中率提升了二十多个百分点。这个数字说明，检索效果的差距很多时候不在模型，而在上游数据组织方式。

2.2 召回不是越多越好，混合检索和重排序才是提分关键

单纯依赖向量检索在不少业务场景下是不够的。向量召回擅长语义相近，但业务问答里很多准确答案依赖关键词精确匹配，比如工单编号、产品型号、人名。两个文本语义很接近，可能根本不是用户要的那个东西。所以我建议生产环境尽量做“向量召回 + 关键词召回”的混合，再通过重排序把两类结果融合成最终上下文。

重排序里最省成本的方案是RRF（Reciprocal Rank Fusion），公式直观到不需要什么深度背景：

RRF(d) = Σ 1 / (k + rank_i(d))

其中rank_i(d)是文档d在第i路召回结果中的排名，k是一个平滑参数，一般取60。它的思路很简单：某个文档在越多路的召回列表里排名越靠前，它的融合分就越高。RRF的好处是不需要训练、不用部署额外模型，跑得快，而且效果通常比简单合并去重要好不少。

如果预算允许，也可以在RRF之后再接一个专门的交叉编码器重排序模型，让模型对“查询和文档”的完整组合做相关性打分。交叉编码器的精度比向量相似度更准，但推理成本高，一般只对前20到30个候选做精排，最后取前5到8个进上下文。

召回数量这块，我建议不要一开始就拉Top-K到20甚至更多。上下文塞得越长，模型在生成时越容易被噪声带偏，还会显著增加Token开销。我自己习惯的做法是：混合检索各取10到20个候选，重排序后只保留5到8个上下文块。这个数量在多数知识库问答场景里够用，回答质量和Token成本也比较平衡。

3. 一套不依赖重型框架的RAG参考实现

3.1 模块划分与代码骨架

很多教程上来就是某个现成框架的封装，方便是方便，但一旦涉及自定义分块、混合检索比例、日志埋点，就会碰壁。所以我在这里写一套尽量精简的原生实现，只依赖requests和numpy，所有大模型接口调用都以兼容的“聊天补全”和“向量嵌入”格式为例，你换成自己实际使用的服务地址就能跑。

代码分几个模块：文本切分、向量存储、混合检索、重排序、上下文组装、最终生成、离线评估。我把它们拆开讲，方便你按需裁剪。

3.2 主流程实现

先看配置和基础调用函数。生产环境里API Key和Base地址不要硬编码，建议从环境变量或配置中心读取：

import os import re import json import requests import numpy as np # 换成你实际使用的大模型服务地址，接口以 /v1/chat/completions 和 /v1/embeddings 兼容格式为例 LLM_BASE = os.getenv("LLM_BASE", "https://your-llm-service/v1") API_KEY = os.getenv("LLM_API_KEY", "") def get_embedding(text: str) -> list: r = requests.post( f"{LLM_BASE}/embeddings", headers={"Authorization": f"Bearer {API_KEY}"}, json={"model": "embedding-model-name", "input": text}, timeout=30, ) r.raise_for_status() return r.json()["data"][0]["embedding"] def chat(messages, temperature=0.2, max_tokens=1024) -> str: r = requests.post( f"{LLM_BASE}/chat/completions", headers={"Authorization": f"Bearer {API_KEY}"}, json={ "model": "chat-model-name", "messages": messages, "temperature": temperature, "max_tokens": max_tokens, }, timeout=60, ) r.raise_for_status() return r.json()["choices"][0]["message"]["content"]

然后是文本切分。这里先按标题拆块，再对超长块做二次切分，保证上下文连续性：

def split_text(text: str, max_chars: int = 1200) -> list[dict]: raw_blocks = [] current = {"title": "", "content": ""} for line in text.splitlines(): # 识别 Markdown 标题或者“第X章/节”这类结构 if re.match(r"^(#{1,6}\s|第[一二三四五六七八九十\d]+[章节])", line.strip()): if current["content"] or current["title"]: raw_blocks.append(current) current = {"title": line.strip(), "content": ""} else: current["content"] += line + "\n" if current["content"] or current["title"]: raw_blocks.append(current) final_blocks = [] for b in raw_blocks: if len(b["content"]) <= max_chars: final_blocks.append(b) else: parts = b["content"].split("\n\n") temp = {"title": b["title"], "content": ""} for p in parts: if len(temp["content"]) + len(p) > max_chars and temp["content"]: final_blocks.append(temp) temp = {"title": b["title"] + "（续）", "content": p + "\n\n"} else: temp["content"] += p + "\n\n" if temp["content"]: final_blocks.append(temp) return final_blocks

接着是向量存储和检索。这里用内存列表实现，生产环境换成你实际在用的向量库客户端即可，接口保持一致：

class VectorStore: def __init__(self): self.texts = [] self.metas = [] self.vectors = [] def add(self, text: str, meta: dict, embedding: list): self.texts.append(text) self.metas.append(meta) self.vectors.append(np.array(embedding, dtype="float32")) def search(self, query_emb, top_k=20): if not self.vectors: return [] q = np.array(query_emb, dtype="float32") scored = [] for i, v in enumerate(self.vectors): score = float(np.dot(q, v) / (np.linalg.norm(q) * np.linalg.norm(v) + 1e-9)) scored.append((score, i)) scored.sort(reverse=True) return [(self.texts[i], self.metas[i], s) for s, i in scored[:top_k]]

再配合一个轻量的关键词召回，模拟“精确匹配”的互补作用。完整BM25在生产环境可以接入现成检索服务，这里用词频重叠做了一个简化版，目的是展示混合召回的路子：

def keyword_retrieve(blocks, query, top_k=10): def tokenize(t): return re.findall(r"[\w\u4e00-\u9fff]{2,}", t.lower()) q_terms = set(tokenize(query)) scored = [] for i, b in enumerate(blocks): words = tokenize(b["content"]) hit_count = sum(1 for w in words if w in q_terms) scored.append((hit_count, i, b)) scored.sort(key=lambda x: x[0], reverse=True) return [(b, {"title": b.get("title", "")}, s) for s, i, b in scored if s > 0][:top_k]

两张召回结果做RRF融合，最后输出Top-N：

def rrf_fusion(ranked_lists, k=60, top_k=8): score_map = {} item_map = {} for source, items in ranked_lists.items(): for rank, (text, meta, score) in enumerate(items): key = (meta.get("title", ""), text[:100]) # 生产环境建议改用文档ID score_map[key] = score_map.get(key, 0) + 1 / (k + rank + 1) item_map[key] = (text, meta) sorted_keys = sorted(score_map.items(), key=lambda x: x[1], reverse=True)[:top_k] return [(item_map[k][0], item_map[k][1], s) for k, s in sorted_keys]

生成侧的上下文组装和提问，这里有一个强调“禁止编造”的系统提示词模板。不要小看这几行字，它对答案可控性的提升是立竿见影的：

def build_messages(query, context_blocks): context = "\n\n".join( f"[{i+1}] {b.get('title', '')}\n{b.get('content', '')[:800]}" for i, b in enumerate(context_blocks) ) system_prompt = ( "你是一个企业知识库问答助手。回答只能基于下面的参考资料。" "如果资料中没有答案，必须明确回答'资料库中未找到相关答案'，不要编造。" "回答时在关键信息后标注来源编号，例如[1]。\n\n参考资料：\n" + context ) return [ {"role": "system", "content": system_prompt}, {"role": "user", "content": query}, ] def rag_answer(question, index, blocks): qemb = get_embedding(question) vector_hits = index.search(qemb, top_k=20) keyword_hits = keyword_retrieve(blocks, question, top_k=10) fused = rrf_fusion( {"vector": vector_hits, "keyword": keyword_hits}, top_k=8, ) context_blocks = [ {"title": meta.get("title", ""), "content": text} for text, meta, _ in fused ] messages = build_messages(question, context_blocks) return chat(messages)

这套代码虽然没有引入重型框架，但已经覆盖了RAG链路的完整骨架：导入、切分、入库、双路召回、融合重排、生成。你可以在此基础上补齐缓存、并发、日志、灰度开关，逐步演进成适合自己团队的检索服务。

3.3 评估脚本：没有评估的RAG等于盲跑

代码跑通了，这只是开始。真正让RAG进入“可维护”状态的，是一套离线评估脚本。我每次改动分块参数、换Embedding模型、调整Top-K，都会先跑一遍回归，再决定要不要上线上灰度。

一个实用的评估集不需要很大，但必须贴近真实用户问题。我建议先整理二十到三十个业务高频问题，每个问题标出“必须命中的关键词”和“期望引用的资料标题片段”。评估脚本判定的标准可以分成两层：第一层是答案里是否包含所有关键要素，第二层是回答是否带上了正确的引用来源。两者都满足，才算这一条通过：

GOLD_SET = [ { "q": "员工的带薪年假标准是什么？", "must_terms": ["年假", "5天"], "source_hint": "休假管理制度", }, { "q": "如何申请远程办公权限？", "must_terms": ["远程办公", "审批"], "source_hint": "IT服务指南", }, ] def evaluate(index, blocks, gold_set=None): gold_set = gold_set or GOLD_SET passed = 0 total = len(gold_set) for item in gold_set: ans = rag_answer(item["q"], index, blocks) ok = all(term in ans for term in item["must_terms"]) ok = ok and (item.get("source_hint", "") in ans) passed += 1 if ok else 0 print(("PASS" if ok else "FAIL"), item["q"]) print(f"准确率: {passed/total:.1%}") return passed / total

这套评估集的价值不只是数字。它会逼着你把“用户到底会问什么”这个问题想清楚，也会在你改完代码后第一时间告诉你“召回链路有没有被改坏”。我自己的习惯是，Gold Set每两周更新一次，把线上出现的坏例回收进来。这种做法成本很低，但收益极其明显。

4. 多智能体架构的模式拆解与应用边界

4.1 从“单次对话”到“多个智能体协作”，到底变了什么

RAG解决的是“让模型先检索再回答”，它本质上是单链条。多智能体解决的是“任务需要多步决策、多个动作、多个信息来源”的复杂场景。一个智能体可以理解为：一组系统提示词、一组工具、一段记忆状态、一个决策循环。它接收任务，判断需要调用哪些工具，根据工具返回再决定下一步。

单智能体模式下，所有逻辑都塞在一个超长提示词里。问题一来，工具变多之后提示词会越来越长，上下文越来越杂，模型在“应该调用哪个工具”这件事上的判断会变得不稳定。实际的效果就是：简单问答还凑合，稍微复杂一点的任务就开始绕圈或者漏步骤。

类比一下，一个全能的个体员工确实很强，但面对跨模块、跨流程、需要并行核实信息的任务时，单个人很容易陷入单点瓶颈。多智能体的思路就是组建一个项目组：负责查资料的查资料，负责算数据的算数据，负责审核的审核。每个角色只维护自己的提示词和工具集合，边界清晰，出错时也好定位。

不过，多智能体不应该是“为了多而多”。我见过不少项目，两三个问题就能解决的事情，硬生生拆出三四个智能体，最后的结果是消耗三倍Token、增加一半延迟，回答质量没提升，反而更容易出错。判断是否值得上多智能体，要看任务是否具备三个特征：可拆解成明确的子任务、子任务之间有清晰的依赖或并行关系、需要不同领域的专业工具或提示策略。缺了这三条，还是老老实实用单链路加工具调用更划算。

4.2 常见的三种编排模式

多智能体的编排方式，我分成三种基本模式，实际项目里往往是它们的组合。

第一种是主管-执行者模式，也就是Orchestrator-Worker。由主管智能体负责拆解任务、分派子任务、汇总结果。优点是职责清晰，主管管计划和路由，执行者管具体动作。缺点是主管一旦拆错任务，整个链路都会跑偏，所以主管的子任务输出最好用JSON做结构化约束，别让它自由发挥。

第二种是流水线模式。任务按固定顺序经过多个智能体，前一个的输出是后一个的输入。适合内容生产线类的场景，比如选题智能体生成大纲，写作智能体扩写初稿，审核智能体做合规检查，最后润色智能体做统一风格。优点是可以针对每一环单独优化，缺点是链路较长时，误差会逐级累积，中间任何一环输出不稳定，最终结果都会受影响。

第三种是评审-反思模式。由一个智能体生成答案，另一个智能体专门挑毛病，再把评审意见反馈回去修正。这个模式非常适合对回答准确性要求高的场景，代价是至少两次以上额外的大模型调用，延迟和成本需要提前算清楚。

我在实际项目里用的最多的是主管-执行者模式，然后在个别关键节点叠加评审-反思。多智能体架构本身不复杂，复杂的是如何控制每一条链路的输出质量。下一节我会给出一个可以直接参考的实现。

5. 多智能体工程落地代码：任务分解、工具调度与状态收敛

5.1 智能体注册表和工具原语

先定义工具规格、智能体规格和任务三种数据结构。工具注册表用来统一存放可调用能力，智能体只关心自己绑定哪些工具：

from dataclasses import dataclass, field from typing import Callable, List, Dict, Optional @dataclass class ToolSpec: name: str description: str handler: Callable TOOL_REGISTRY: Dict[str, ToolSpec] = {} def register_tool(spec: ToolSpec): TOOL_REGISTRY[spec.name] = spec @dataclass class AgentSpec: name: str system_prompt: str tools: List[str] @dataclass class Task: id: str goal: str input_data: dict = field(default_factory=dict) output: str = "" assigned_agent: str = "" status: str = "pending"

再注册几个示例工具。这里我以一个知识检索工具和一个文档解析工具为例，你可以替换成真实业务里的查询接口、计算服务、审批流工具：

def knowledge_search(query: str, limit: int = 5, context: Optional[dict] = None) -> dict: # 实际项目里这里会调用第 3 节的 RAG 检索服务 return {"query": query, "top": limit, "result": "检索到的参考内容（示例）"} def document_parse(file_id: str, context: Optional[dict] = None) -> dict: return {"file_id": file_id, "parsed": True, "page_count": 12} register_tool(ToolSpec(name="knowledge_search", description="检索企业知识库", handler=knowledge_search)) register_tool(ToolSpec(name="document_parse", description="解析并结构化指定文档", handler=document_parse))

工具注册表这个设计很小，但它带来了两个直接好处：一是新增能力不需要改智能体循环代码，插拔式扩展；二是工具层的调用记录可以统一埋点，方便做可观测性。

5.2 主管-执行者编排的核心循环

智能体运行时的核心逻辑是“循环决策”。模型每一步输出两种结果之一：要么是一个结构化工具调用，要么是最终答案。每次工具返回后，把结果追加进对话上下文，继续循环，直到模型给出最终答案或达到最大轮数。

下面是简化版的核心循环，我保留了解析、失败兜底和轮数上限：

class AgentRuntime: def __init__(self, spec: AgentSpec, max_turns: int = 5, trace: Optional[list] = None): self.spec = spec self.max_turns = max_turns self.trace = trace if trace is not None else [] def available_tools(self): return { name: spec for name, spec in TOOL_REGISTRY.items() if name in self.spec.tools } def run(self, task: Task) -> Task: messages = [ {"role": "system", "content": self.spec.system_prompt}, {"role": "user", "content": task.goal}, ] tool_descs = "\n".join( f"- {name}: {spec.description}" for name, spec in self.available_tools().items() ) tool_prompt = ( "你可以使用以下工具：\n" + tool_descs + "\n如果需要调用工具，严格按下面的 JSON 格式输出：" '{"tool": "工具名", "args": {...}}\n' "如果不需要调用工具，直接输出最终答案。" ) messages.append({"role": "system", "content": tool_prompt}) for turn in range(self.max_turns): reply = chat(messages).strip() if reply.startswith("{"): try: cmd = json.loads(reply) tool_spec = self.available_tools().get(cmd["tool"]) if not tool_spec: raise ValueError(f"未知工具: {cmd['tool']}") self.trace.append({"turn": turn + 1, "tool": cmd["tool"], "args": cmd.get("args", {})}) tool_result = tool_spec.handler(**cmd.get("args", {}), context={"task": task.input_data}) messages.append({"role": "assistant", "content": reply}) messages.append({ "role": "user", "content": f"工具返回结果：{json.dumps(tool_result, ensure_ascii=False)}", }) except Exception as e: messages.append({"role": "user", "content": f"工具调用失败，请重试：{e}"}) else: task.output = reply task.status = "done" return task task.output = "达到最大调用轮数，任务未完成。" task.status = "timeout" return task

主管智能体的核心职责是把一个主任务拆解成多个子任务，并指定每个子任务对应的执行智能体。同样，子任务计划也要求结构化输出，方便解析和兜底：

def orchestrator_run(plan_agent, workers, synthesizer, main_task): # 第一步：让主管智能体产出子任务计划 plan_task = plan_agent.run(Task( id=f"{main_task.id}-plan", goal=f"请为以下任务生成执行计划，输出 JSON：" f'{{"steps": [{{"id": "s1", "target": "worker名称", "goal": "子任务目标"}}]}}\n任务：{main_task.goal}', )) try: plan = json.loads(plan_task.output) except Exception: # 兜底：如果主管输出无法解析，直接交给综合智能体汇总 return synthesizer.run(main_task).output step_results = {} for step in plan.get("steps", []): worker = workers.get(step.get("target")) if not worker: step_results[step["id"]] = "未找到对应执行智能体" continue sub_task = Task(id=step["id"], goal=step.get("goal", "")) sub_task.assigned_agent = step.get("target", "") step_results[step["id"]] = worker.run(sub_task).output final_task = Task( id=f"{main_task.id}-final", goal=( f"根据下面的子任务执行结果，回答用户原始问题。\n" f"原始问题：{main_task.goal}\n" f"子任务结果：{json.dumps(step_results, ensure_ascii=False)}" ), ) return synthesizer.run(final_task).output

这段代码已经是一个能跑的主管-执行者框架。实际生产环境要补的内容还有不少：计划结果要加一层校验规则，子任务要做幂等控制，执行结果要缓存，失败要进行一次性重试。但核心骨架已经很清楚了。

5.3 并发控制、超时与回滚设计

多智能体系统最大的工程问题是“失控”。失控的表现有三个：单个智能体循环不退出、多个子任务串行导致延迟成倍上升、某个子任务失败后把错误结果带进最终答案。我逐个说。

第一个问题，循环不退出。核心循环里的最大轮数就是为此准备的，但我建议再给模型的每次输出增加一层校验：如果连续三次输出同一个工具调用，直接终止并标记异常，而不是傻傻地继续烧Token。

第二个问题，子任务串行。没有依赖关系的子任务完全可以并行。Python里用线程池就能解决，简单加一个并发层：

from concurrent.futures import ThreadPoolExecutor def run_steps_parallel(workers, steps, max_workers=4): def one_step(step): worker = workers.get(step.get("target")) if not worker: return step.get("id"), "未找到对应执行智能体" sub_task = Task(id=step.get("id", ""), goal=step.get("goal", "")) sub_task.assigned_agent = step.get("target", "") return sub_task.id, worker.run(sub_task).output with ThreadPoolExecutor(max_workers=max_workers) as pool: return dict(pool.map(one_step, steps))

用并发之前要先检查依赖关系。如果子任务之间有关联，比如第二步要用第一步的结果，那就不能并行，硬要并行只会造出一堆脏数据。

第三个问题，失败回滚。子任务失败时，不要直接把异常信息拼进最终答案。较稳妥的做法是：把失败子任务标记为failed，把它从最终汇总结果中剔除，并让综合智能体明确告诉用户“部分步骤失败”。同时，在链路入口生成一个request_id，所有子任务的trace都挂在这个ID下，这样排查问题的时候，一条链路从头到尾都能串起来。这个细节看起来不起眼，真上线之后能救你很多次。

6. 从RAG到多智能体：演进路线与实践踩坑

6.1 不要一上来就上多智能体，先按阶段演进

我建议的演进路线是：先做扎实的RAG，再引入工具调用，然后变成智能体，最后才是多智能体。每一步都要有明确的功能驱动，不要靠想象。

阶段一是RAG，解决“模型没有内部知识、需要对接外部知识库”的问题。阶段二是RAG加上工具调用，让模型在检索之外还能执行计算、查询接口、写工单这类动作。阶段三是单智能体加路由，把不同领域的任务分发给不同提示词和工具组合，让路由层根据用户意图做分发。阶段四才是多智能体，在任务足够复杂、天然需要多角色配合时引入。

很多团队跳过了前两个阶段，直接想上多智能体，结果发现最缺的不是智能体框架，而是“检索质量太差、工具接口不完整”。多智能体放大了底层能力的问题，而不是替代它。

6.2 成本、延迟与稳定性的三重平衡

多智能体的代价一定要提前算清楚。主管一次拆解、每个执行者一到三次决策循环、最终综合智能体一次汇总，一套链路跑下来至少是三到五次大模型调用。如果再加上并行执行、失败重试、评审反思，Token消耗很容易变成单链路的三到五倍。

延迟也是一个问题。虽然并行能缓解一部分，但每个智能体都有固定的模型推理开销。如果主管拆出来的多个子任务是串行依赖的，延迟会直接叠加。我遇到过一个案例，某次任务拆成了四层链式依赖，单次请求从两秒变成四十秒，用户直接放弃等待。后来通过合并无关步骤、减少链路层级，才把延迟压回可以接受的范围。

稳定性上，每多一个智能体，就多一个输出不可控的节点。我的对策是给每个智能体的关键输出加结构化约束，比如计划必须输出JSON，结果必须带来源摘要，失败必须显式返回异常码。结构化约束会牺牲一点灵活性，但换来的是可解析、可回滚、可兜底。

6.3 可观测性：每次调用都要能追溯

多智能体系统上线之后，最怕的不是效果差，是出了问题查不到根因。用户说“你回答错了”，如果没有日志，你连是哪一步错的都定位不出来。

所以我会在每次智能体运行时传一个共享的trace对象，记录每个智能体的输入目标、调用的工具、工具返回摘要、模型输出片段、耗时和Token消耗。日志不需要很复杂，能回答“谁在什么时候调用了什么工具、拿到了什么结果”就够了。这个小习惯，让我在一次大型线上故障里快速定位到是某个第三方知识工具返回了过期数据，而不是生成智能体的问题，排查时间从小时级压缩到了十分钟以内。

关于评估，我把离线回归当作硬性门槛。换模型、换Embedding、调分块参数、增删工具，任何影响到链路的改动，都必须先跑一遍离线评估集，通过率不达标就直接拒绝上线。这套流程看起来朴素，但它保证了整个系统在多次迭代后不会越改越差。

最后说一点个人心得。大模型应用开发走到后面，真正拼的不是会用多少框架、调得动多少模型，而是能不能把数据、检索、生成、评估、观测这五件事形成闭环。我也见过一些做得非常酷炫的架构，最后因为无法维护而被推倒重来。反过来，一套看似朴素的RAG加轻量多智能体体系，如果你能长期坚持评估、坚持追溯、坚持守住“不编造、有依据、可解释”这三点，它带来的实际价值往往远超大多数炫技方案。

如果你正在规划一个类似的项目，我建议从最小闭环开始：先跑通一套RAG，维护好二十条黄金评估样例，再加上一个工具调用的智能体循环。等这层地基稳了，再去考虑多智能体编排。这个顺序能帮你避开我踩过的大多数坑。