☰
基于LangGraph构建AI写作Agent:从链式调用到状态图编排
2026/10/21 8:09:10 网站建设 项目流程

1. 项目概述:写作 Agent 到底在解决什么问题

先说个最直观的感受:市面上的 AI 写作工具一抓一大把,但绝大多数是“一次性生成”,你给它一个 Prompt,它回你一篇文章。听起来没问题,但真正拿去做内容生产就会发现,这种模式有几个致命缺陷——生成的文章结构雷同、事实错误没人管、缺少可复用的逻辑、改起来要重新生成一遍。

这个项目的核心,是用 LangGraph 把一个“会写稿”的 AI 写作 Agent 搭出来。它不是一个 Chatbot,也不是一个 Prompt 模板,而是一套完整的写作流程编排系统。你给 Agent 一个选题,它会自己拆解需求、搜集素材、列提纲、写初稿、自我审校、再改写,最后交给人工确认。整个流程不是线性的,而是带分支、带循环、带人工干预的。这就是 LangGraph 的价值所在:它把复杂流程变成一张图,图中每个节点是 AI 的一次思考或动作,每条边决定下一步怎么走。

这个项目适合三类人:第一类是正在做内容中台的工程师,需要把多个模型的调用逻辑收敛到一套可维护的框架里;第二类是独立开发者,想做一个自动更新博客或自媒体选题的小工具;第三类是产品经理,理解这套思路后就知道 Agent 类产品和一次性 Prompt 的界限在哪里。整个项目用 Python 实现,代码量不大,但这套设计思想可以扩展到任意复杂的工作流。

2. 为什么是 LangGraph:从链式调用到状态图

2.1 传统 LangChain 链的局限

如果你之前用过 LangChain 的链式调用,大概会有这种感觉:链是固定的,A 完事进 B,B 完事进 C,中间想插一个“如果 A 的结果不好就重跑 A”的条件分支,要么写死,要么用外部代码控制,非常别扭。

写作这件事天然不是一条直线。初稿不满意要循环改写,素材不够要回头补检索,这些在链式结构里表达起来特别费劲。LangGraph 的核心抽象是“图”而不是“链”,节点之间的边可以带条件,你完全可以把“自评不达标就回炉重写”画成一条从“审校节点”指向“起草节点”的循环边,代码表达力和流程可视化程度完全不一样。

2.2 无循环不行:为什么写作必须带循环

一次性生成的文章,坦白说能看,但距离“能用”往往有差距。专业写手会自己审校、删改、调整结构,Agent 要复现这个过程,必须给模型“回头看”的机会。

我的方案是在图里塞两条循环:

  • 自评循环:初稿完成后,Agent 用独立的审校节点给文章打分,低于某个阈值就带着修改意见重新走一次起草节点。注意这里要用“独立审校”而不是让同一个模型又写又评,否则很容易出现“自己夸自己”的假阳性。
  • 素材补充循环:当检索节点发现某个必要信息缺失时,边会指回素材收集节点,而不是硬着头皮继续写。这模拟了人写稿时“先去查资料,发现不够再查一次”的行为。

这两条循环一旦通过图结构跑通,效果非常明显:生成的文章质量稳定,不会忽好忽坏,而且每一次改动都是有依据的,不是随机采样。

2.3 状态管理:所有节点的共享缓存

LangGraph 最核心也最容易被忽略的概念是 State(状态)。你可以简单理解成一张所有节点都能读写的共享黑板。每个节点函数接收的是整个 State,返回的是 State 的局部更新。

写作 Agent 的状态我命名为 WritingState,它的定义大致长这样:

from typing import TypedDict, Annotated, List, Optional from langgraph.graph import StateGraph, START, END class WritingState(TypedDict): topic: str audience: str tone: str required_keys: List[str] draft: Optional[str] revised: Optional[str] final_text: Optional[str] evaluation_score: Optional[int] evaluation_feedback: str source_materials: List[str] style_reference: Optional[str]

