企业官网建设流程全解析

开门见山，这篇文章想聊聊我最近独立做完的一个“多智能体写作团队”项目。不是 PPT 上的架构图，也不是为了炫技硬凑的 Demo，而是一个真正用 LangGraph 跑起来、能自动完成“研究、分析、写作、审核”全流程的协作系统。我把完整实现拆开揉碎写出来，包括核心设计、代码细节、中间踩过的问题以及最后的调优方案。如果你想落地多 Agent 项目，又对状态流控制、多角色协作和自动反馈循环感兴趣，这篇应该能帮你少走不少弯路。

先说结论：LangGraph 这类框架，最值钱的不是帮你调大模型接口，而是把“多个角色协作”这件事变成了可控的图状态流转。你不需要自己去维护复杂的状态机，也不需要手工编排漫天的 prompt 调用。只要定义好每个 Agent 的职责，再用图把它们串起来，系统自己就能按流程往下走。但对于初学者，最容易被绕晕的就是“节点”“边”“状态”这三者的关系。很多人一开始都会写一堆互相调用的函数，最后根本没有发挥出图编排的优势。

1. 多智能体协作：为什么工作流需要的是“团队”，而不是单个 Agent

先说动机。如果你只是让一个大模型写一篇几百字的文案，那单独用 Prompt 就够了，没必要搞多智能体。但一旦任务变成“要写一份完整的行业分析报告”，或者“要生产一篇带观点、带数据、需要层层审核的长文”，单个 Agent 的劣势就非常明显。

第一是上下文窗口不够用。你让一个 Agent 既去搜资料、又整理数据、还要写长篇大论，中段很容易就把上下文塞满。第二是质量不可控。写出来的内容可能第一次还行，第二次就彻底跑偏，因为没有任何校验环节去纠偏。第三是职责不分。搜索要用工具，分析要做计算，写作要做表达，审核要做判断。这些混合在一个系统提示词里，效果往往是灾难性的。

“研究-分析-写作-审核”这套团队模型，本质上就是把人工作业的标准流程搬到了大模型世界里。研究 Agent 负责收集素材和事实，分析 Agent 负责提炼关键数据和结构，写作 Agent 负责生成通顺有观点的内容，审核 Agent 负责最后把关并给出修改意见。一旦审核不通过，自动把意见反馈给写作 Agent，让它重写。这样，整个生产过程就形成了一个可控的闭环。

那为什么选 LangGraph？市面上其实也有一些多 Agent 工具，但很多封装太重，调度逻辑不透明，出了问题很难排查。LangGraph 的好处在于它把整个流程显式地定义成了“图”：节点（Node）就是 Agent 或工具，边（Edge）就是流转规则，全局状态（State）则是所有节点共享的信息总线。你一眼能看出来数据是怎么流动的，逻辑在哪儿分叉，在哪儿循环。这套设计非常符合程序员直觉，也方便做精细化调整。

2. 团队分工设计：研究、分析、写作、审核的功能与状态流

整个项目我起了一个内部代号，叫“某跨平台写作工作台”。听起来挺大，其实核心就是一个图，上面挂了四个 Agent 节点。下面把这几个角色的职责边界先拆开说清楚。

2.1 核心状态结构：Agent 之间靠什么通信

LangGraph 里的多智能体协作，不是让四个 Agent 直接对话，而是让它们读写同一份全局状态。我把状态定义成下面这样，这是整个设计的地基：

from typing import TypedDict, List, Annotated, Optional import operator class WorkflowState(TypedDict, total=False): # 原始任务描述 task: str # 研究阶段产出 research_notes: List[str] # 分析阶段产出 key_points: List[str] data_stats: List[str] # 写作阶段产出 draft_article: Optional[str] # 审核阶段产出 review_comments: List[str] review_passed: Optional[bool] # 记录迭代次数，防止死循环 iteration_count: int # 历史错误或修改意见 revision_history: Annotated[List[str], operator.add]

我要特别说明一下revision_history这个字段。在 LangGraph 里面，普通字段是覆盖写，后写的节点会把前一个节点的值直接替换掉。但有些历史信息我们希望保留，所以用了Annotated[List[str], operator.add]，这是 LangGraph 提供的一个技巧：新值不是覆盖旧值，而是往列表里追加。这样一来，审核 Agent 每次提出的修改意见，都会原样存在状态里，写作 Agent 下一次重写时能看到完整的历史上下文。这个设计在迭代修改场景里非常关键，否则每次审核意见都会被冲掉，Agent 根本不知道上一篇为什么被退回。

