企业官网建设流程全解析

1. 多Agent协作到底在解决什么问题

单Agent跑不通复杂任务，这是我做了大半年Agent开发后最深的体会。你让一个Agent去完成“分析一份财报、生成摘要、翻译成英文、再做成PPT大纲”这种链路，它大概率会在第三步开始胡言乱语，或者干脆忘记第一步拿到的数据。原因不复杂——单Agent的上下文窗口是有限的，工具调用能力是有限的，而且它没有“第二双眼睛”帮它检查错误。

多Agent协作的核心思路就一句话：把一个大任务拆成多个子任务，每个子任务交给专门的Agent，再通过一套调度机制把它们串起来。这跟人类团队干活是一个道理——你不会让一个人同时做财务分析、翻译和设计，而是分给不同角色，最后有人负责汇总。

这套东西能做什么？举几个我实际接触过的场景。第一个是研究报告生成：一个Agent负责检索资料，一个负责数据分析，一个负责撰写，最后一个负责质量校准。第二个是代码审查流水线：一个Agent写代码，一个Agent跑测试，一个Agent做安全审查，还有一个负责合并建议。第三个是客服工单处理：分类Agent先判断工单类型，然后路由给对应的处理Agent，复杂工单再升级给人工。

适合谁来参考？如果你已经用过大模型API，写过简单的Prompt，想从“单次对话”升级到“多步骤自动化”，那这篇内容就是给你准备的。如果你还没碰过Agent，建议先把工具调用（Function Calling）和ReAct模式搞清楚，再来看协作架构，不然容易一头雾水。

注意：多Agent不是银弹。任务越复杂，Agent数量越多，通信开销和出错概率也会上升。我见过有人上来就搞七八个Agent，结果调试了两周还没跑通一条完整链路。建议从两个Agent开始，跑通了再加。

2. 协作架构的几种主流模式与选型逻辑

2.1 顺序流水线：最简单也最常用

顺序流水线就是Agent A做完交给Agent B，B做完交给C。这种模式的好处是逻辑清晰、调试容易，每个Agent的输入输出都是确定的。坏处是没有反馈回路，如果A的输出质量差，B和C只能跟着错。

我一般用这种模式处理“步骤固定、依赖明确”的任务。比如前面说的报告生成：检索→分析→撰写→校对，每一步的输入就是上一步的输出，不需要来回沟通。

实现上，你可以用一个简单的列表来定义流水线：

pipeline = [ {"name": "retriever", "prompt": "根据主题检索相关资料并整理成结构化数据"}, {"name": "analyzer", "prompt": "对结构化数据进行统计分析，提取关键指标"}, {"name": "writer", "prompt": "根据分析结果撰写报告初稿"}, {"name": "reviewer", "prompt": "检查报告的逻辑性和数据准确性，输出修改建议"} ] context = {"topic": "某行业季度趋势"} for step in pipeline: result = call_llm(step["prompt"], context) context[step["name"] + "_output"] = result

这段代码的关键在于context字典——它像一个共享白板，每个Agent往上面写自己的输出，后面的Agent从上面读需要的信息。实际用的时候，你需要在每个Prompt里明确告诉Agent“你可以从context里拿到哪些字段”。

2.2 层级调度：主管Agent分配任务

层级调度模式里有一个“主管Agent”，它负责理解用户需求、拆解任务、分配给下面的“工人Agent”，最后汇总结果。这种模式适合任务类型不固定、需要动态决策的场景。

举个例子，用户说“帮我分析一下这份销售数据，看看哪个区域表现最好，再预测下个季度的趋势”。主管Agent需要判断：这涉及数据分析和趋势预测两个子任务，然后分别调用对应的Agent。

主管Agent的Prompt设计是关键。我通常这样写：

你是一个任务调度主管。你的职责是： 1. 理解用户需求，判断需要哪些子任务 2. 将子任务分配给对应的工人Agent 3. 收集所有工人的输出，整合成最终答案 可用的工人Agent： - data_analyst: 擅长数据统计、对比分析 - trend_predictor: 擅长基于历史数据做趋势预测 - report_writer: 擅长将分析结果整理成可读报告 输出格式要求：先列出任务拆解，再逐个调用工人Agent，最后汇总。

这种模式的风险在于主管Agent可能拆错任务。我踩过的坑是：主管把“预测趋势”拆成了“计算平均值”，因为它没理解“预测”和“统计”的区别。解决办法是在主管的Prompt里加几个Few-shot示例，明确什么任务该分给谁。