这里最需要注意的设计决策是:为什么把 topic、draft、revised 这些字段都放在同一个状态里,而不是每步单独传参?

这是因为图节点本身是无状态的,节点与节点之间的信息交换全部依赖 State。比如起草节点需要知道选题、受众、语气,如果每一步都单独传,你得维护一份很长的参数表,图稍微复杂一点就会失控。共享状态让每个节点只关心自己需要的那几个字段,其他字段自然流过。

2.4 节点注册与边的声明

LangGraph 的用法很固定:

from langgraph.graph import StateGraph, START, END def node_research(state: WritingState) -> dict: # 做素材收集 return {"source_materials": new_materials} def node_outline(state: WritingState) -> dict: # 生成提纲 return {"outline": outline_text} def node_draft(state: WritingState) -> dict: # 写初稿 return {"draft": draft_text} def node_evaluate(state: WritingState) -> dict: # 自评 return {"evaluation_score": score, "evaluation_feedback": feedback} def node_revise(state: WritingState) -> dict: # 修改 return {"revised": revised_text} def node_human_review(state: WritingState) -> dict: # 留出人工确认点 return {"final_text": state["revised"] if state.get("revised") else state["draft"]} graph_builder = StateGraph(WritingState) graph_builder.add_node("research", node_research) graph_builder.add_node("outline", node_outline) graph_builder.add_node("draft", node_draft) graph_builder.add_node("evaluate", node_evaluate) graph_builder.add_node("revise", node_revise) graph_builder.add_node("human_review", node_human_review) graph_builder.add_edge(START, "research") graph_builder.add_edge("research", "outline") graph_builder.add_edge("outline", "draft") graph_builder.add_edge("draft", "evaluate") # 条件边:自评低于 70 分则重新修订 graph_builder.add_conditional_edges( "evaluate", lambda state: "revise" if state["evaluation_score"] < 70 else "human_review", {"revise": "revise", "human_review": "human_review"} ) graph_builder.add_edge("revise", "evaluate") graph_builder.add_edge("human_review", END) graph = graph_builder.compile()

上面这段代码已经是一个可运行的骨架。注意条件边的写法:add_conditional_edges接收一个映射函数和一个路由表,函数返回的字符串对应路由表里的键。这里返回"revise"就跳到修订节点,返回"human_review"就跳去人工审核,最终走向 END。

3. 核心节点实现:把每一步都做成可维护的高质量模块

3.1 素材研究节点:别让模型裸奔写稿

我见过太多人让 AI 直接写“关于XX”,然后模型一本正经地编数据。写作 Agent 必须有一个独立的素材研究节点,它的职责是让后续生成有依据。

我通常的做法是给这个节点挂两个工具。第一个是搜索引擎 API,用于获取最新的网页标题和摘要;第二个是内部知识库检索器,用于匹配历史文章、行业规范文档。两个工具的召回结果会被合并去重,然后作为source_materials写入 State。

def node_research(state: WritingState) -> dict: query = f"{state['topic']} 行业数据 最新趋势" search_results = search_api(query, top_k=5) kb_results = knowledge_base.search(state["topic"], top_k=3) materials = [] for r in search_results: materials.append({"title": r.title, "url": r.url, "snippet": r.snippet}) for r in kb_results: materials.append({"title": r.metadata["title"], "content": r.snippet}) return {"source_materials": materials}

注意这里我把返回的字典里加了一个新字段source_materials,这在 TypedDict 里已经定义过。LangGraph 的 State 合并机制会把返回的字典里的键值更新到全局状态,没有返回的键保持不变。

3.2 提纲生成节点:结构化输出是关键

初稿质量取决于提纲,提纲质量取决于有没有把素材消化。所以我让提纲节点同时输入选题和素材,并且强制要求按 Markdown 结构输出。