2.2 研究节点：资料搜集与事实沉淀

研究节点的任务是给后续所有环节提供弹药。我给它挂了一个联网搜索工具，输入是任务主题，输出是一组结构化的研究笔记。

from langchain_core.tools import tool from langchain_community.tools import DuckDuckGoSearchRun @tool def search_web(query: str) -> str: """搜索最新的行业信息""" search = DuckDuckGoSearchRun() return search.run(query) async def research_node(state: WorkflowState): task = state["task"] queries = generate_search_queries(task) all_notes = [] for q in queries: raw_result = await search_web.ainvoke({"query": q}) all_notes.append(f"来源：{q}\\n内容摘要：{raw_result[:1000]}") return {"research_notes": all_notes}

这里有个细节：research_notes字段我定义成了普通List[str]，但在节点返回值里用了完整列表覆盖，而不是增量追加。为什么？因为研究节点只运行一次，它需要把新研究中获取的内容完整地写入状态，负责“源头控制”。这个过程实际上就是收集原始素材的“弹药库”。

2.3 分析节点：从资料中提炼骨架

分析节点的输入是research_notes，输出是key_points和data_stats。它的职责不是写最终内容，而是做信息降维：把一堆搜索结果里重复的、无效的去掉，提炼出五到六个核心要点，顺便整理出可以引用的关键数据。

为了让分析结果更加结构化，我规定分析节点必须输出 JSON 格式，避免大模型在中间层生成一些乱七八糟的格式。