2.3 辩论与投票：多个Agent互相校验

这种模式让多个Agent对同一个问题给出答案，然后通过投票或辩论选出最优解。适合需要高准确性、容错率低的场景，比如事实核查、代码安全审查。

我做过一个实验：让三个Agent分别判断一段代码是否有SQL注入风险，然后取多数意见。结果发现，单个Agent的准确率是78%，三个Agent投票后提升到了91%。代价是Token消耗翻了三倍，响应时间也变长了。

辩论模式更复杂一些：Agent A提出观点，Agent B反驳，Agent C做裁判。这种模式我一般只在争议性大、需要多角度分析的任务里用，比如政策解读、竞品分析。

2.4 选型对照表

选型的时候，我一般问自己三个问题：任务步骤是否固定？需不需要动态决策？错误容忍度有多高？三个问题的答案基本就能确定用哪种模式。

3. 任务调度的核心机制与实现细节

3.1 任务拆解：从自然语言到可执行单元

任务拆解是多Agent协作里最容易被低估的环节。很多人以为“让大模型拆一下就行了”，实际上拆解的质量直接决定了后续所有环节的成败。

我用的拆解策略是两层拆解：第一层是“意图识别”，判断用户到底想要什么；第二层是“步骤生成”，把意图转化成具体的执行步骤。

意图识别可以用一个轻量级的Prompt：

分析用户输入，判断属于以下哪类意图： - 信息检索：用户想要查找某些事实或数据 - 分析计算：用户想要对数据进行统计或推理 - 内容生成：用户想要创作文章、报告、代码等 - 多步复合：用户需求包含以上多种类型 用户输入：{user_input} 输出格式：{"intent": "意图类型", "confidence": 0.95}

步骤生成则要结合可用Agent的能力列表：

根据意图和可用Agent，生成执行步骤： 可用Agent： - search_agent: 联网检索 - calc_agent: 数据计算 - write_agent: 内容撰写 意图：多步复合 用户需求：分析某公司财报并写摘要 输出： [ {"step": 1, "agent": "search_agent", "task": "检索该公司最新财报数据"}, {"step": 2, "agent": "calc_agent", "task": "计算营收增长率、利润率等关键指标"}, {"step": 3, "agent": "write_agent", "task": "根据指标撰写200字摘要"} ]

实操心得：步骤生成的时候，一定要让大模型输出结构化的JSON，不要输出自然语言。自然语言的步骤描述在后续解析时容易出歧义，JSON虽然看起来死板，但稳定得多。

3.2 状态管理：让每个Agent知道“现在到哪了”

多Agent协作最头疼的问题之一是状态同步。Agent A完成了任务，Agent B怎么知道？Agent B执行到一半失败了，怎么回滚？

我的做法是维护一个共享状态对象，所有Agent都读写这个对象。状态对象里至少包含这几个字段：

shared_state = { "task_id": "唯一标识", "current_step": 2, "total_steps": 4, "step_results": { "step_1": {"status": "completed", "output": "..."}, "step_2": {"status": "running", "output": None} }, "errors": [], "metadata": {"created_at": "...", "updated_at": "..."} }

每个Agent在执行前先读current_step，知道自己该做什么；执行后更新step_results和current_step。如果出错，就往errors里追加一条记录，调度器根据错误类型决定是重试还是跳过。

这种设计的好处是可追溯。出了问题，你看一眼step_results就知道哪个环节卡住了，不用去翻日志。

3.3 通信协议：Agent之间怎么说话

Agent之间的通信有两种方式：共享内存和消息传递。

共享内存就是前面说的共享状态对象，适合同一进程内的Agent协作。消息传递则是Agent之间发消息，适合分布式部署的场景。

我大多数项目用的是共享内存，因为实现简单、延迟低。但如果Agent数量超过五个，或者需要跨机器部署，就会换成消息队列。常用的消息格式是JSON：

{ "from": "analyzer_agent", "to": "writer_agent", "type": "task_result", "payload": { "analysis": "...", "confidence": 0.87 }, "timestamp": "2024-01-15T10:30:00Z" }

消息传递的坑在于消息丢失和重复。我遇到过Agent B没收到Agent A的消息，导致整个流水线卡死。解决办法是加确认机制：接收方收到消息后回一个ACK，发送方在超时后重发。