from langchain_core.prompts import ChatPromptTemplate OUTLINE_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是资深内容策划,请根据选题和资料输出文章提纲。要求:1. 有明确的引入、正文分层、结论;2. 每个一级标题下列出 2-3 个要点;3. 必须交代关键数据的使用位置。只输出提纲,不要输出正文。"), ("human", "选题:{topic}\n可用素材:{materials}") ]) def node_outline(state: WritingState) -> dict: response = llm.invoke(OUTLINE_PROMPT.format( topic=state["topic"], materials="\n".join([m["title"] + ": " + m["snippet"] for m in state["source_materials"]]) )) return {"outline": response.content}

这里我用了ChatPromptTemplate,只是为了让提示词和逻辑分离。实际上你也可以直接拼字符串,但用模板的好处是后续换模型、调整语气都不影响主代码。提纲生成后存入 State,后面所有节点都可以看到。

3.3 初稿生成节点:控制长度和风格变量

初稿节点是最重的,也是最容易出问题的。我的经验是:不要指望一次调用就拿到完美初稿,宁可第一次写得“干巴”一点,让审校节点去驱动第二次修订。

DRAFT_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是撰稿人。请按照提纲撰写正文,要求:1. 每段不少于 150 字;2. 开头 100 字内点明核心观点;3. 数据引用必须标记[n];4. 语气{style}。"), ("human", "选题:{topic}\n提纲:{outline}\n素材:{materials}") ]) def node_draft(state: WritingState) -> dict: response = llm.invoke(DRAFT_PROMPT.format( topic=state["topic"], outline=state["outline"], materials=format_materials(state["source_materials"]), style=state["tone"] )) return {"draft": response.content}

这里有个容易被忽略的参数:temperature。初稿生成时我习惯设置为 0.6 到 0.7,保留一定创造力;但审校节点和修订节点的 temperature 会降到 0.2 以下,因为审校要求客观、修订要求精准,太高的 temperature 会让模型“自由发挥”,改出来的东西可能把对的地方改错。

3.4 自评审校节点:用第二双眼睛把关

自评节点的本质是一个多标准评委。我会让它从结构完整性、数据准确性、逻辑通顺度、风格一致性四个维度打分,然后返回一个总分和一句修改建议。

EVALUATE_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是严格的内容审校。请对给定文章按 0-100 打分,并给出修改建议。打分维度:结构(30分)、数据准确(30分)、逻辑(25分)、风格(15分)。输出 JSON:{{'score': int, 'feedback': str}}"), ("human", "选题:{topic}\n文章:{draft}") ]) def node_evaluate(state: WritingState) -> dict: response = llm.invoke(EVALUATE_PROMPT.format( topic=state["topic"], draft=state["draft"] )) result = json.loads(response.content) return {"evaluation_score": result["score"], "evaluation_feedback": result["feedback"]}

需要说明的是,大模型输出 JSON 并不一定总是合法。我的做法是在调用后加一段容错逻辑:先尝试json.loads,如果失败就正则提取"score"和"feedback"字段;再不行就让模型重新输出一次。实际跑下来的经验是,只要模型选得不太差,90% 以上的概率是能正常解析的。

3.5 修订节点:带着审校反馈二次创作

修订节点和初稿节点最大的区别是,它必须同时读到初稿和审校意见。这样模型才知道哪里该改、为什么改。我甚至会在 prompt 里强调“只改被指出的问题,不要重写全文”,这样能避免模型“顺手”把其他段落改坏。

REVISE_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是修订编辑。根据审校反馈对文章进行修改。要求:只处理反馈中提到的问题,保留原文其他部分的风格和结构。输出修改后的完整文章。"), ("human", "原文:{draft}\n审校反馈:{feedback}") ]) def node_revise(state: WritingState) -> dict: response = llm.invoke(REVISE_PROMPT.format( draft=state["draft"], feedback=state["evaluation_feedback"] )) return {"revised": response.content}

注意:修订节点返回的是revised字段。如果修订后还要再走一次审校,那审校节点的输入需要同时兼容draft和revised。一种更省心的做法是,在进入审校前把revised统一赋给draft:

def normalize_state(state: WritingState) -> WritingState: if state.get("revised"): state["draft"] = state["revised"] state["revised"] = None return state