from langchain_core.prompts import ChatPromptTemplate analysis_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个资深行业分析师。请根据研究资料，提炼出写作大纲。 要求： 1. 只输出 JSON，不要任何额外文字。 2. JSON格式为 {{"key_points": ["要点1", "要点2"], "data_stats": ["数据或事实1", "数据或事实2"]}} 3. 每个要点必须可以在研究资料中找到出处。 """), ("user", "研究资料：{research_notes}") ]) async def analysis_node(state: WorkflowState): research_notes = state["research_notes"] result = await llm.ainvoke(analysis_prompt.format(research_notes="\\n".join(research_notes))) parsed = json.loads(result.content) return {"key_points": parsed["key_points"], "data_stats": parsed["data_stats"]}

这一步容易犯的错是把所有原始资料直接丢给分析 Agent。有些资料的长度远超预期，在分析阶段就触发上下文超限，后面写作和审核环节基本就废了。我试过几次失控之后，直接在研究节点就对单条结果做了截断（代码里raw_result[:1000]就是这个用途），这样进入分析阶段的输入始终处在可控范围。

分析节点的输出是关键。因为后续写作 Agent 不需要再看原始搜索记录，只需要看key_points和data_stats，这样就把上下文窗口的压力大大减轻了。真正的多智能体协作并不是信息越多越好，而是每一层只传递下一层需要的最小必要信息。

2.4 写作节点：基于要点生成连贯内容

写作节点拿到 key_points 和 data_stats 之后，开始生成正式文章。这里我用了结构化输出：

from pydantic import BaseModel, Field class ArticleResponse(BaseModel): title: str = Field(description="文章标题") content: str = Field(description="正文内容，使用Markdown格式") writing_prompt = ChatPromptTemplate.from_messages([ ("system", """你是一位资深写作专家。请基于给定要点，撰写一篇深度长文。 要求： - 开头吸引人，中间论证扎实，结尾有简洁收束。 - 必须使用提供的所有要点，并自然融入。 - 需要扩展细节，但不能编造研究资料中不存在的事实。 - 只输出 JSON，格式为 {{"title": "标题", "content": "正文"}} """), ("user", "写作要点：{key_points}\\n关键数据：{data_stats}\\n\\n迭代意见：{revision_history}") ]) async def writing_node(state: WorkflowState): key_points = state["key_points"] data_stats = state["data_stats"] revision_history = state.get("revision_history", []) result = await llm.ainvoke(writing_prompt.format( key_points="\\n".join(key_points), data_stats="\\n".join(data_stats), revision_history="\\n".join(revision_history) if revision_history else "无" )) parsed = ArticleResponse.model_validate_json(result.content) return {"draft_article": parsed.content, "iteration_count": state.get("iteration_count", 0) + 1}

注意这个迭代意见的传递。第一次写作时revision_history是空的，所以写作节点不会受历史干扰。第二次重写时，它能看到上一次审核 Agent 的具体意见。这个机制是整个闭环真正能用的关键。

2.5 审核节点：结合评价标准与自动路由

审核节点负责做质量门禁。它不直接修改文章，而是输出“通过/打回”的判断。如果打回，则必须附带修改意见。

class ReviewResult(BaseModel): passed: bool = Field(description="是否通过审核") comments: List[str] = Field(description="修改意见列表，未通过时必须提供") review_prompt = ChatPromptTemplate.from_messages([ ("system", """你是严格的编辑审核专家。请从以下维度审核文章： 1. 逻辑结构是否连贯。 2. 论证是否使用了提供的要点和数据。 3. 是否有明显的事实编造。 4. 语言是否专业、可读。 只输出 JSON：{{"passed": true/false, "comments": ["意见1", "意见2"]}} """), ("user", "待审文章：{article}\\n审核要点：{key_points}") ]) async def review_node(state: WorkflowState): article = state["draft_article"] key_points = state["key_points"] result = await llm.ainvoke(review_prompt.format(article=article, key_points="\\n".join(key_points))) parsed = ReviewResult.model_validate_json(result.content) return { "review_passed": parsed.passed, "review_comments": parsed.comments }

审核节点的设计有一个需要注意的地方：审核 Agent 可能会受到写作 Agent 风格的干扰。如果它只看到了烂文章，它可能也会被带进去，标准会降低；如果它看到了过高的标准，它又会过度苛刻。为了解决这个问题，我把审核提示词里明确限定了四个维度，让它必须按维度打分，而不是凭感觉判断。

3. 环境准备与基础框架搭建

这块比较基础，但还是要提一下，因为 LangGraph 的版本迭代很快，API 变动很频繁。我从零开始搭的时候，就是因为版本不一致浪费了不少时间。

3.1 软件包安装

我使用的核心依赖如下：

pip install langgraph langchain langchain-core langchain-community langchain-openai

需要提醒一下，langchain-community里的工具包有很多历史遗留问题，一些搜索工具的 import 路径已经变了。如果你在加载搜索模块时遇到报错，建议去官方文档确认工具的路径和名称，不要直接复制老文章里的代码。这个坑我在 1.1 节也列了。

3.2 基础环境变量配置

我用的是兼容 OpenAI 协议的大模型接口。主要配置三个环境变量：

export LLM_BASE_URL="https://api.某模型平台.com/v1" export LLM_API_KEY="你的密钥" export LLM_MODEL="某大模型-16k"

注意，我在测试不同模型时，发现有些模型在JSON mode上表现不行，经常输出一些额外解释文本。所以我在提示词里都加了“只输出 JSON”的强调，并且要求使用response_format参数做兜底。下面这段代码是实际初始化大模型的地方：

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model=os.environ["LLM_MODEL"], api_key=os.environ["LLM_API_KEY"], base_url=os.environ["LLM_BASE_URL"], temperature=0.2, response_format={"type": "json_object"} )

设置temperature=0.2是常用技巧，让研究、分析、审核这类需要稳定性的节点输出结果更偏向保守与一致；写作节点的 temperature 则可以调高一些，比如 0.6，给语言表达留点创造空间。不过受限于共用同一个 LLM 实例，我通常还是会倾向于让整体温度较低，来保持流程的稳定。

4. 核心代码实现：从“零散调用”到“图状态编排”

这一节是重点。上面我们把四个节点函数写完了，但如果不把它们放到图里，它们只是四个互不相关的普通函数。LangGraph 的魔力在于把这些函数组织成一张网络，且带条件路由和循环。

4.1 构建基础工作流图

直接上代码：

from langgraph.graph import StateGraph, START, END workflow = StateGraph(WorkflowState) # 添加节点 workflow.add_node("research", research_node) workflow.add_node("analysis", analysis_node) workflow.add_node("writing", writing_node) workflow.add_node("review", review_node) # 添加入口 workflow.add_edge(START, "research") workflow.add_edge("research", "analysis") workflow.add_edge("analysis", "writing") workflow.add_edge("writing", "review")

到这里，图的骨架已经有了，还缺最关键的一步：审核之后怎么走。如果review_passed为True，直接走到 END；如果为False，路由回“writing”节点，让它根据意见修改。但我们需要设置一个“迭代次数上限”，不让它无限循环。

4.2 条件路由：让图具备“自我纠错”能力

MAX_ITERATIONS = 3 def should_continue_after_review(state: WorkflowState) -> str: if state.get("review_passed", False): return "end" if state.get("iteration_count", 0) >= MAX_ITERATIONS: return "end" return "rewrite" workflow.add_conditional_edges( "review", should_continue_after_review, { "rewrite": "writing", "end": END } ) app = workflow.compile()

这就是 LangGraph 最核心的设计。传统方式下，如果要实现这个流程，你得自己写while循环，维护 index，并且手动整合最新数据；但在 LangGraph 里，条件函数should_continue_after_review接收全局状态，然后返回字符串，图就能自动切换到对应的边。这样看问题的方式从“流程控制”视角转变成了“状态路由”视角，代码的可读性和可维护性都高了很多。

4.3 控制流中最重要的迭代计数

有人可能会问，iteration_count在哪加？答案是写作节点里加。因为每重写一次，我们就认为多了一轮迭代。如果在审核节点加，就会多算一次。所以写作节点开头这一行也很关键：

return {"draft_article": parsed.content, "iteration_count": state.get("iteration_count", 0) + 1}

配合MAX_ITERATIONS = 3，整个系统最多允许三次重新写作。超过这个数，即使审核不通过，流程也会强制结束并输出当前带问题的草稿。这样做的目的很直接：防止某个问题导致 Agent 陷入无限循环，给你的 API 账单造成不必要的压力。

4.4 正式运行工作流

图编译完成后，我们用一个具体任务来测试整个流程。假设现在我们要写一篇关于某个行业趋势的分析文章，输入的任务描述如下：

def run_agent_team(task_description: str): config = {"recursion_limit": 20} initial_state = { "task": task_description, "research_notes": [], "key_points": [], "data_stats": [], "draft_article": None, "review_comments": [], "revision_history": [], "iteration_count": 0, } result = app.invoke(initial_state, config=config) return result

于是我们用下面的命令跑一次：

task_desc = "调研智能协作工具在内容生产中的应用现状，写一篇深度分析文章" result = run_agent_team(task_desc) print(result["draft_article"])

第一次跑的时候，我用的正好是上面这套代码。整体输出流程是能走通的，大约花了四十多秒，生成了 2000 字左右的文章。注意recursion_limit这个参数，它控制图的深度。如果图里有循环，这个参数不设高一点，在迭代次数达到限制前，图可能会因为递归层级超限而报错。我设成 20，能确保迭代次数 3 的循环深度够用。

5. 实战调整与效果提升：从“能用”到“好用”

现在系统能跑通了，但我们要的是能够实际交付的高质量输出。从第一版能用的 Demo 到真正稳定的团队，中间有几个比较重要的调优路径。

5.1 审核反馈的应用：避免“一稿改到底”

早期版本里，写作节点在重写时只看最新一次的review_comments，不看历史意见。结果经常出现“修改了 A 问题，又忽略了 B 问题”的情况。后来我引入了revision_history，并且让它在写作节点里拼接呈现，效果明显变好。这就是多轮迭代的价值所在。

5.2 节点级 prompt 提炼

节点之所以重要，是因为不同角色对同一个任务的关注点是不同的。你不能让研究 Agent 直接生成文章，也不能让审核 Agent 只看文章结构。因此给每个节点都设置了独特的系统提示词，而不是共用一个。下面是经过调优后的几个节点提示词设计，它们在分工上体现得比较明显：

5.3 降低多次调用成本

因为循环的存在，同一个任务可能调用多次大模型接口。如果你的任务比较简单，直接跑三到四次大调用显然成本过高。我的建议是设置更低的迭代上限，比如尝试一次，没有通过就直接把产物转发到人工编辑，而不是让 Agent 无限修改。这个方案对成本控制很管用。

5.4 将写作与审核节点配置独立

同一个模型同时承担写作和审核，推理方式的差异会影响到最终结果。某些情况下，写作节点喜欢发挥想象，审核节点就需要特别强调“不要编造事实”。如果模型支持不同的部署，建议给写作节点配置一个长上下文、高创造性的模型；给审核节点配置一个更强逻辑判断的模型。这种隔离设计能解决许多提示词层面解决不了的问题。

6. 踩坑记录与排查方法：给后来者的避坑指南

任何真实项目都离不开踩坑。这一节我整理了六个我在项目开发中遇到的高频问题，以及排查方案，希望能帮大家节省时间。

6.1 状态字段被覆盖，导致历史数据丢失

刚写完第一版时，我发现审核意见总是只有最新的那条，历史意见全丢了。排查下来发现是忘了在WorkflowState中使用Annotated[List[str], operator.add]进行累加设置。这里有一个细节：operator.add只对已经存在的字段生效，字节码中运行顺序不同，有时会出现覆盖。解决方式是保证所有会产生增量信息的字段都明确加上这个注解。

6.2 搜索工具失效或返回异常

研究节点的搜索工具，如果访问频繁或受平台风控限制，容易抛出异常。一旦异常没有被捕获，整个流程直接中断，后面的组装和无菌操作全部作废。我在搜索工具的外围包裹了try...except，并且还加了重试逻辑：

async def safe_search(query: str, retries: int = 2): for i in range(retries): try: return await search_web.ainvoke({"query": query}) except Exception as e: if i == retries - 1: return f"搜索失败：{e}" await asyncio.sleep(2)

这样即使搜索失败，研究节点也能返回一条占位信息，而不是让整个图崩溃。

6.3 结构化输出解析失败

在使用 JSON 输出时，经常遇到模型输出多余的前缀，好比“这是我的回答：”或者 Markdown 的代码块包围。虽然我在 prompt 里写了“只输出 JSON”，但模型偶尔会手误。史训就是在解析 JSON 前做一层清理。这个清理不是 “Replace 掉大括号”，而是截取第一个{到最后一个}之间的内容再解析：

def extract_json(text: str): start = text.find("{") end = text.rfind("}") if start == -1 or end == -1: raise ValueError("No JSON found") return json.loads(text[start:end+1])

这一招在实战中救了我很多次，模型怎么着都不会完全忽略这个提取规则。

6.4 单个 Agent 上下文撑爆

我第一次写研究节点时，把整篇搜索摘要都放进了提示词。结果研究结果一多，加上其他节点还要引用研究资料，很快就超过了上下文的承载能力。

处理方式是分两层：先限制单个搜索结果长度，再在分析节点完成信息压缩。这个设计属于典型的分治策略。搜索资料再丰富，经过压缩后只剩核心要点，后面所有 Agent 的输入都能保持在一个合理的长度。

6.5 路由条件判断过于依赖模型输出

之前我把审核结果写成了一段自然语言，比如“可以发布”或者“建议修改”，然后让代码通过关键词判断。这种方式极不稳定。后来我改成用pydantic模型来做结构化输出，让passed字段枚举布尔值，代码只做硬判断，不再依赖字符串匹配。

6.6 调试排查的基本方法

如果你运行时报错，不知道该从哪查起。LangGraph 提供了一些内置的 debug 工具，比如运行时打印关键信息：

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() debug_app = workflow.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "debug_thread"}} result = debug_app.invoke(initial_state, config=config) print(debug_app.get_state(config))

使用 MemorySaver 能检查图中每一步的状态快照。这样当遇到问题时，你可以看到某个节点执行完后的完整 state，定位是哪个节点没有正确更新字段，还是哪个边上条件判断出了错。

结尾：一点个人的实践体会

在折腾完这个项目之后，我最大的感受是，多智能体协作的真正难点，不在于把四个 Agent “接”在一起，而在于定义它们之间高效的通信协议与边界职责。如果只是从一个 Agent 的返回结果里抽出文本，拼接到另一个 Agent 的输入里，那本质上还是流水线函数，不是多智能体系统。

LangGraph 的价值是给了你一张“地图”。你可以在图上看到每个节点的输入输出，控制它们可以循环、可以跳过、可以并行。当任务需要高质量和迭代修改时，这会带来质的改变。这次做研究-分析-写作-审核，最直观的收获是输出稳定了很多，不会再出现第一版能看、第二版就完全放飞的情况。后面我想再接一个偏事实核查的 Agent，专门去验证文中引用的数据是否来自可信源，把这套团队的最后一块短板补上。