3.4 超时与重试：别让一个Agent拖垮整个系统

大模型API的响应时间不稳定，有时候几秒，有时候几十秒。如果不设超时，一个慢请求可能把整个流水线堵死。

我的配置是：单个Agent调用超时30秒，重试2次，重试间隔5秒。超过重试次数就标记为失败，调度器决定是跳过还是终止。

import time def call_agent_with_retry(agent, input_data, max_retries=2, timeout=30): for attempt in range(max_retries + 1): try: result = agent.run(input_data, timeout=timeout) return {"status": "success", "output": result} except TimeoutError: if attempt < max_retries: time.sleep(5) continue return {"status": "timeout", "output": None} except Exception as e: return {"status": "error", "output": str(e)}

注意：重试不是万能的。如果Agent是因为输入数据有问题而失败，重试只会浪费Token。我一般会在重试前判断错误类型——网络超时可以重试，数据格式错误直接跳过。

4. 完整实操：搭建一个四Agent协同的研究报告生成系统

4.1 系统架构与Agent角色定义

这个系统包含四个Agent：检索Agent、分析Agent、撰写Agent、校对Agent。整体走顺序流水线模式，但在校对环节加了一个反馈回路——如果校对不通过，打回给撰写Agent重写。

每个Agent的角色定义如下：

4.2 检索Agent的实现细节

检索Agent的核心是查询生成和结果过滤。用户给一个主题，Agent需要生成多个检索查询，然后从返回结果里筛选出相关的。

RETRIEVER_PROMPT = """ 你是一个研究资料检索专家。根据用户提供的主题，生成3-5个检索查询， 每个查询从不同角度覆盖主题。 主题：{topic} 输出格式： { "queries": ["查询1", "查询2", "查询3"], "focus_areas": ["重点领域1", "重点领域2"] } """ def retriever_agent(topic): # 第一步：生成查询 queries = call_llm(RETRIEVER_PROMPT.format(topic=topic)) # 第二步：执行检索（这里用模拟的检索函数） raw_results = [] for q in queries["queries"]: results = search_engine.search(q, top_k=5) raw_results.extend(results) # 第三步：过滤和去重 filtered = filter_by_relevance(raw_results, topic, threshold=0.7) deduplicated = deduplicate_by_url(filtered) return { "topic": topic, "sources": deduplicated[:10], "focus_areas": queries["focus_areas"] }

这里的关键参数是threshold=0.7——相关性低于0.7的结果直接丢掉。这个阈值是我试了好几次才定下来的，太低会引入噪音，太高会漏掉有用信息。

4.3 分析Agent的数据处理流程

分析Agent拿到检索结果后，需要提取关键数据点、计算指标、识别趋势。我一般让它输出结构化的JSON，方便后续Agent解析。

ANALYZER_PROMPT = """ 你是一个数据分析专家。根据以下资料，提取关键数据点并进行分析。 资料： {sources} 分析要求： 1. 提取至少5个关键数据点，每个数据点包含数值、来源、时间 2. 计算同比/环比变化率（如果有历史数据） 3. 识别至少2个趋势或模式 4. 标注数据可信度（高/中/低） 输出格式： { "key_metrics": [ {"name": "指标名", "value": "数值", "source": "来源", "confidence": "高"} ], "trends": [ {"description": "趋势描述", "evidence": "支撑数据"} ], "data_quality": "整体数据质量评估" } """

实操心得：分析Agent最容易犯的错误是“编数据”。如果资料里没有某个指标，它可能会根据常识推测一个数值。解决办法是在Prompt里明确写“如果资料中没有相关数据，标注为‘数据缺失’，不要推测”。

4.4 撰写Agent与校对Agent的反馈回路

撰写Agent根据分析结果写报告，校对Agent检查逻辑性、数据准确性和可读性。如果校对不通过，撰写Agent根据反馈修改，最多循环三次。

def writer_reviewer_loop(analysis_result, max_iterations=3): draft = writer_agent(analysis_result) for i in range(max_iterations): review = reviewer_agent(draft) if review["status"] == "approved": return {"report": draft, "iterations": i + 1} # 根据反馈修改 draft = writer_agent(analysis_result, feedback=review["feedback"]) return {"report": draft, "iterations": max_iterations, "warning": "达到最大迭代次数"}