不过如果你使用 LangGraph 的状态更新规则,直接返回{"draft": state["revised"]}也能达到同样效果。关键是别让审校节点每次只审同一个draft导致循环空转。

3.6 人工审核节点:给人类留一个“插队”入口

人工审核是 Agent 类产品里非常重要的一环,但在 Demo 里经常被省略。实际上写作是强主观活动,有些内容 AI 改三遍也不一定满足要求,人工确认可以避免“机器觉得很好、人觉得很差”的尴尬。

LangGraph 提供了两种常见的人工介入方式。一种是我们前面用到的普通节点,等模型跑到这个节点时自行停止,用户手动检查后决定继续或中止;另一种是利用interrupt机制,这个官方特性专门用于 Human-in-the-loop 场景。

下面用interrupt改造人工审核节点:

from langgraph.types import interrupt, Command def node_human_review(state: WritingState) -> dict: current_text = state.get("revised") or state["draft"] result = interrupt({"text": current_text, "topic": state["topic"]}) if result.get("approved"): return {"final_text": result.get("text", current_text)} else: return Command( goto="draft", update={"draft": result.get("feedback", ""), "evaluation_score": 40} )

这段代码里interrupt会让图在到达人工审核节点后暂停,把待确认的文章展示给操作者。操作者可以选择批准,或者打回重写。如果打回,我通过返回Command强制跳回draft节点,同时把分数改成 40 分,这样即使不走审校节点,也会触发一次修订流程。这种方式比纯条件边更动态,适合运营团队使用。

4. 实操过程:从零搭出一个可用版本

4.1 环境准备与依赖安装

如果你是在一台干净的机器上复现,建议先用venv隔离环境。我自己使用的是 Python 3.10 以上版本,LangGraph 较新版本对 3.10/3.11 支持最稳定。

python -m venv .venv source .venv/bin/activate pip install langgraph langchain-core langchain-openai

这里我安装了langchain-openai,但并不是说必须使用某一个固定厂商的模型。LangGraph 的模型调用层设计得很干净,只要封装成接口即可。如果你更习惯用本地方案,可以换成对应的 provider 或者直接接入基于 OpenAI 协议兼容的本地推理服务。关键是先跑通流程,再优化模型选择。

4.2 初始化模型与图编译

下面给出一段更完整的初始化代码。这里我把常用的几个节点和条件边整合进一个build_writing_agent函数,方便复用。

import json from typing import TypedDict, Optional, List from langchain_core.prompts import ChatPromptTemplate from langgraph.graph import StateGraph, START, END from langgraph.types import interrupt, Command # ---------- 1. 定义 State ---------- class WritingState(TypedDict): topic: str audience: str tone: str outline: Optional[str] draft: Optional[str] revised: Optional[str] evaluation_score: Optional[int] evaluation_feedback: str source_materials: List[str] # ---------- 2. 定义节点 ---------- # 以下节点函数体同上一节,这里用注释占位 # def node_research(...) # def node_outline(...) # def node_draft(...) # def node_evaluate(...) # def node_revise(...) # def node_human_review(...) # ---------- 3. 构图 ---------- builder = StateGraph(WritingState) builder.add_node("research", node_research) builder.add_node("outline", node_outline) builder.add_node("draft", node_draft) builder.add_node("evaluate", node_evaluate) builder.add_node("revise", node_revise) builder.add_node("human_review", node_human_review) builder.add_edge(START, "research") builder.add_edge("research", "outline") builder.add_edge("outline", "draft") builder.add_edge("draft", "evaluate") builder.add_conditional_edges( "evaluate", lambda s: "revise" if (s.get("evaluation_score") or 0) < 70 else "human_review", {"revise": "revise", "human_review": "human_review"} ) builder.add_edge("revise", "evaluate") builder.add_edge("human_review", END) app = builder.compile()

编译后的app是一个可执行对象,可以用app.invoke(...)或app.astream(...)来运行。

4.3 完整调用流程:生成一篇示例文章

我专门拿一个示例来跑跑看:标题是“如何写出高质量的行业分析文章”。初始状态只需要携带topic、audience和tone。

result = app.invoke({ "topic": "如何写出高质量的行业分析文章", "audience": "中级内容从业者", "tone": "专业但不晦涩", }) print("最终得分:", result.get("evaluation_score")) print("审校反馈:", result.get("evaluation_feedback")) print("最终文章:", result.get("final_text") or result.get("revised") or result.get("draft"))

第一次跑出来的结果很有意思:初稿得分为 63 分,审校节点给出的反馈是“开篇缺少数据支撑,第二部分逻辑跳跃,建议补充一段背景说明”。由于分数低于 70,条件边自动把流程送进了修订节点,修订后的文章得分提升到 82 分,然后进入人工审核节点。我在人工审核时手动做了一次删减,把开头一段偏理论的内容压缩成两句话,最终输出的成品质量明显高于普通单轮生成。

4.4 参数调整:temperature、最大迭代次数与超时控制

如果你试跑后发现效果不稳定,先检查几个参数。

温度(temperature):我建议在不同节点使用不同温度。初稿节点用 0.6-0.7,审校和修订节点用 0.1-0.2。温度过高会让审校节点打出忽高忽低的分数,导致循环次数不稳定。

最大迭代次数:LangGraph 默认图可以在循环边之间无限循环吗?答案是不会真的无限循环,你可以用recursion_limit限制。用法如下:

config = {"recursion_limit": 10} result = app.invoke(initial_state, config=config)

如果超过递归上限,LangGraph 会抛出异常。所以我在生产环境里通常把上限设为 8,防止模型反复进入”修订->审校->修订“的死循环。

超时控制也是生产环境里必须做的。这里的超时不是指模型单次调用的超时,而是整体图执行的超时。可以给每个节点单独设置超时,也可以用asyncio.wait_for包裹整个调用。实际部署时我更倾向于在调用图之前加一层超时装饰器,超过 120 秒就放弃本次生成,返回中间产物,让用户决定是否重新跑。

4.5 用astream做流式输出,体验会好很多

文字的生成过程是流式的,如果你想在前端做一个像打字机一样的效果,直接用invoke是不行的。LangGraph 的astream可以逐节点输出状态更新,再结合模型本身的流式能力,效果很好。

async def stream_run(state): async for event in app.astream(state, stream_mode="values"): latest = event # 可以根据 latest 里的哪个字段更新来渲染 if latest.get("draft") is not None: print("\n[说人话] 当前草稿已生成...")

实际体验是:素材研究阶段几乎无感,提纲生成后可以展示给用户确认,初稿阶段就是一段一段蹦出来的。这种交互方式比最后一次性返回全部文本要自然得多。

5. 常见问题与排查技巧实录

5.1 State 字段越界或拼写错误

LangGraph 对 TypedDict 的兼容并不是强制的,运行时如果某个节点返回的键不存在,会触发校验错误。我的排查方式是先做一次最简单的 invoke,只给topic,然后把断点打在节点入口,打印整个 State 看一眼。

import pdb def node_draft(state: WritingState) -> dict: pdb.set_trace() ...

更推荐的做法是利用 LangGraph 的debug模式或插入日志中间件。每次节点返回后,自动记录 diff 形式的 State 变更。这一招在排查“某个字段被覆盖了”的时候特别有用。

5.2 循环卡死:修订与审校反复横跳

这个问题的典型表现是:审校分数在 65 到 75 之间反复横跳,导致图进不了人工审核节点。

我的判断标准是:如果连续三次修订后分数都在 70 分附近波动,继续循环已经没有意义了。解决办法是在审校节点引入一个“已修订次数”字段:

state.setdefault("revision_count", 0) if state["revision_count"] >= 3: return {"evaluation_score": max(70, state["evaluation_score"]), "force_exit": True}

最终在条件边里判断force_exit:

lambda s: "human_review" if (s.get("force_exit") or (s.get("evaluation_score") or 0) >= 70) else "revise"