校对Agent的Prompt里，我会明确列出检查项：

检查以下报告，逐项判断是否通过： 1. 数据准确性：报告中的数据是否与分析结果一致？ 2. 逻辑连贯性：段落之间是否有清晰的逻辑关系？ 3. 可读性：语言是否通顺，有没有明显的语病？ 4. 完整性：是否覆盖了所有关键分析点？ 输出格式： { "status": "approved/rejected", "issues": ["问题1", "问题2"], "feedback": "具体的修改建议" }

4.5 调度器的完整代码框架

把上面四个Agent串起来，调度器的核心逻辑如下：

class MultiAgentOrchestrator: def __init__(self): self.state = { "task_id": generate_id(), "current_step": 0, "results": {}, "errors": [] } def run(self, topic): try: # Step 1: 检索 self.state["current_step"] = 1 retrieval = retriever_agent(topic) self.state["results"]["retrieval"] = retrieval # Step 2: 分析 self.state["current_step"] = 2 analysis = analyzer_agent(retrieval["sources"]) self.state["results"]["analysis"] = analysis # Step 3: 撰写+校对循环 self.state["current_step"] = 3 final = writer_reviewer_loop(analysis) self.state["results"]["final"] = final return {"status": "success", "data": final} except Exception as e: self.state["errors"].append(str(e)) return {"status": "failed", "state": self.state}

这个框架跑下来，一个完整的研究报告生成大概需要45-90秒，消耗8000-15000个Token，具体取决于报告长度和校对迭代次数。

5. 踩坑记录与常见问题排查

5.1 Agent之间“踢皮球”怎么办

我遇到过最诡异的问题是：检索Agent返回了空结果，分析Agent说“没有数据无法分析”，撰写Agent说“没有分析结果无法撰写”，最后用户收到一个空报告。

排查后发现，检索Agent的查询生成有问题——它生成的查询太具体，搜索引擎返回了零结果。解决办法是加一个兜底逻辑：如果检索结果少于3条，自动放宽查询条件重新检索。

if len(filtered_results) < 3: # 放宽条件：去掉限定词，用更宽泛的查询 broad_query = extract_core_keywords(topic) raw_results = search_engine.search(broad_query, top_k=10) filtered_results = filter_by_relevance(raw_results, topic, threshold=0.5)

5.2 Token消耗失控的三种情况

多Agent系统的Token消耗是单Agent的3-5倍，如果不加控制，成本会很难看。我总结的三种失控情况：

第一种是上下文无限增长。每个Agent都把完整的历史记录传给下一个Agent，导致Prompt越来越长。解决办法是只传必要字段，不要传完整对话历史。

第二种是校对循环不收敛。撰写Agent和校对Agent来回改，改了十几次还没通过。解决办法是设最大迭代次数，超过就强制输出并标记警告。

第三种是重复检索。多个Agent都需要同一份数据，各自去检索了一遍。解决办法是加缓存层，相同查询直接返回缓存结果。

5.3 输出格式不稳定的处理技巧

大模型输出JSON的时候，偶尔会多一个逗号、少一个引号，导致解析失败。我试过三种解决办法：

第一种是用JSON Schema约束输出。现在很多大模型API支持response_format参数，直接指定JSON Schema，输出格式会稳定很多。

第二种是加解析容错。用正则表达式提取JSON部分，然后用json.loads的strict=False模式解析。

第三种是让模型自己修复。解析失败时，把错误信息和原始输出一起发给模型，让它重新输出正确的JSON。

def safe_json_parse(text): try: return json.loads(text) except json.JSONDecodeError: # 尝试提取JSON部分 match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group(), strict=False) except: pass # 让模型修复 fixed = call_llm(f"以下JSON格式有误，请修复：{text}") return json.loads(fixed)

5.4 常见问题速查表

最后分享一个小技巧：调试多Agent系统的时候，我会给每个Agent的输出加一个debug字段，记录它的输入摘要、输出摘要和耗时。这样出问题的时候，一眼就能看出是哪个环节的锅，不用去翻完整的日志。

这套多Agent协作架构我前后迭代了四个版本，从最初的两个Agent顺序流水线，到现在支持动态调度的混合模式。最大的体会是：不要追求一步到位。先把最简单的流水线跑通，再逐步加Agent、加反馈回路、加容错机制。每加一个东西，都要问自己“它解决了什么具体问题”，如果答不上来，就不加。