这种“软上限”比直接报错优雅得多,用户拿到 65 分的文章后还能通过人工审核节点再做兜底。

5.3 模型输出 JSON 解析失败

不少模型在返回 JSON 时会在前后加上```json这类标记。直接用json.loads会报错。我写了一个小的解析函数:

import re import json def parse_json_response(text: str) -> dict: # 优先尝试完整解析 try: return json.loads(text) except Exception: pass # 提取 JSON 块 match = re.search(r'\{.*\}', text, re.DOTALL) if match: try: return json.loads(match.group(0)) except Exception: pass # 最后尝试提取键值对 score = re.search(r'"score"\s*:\s*(\d+)', text) feedback = re.search(r'"feedback"\s*:\s*"(.*?)"', text) if score and feedback: return {"score": int(score.group(1)), "feedback": feedback.group(1)} raise ValueError(f"Cannot parse model response: {text}")

这个函数看起来不那么“高级”,但非常实用。我见过很多项目因为在这里偷懒,导致整条工作流在两三天内频繁崩溃。

5.4 上下文长度超限:素材或草稿超出模型窗口

写作 Agent 非常容易把上下文塞满,因为source_materials可能有几十条记录。我的经验是:不要在进入 draft 节点前把全部素材一股脑丢进去,而是让 outline 节点先做筛选,只保留与提纲强相关的素材。

def node_outline(state) -> dict: # 第一步:调用模型生成提纲 # 第二步:根据提纲关键词筛选素材 # 第三步:把筛选后的素材放入 source_materials 子集 return {"outline": outline_text, "selected_materials": relevant_materials}

这样草稿节点的输入量会小很多,同时还能提升内容聚焦度。另一个技巧是把素材里的长文本提前做摘要,用摘要而不是原文。

5.5 人工审核节点没有真正暂停

很多人第一次用interrupt时会发现图并没有停下来,还是一口气跑完了。原因是你用的是invoke而不是astream,而且没有处理interrupt抛出的特殊异常。

官方推荐的执行方式是使用流式接口:

events = app.astream(initial_state, config) for event in events: # 检查是否有 interrupt ...

如果你坚持用invoke,需要捕获GraphInterrupt异常,然后在外部基于Command(resume=...)恢复执行。这块逻辑在复杂场景下容易绕,我的建议是新手尽量用stream_mode="values"配合前端交互,语义更清晰。

6. 从 Demo 到生产环境:几个可以继续扩展的方向

如果你已经跑通了上面这套流程,恭喜,你已经拥有了一个非常扎实的写作 Agent 底子。再往后,有几个方向我认为价值很高。

第一,把人工审核节点做得更灵活。现在它只能操作“通过/打回”,你可以扩展成“支持标注修改”,让 Agent 在下一轮迭代时参考这些标注。这相当于给 Agent 注入了一个持续学习的人类反馈闭环。

第二,增加多 Agent 协作。写行业分析文章往往需要两三个视角:数据专家、行业分析师、文字编辑。你可以把目前的一个写作 Agent 拆成三个子图,分别负责“找数据、定视角、写文本”,再用一个总图把它们编排起来。LangGraph 本身也支持子图嵌套,复杂度可控。

第三,设计一套缓存策略。决定文章质量的核心因素之一是素材研究节点。同一个选题在短期内可能被反复使用,如果能把source_materials按主题缓存起来,能省掉大量搜索 API 调用,也降低延迟。我实际部署时会在每个节点内部加一层缓存装饰器,命中后直接返回历史结果。

我自己的体会是,LangGraph 真正的优势不在于它比那些“无代码流程编排”工具更简单,而在于它给你提供了完整的状态管理和图执行语义。写作 Agent 只是我拿它练手的第一个场景,等你实际用过一段时间后会发现,这个框架几乎可以复用到所有需要“多步决策、条件循环、人工介入”的业务里。重点不是写代码,而是把脑海中的流程画成一张图,然后把每个节点变成可测试的函数,整条链路立刻就有了一种稳健可控的感觉。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询