ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

【智能体开发】LangChain实战:用StateGraph构建Self-Refine自我纠错链,让LLM复杂生成不再翻车

【智能体开发】LangChain实战:用StateGraph构建Self-Refine自我纠错链,让LLM复杂生成不再翻车 1. 为什么单次生成总翻车Self-Refine 要解决的真实场景如果你用 LangChain 做过稍微复杂一点的生成任务大概率遇到过这种场面让模型写一个带分页、缓存、异常兜底的 Flask 接口第一次输出看着挺像回事粘到项目里一跑要么KeyError要么分页参数没校验要么缓存 key 拼错。你回头改 Prompt加了一堆“请务必考虑边界情况”结果模型换了个姿势继续漏。这不是模型不行而是单次生成这个模式本身就不适合复杂任务。复杂生成有几个天然难点依赖关系多一个函数要同时满足参数校验、错误码、日志、性能、上下文长超过 200 行代码后模型容易“忘记”前面的变量命名、边界条件碎空值、超时、并发。指望一次推理全覆盖概率很低。Self-Refine 的思路很朴素把“一次写完”拆成“生成 → 自评 → 修正”的循环。模型先出一版然后切换成审查者角色挑毛病再基于具体反馈改。每一轮只聚焦一类问题反而比一次性要求“面面俱到”更靠谱。在 LangChain 生态里落地这个循环最顺手的工具是LangGraph 的 StateGraph。它提供状态管理、条件边、循环控制正好对应 Self-Refine 需要的“记住当前是第几轮、评分多少、该不该继续”。如果你只用LLMChain串行拼很快就会遇到“不知道怎么停”“中间状态丢了”“出错没法回溯”这三个坑。这篇文章面向的是已经会用 LangChain 基础组件、想把手上的生成链路做稳的开发者。我会用 StateGraph 搭一个完整的 Self-Refine 图给出可复制的节点代码、条件边配置、Critic 提示词模板并且用 TaoToken 的统一 Key 把全链路跑通最后对比开启前后同一任务的输出质量。全程可以跟着敲不需要你有 LangGraph 使用经验。核心检索词先明确LangChain Self-Refine 自我纠错配合StateGraph 条件边和LLM 复杂生成质量优化这三个词贯穿全文。2. TaoToken 前置准备统一 Key 跑通 LangChain 全链路在写 StateGraph 之前得先把模型调用这一层搞定。Self-Refine 会在一轮任务里调用模型多次生成一次、批评一次、修正一次循环起来可能 6 到 10 次如果每次都要切换不同厂商的 Key、改 base_url、对不同的 SDK 参数调试成本会非常高。我的做法是用 TaoToken 做统一入口一个 Key 覆盖生成和批评两个角色。TaoToken 的定位是给开发者提供统一的模型调用入口官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。它兼容 OpenAI 风格的接口所以 LangChain 里直接用ChatOpenAI就能接不需要额外写适配层。具体操作步骤第一步打开控制台创建 Key。地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 登录后在 API Keys 页面新建一个复制出来先存到环境变量里别硬编码进代码。第二步确认你要用的模型 ID。Self-Refine 里我建议生成和批评用同一个模型先跑通稳定后再考虑用便宜模型做 Critic 降本。模型列表可以在模型对话页面查看 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 选一个你额度够用的。第三步配置环境变量。Linux/macOS 下export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/apiWindows PowerShell$env:TAOTOKEN_API_KEYsk-你的key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api第四步装依赖。LangGraph 现在和 LangChain 是分开的包别只装langchainpip install langchain langchain-openai langgraph这里有个容易踩的点langchain-openai的ChatOpenAI默认会去读OPENAI_API_KEY我们要显式传参覆盖否则会报 401。后面配置章节会给完整写法。如果你打算长期跑编码类 Agent 任务迭代次数多、Token 消耗大可以看下 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 按套餐走比按量计费更可控。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数问题先查这里。Key 拿到后先做一次最小验证别等 StateGraph 写完才发现 Key 不通import os from langchain_openai import ChatOpenAI llm ChatOpenAI( model你的模型ID, api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], temperature0.2, ) print(llm.invoke(回复两个字通了).content)能打印出内容说明 Base URL Key Model ID 三件套没问题可以进入下一步。3. 可复制配置StateGraph 节点与条件边完整代码这一节是全文核心给出可以直接跑的 Self-Refine 图。先讲状态设计再讲三个节点最后讲条件边。状态用TypedDict定义字段要覆盖整个循环需要的信息from typing import TypedDict, List class RefineState(TypedDict): task: str # 原始任务描述 draft: str # 当前版本的输出 feedback: str # Critic 给出的结构化反馈 score: float # 本轮评分 history: List[float] # 历史评分用于停滞检测 iteration: int # 当前轮次生成节点第一轮用原始任务后续轮次把反馈拼进去。def generate_node(state: RefineState) - dict: if state[iteration] 0: prompt f请完成以下任务直接输出结果\n{state[task]} else: prompt ( f任务{state[task]}\n\n f上一版输出\n{state[draft]}\n\n f审查反馈\n{state[feedback]}\n\n f请根据反馈修正只输出修正后的完整结果。 ) resp llm.invoke(prompt) return { draft: resp.content, iteration: state[iteration] 1, }批评节点要求模型输出 JSON方便解析评分。import json, re CRITIC_TMPL 你是严格的代码审查员请从以下维度审查 1. 功能完整性是否覆盖任务所有要求 2. 异常处理空值、边界、错误分支 3. 代码规范命名、结构、可读性 4. 性能是否有明显瓶颈 输出必须是 JSON格式 {{score: 0-10 的数字, issues: [问题1, 问题2], suggestion: 总体修改建议}} 待审查内容 {draft} def critique_node(state: RefineState) - dict: prompt CRITIC_TMPL.format(draftstate[draft]) resp llm.invoke(prompt) text resp.content match re.search(r\{.*\}, text, re.S) data json.loads(match.group()) if match else {score: 0, issues: [], suggestion: text} feedback f问题{data[issues]}\n建议{data[suggestion]} return { feedback: feedback, score: float(data[score]), history: state[history] [float(data[score])], }条件边决定继续还是结束。from langgraph.graph import END def should_continue(state: RefineState) - str: if state[iteration] 5: return END if state[score] 8.5: return END if len(state[history]) 2: if state[history][-1] - state[history][-2] 0.5: return END return generate组装图from langgraph.graph import StateGraph workflow StateGraph(RefineState) workflow.add_node(generate, generate_node) workflow.add_node(critique, critique_node) workflow.set_entry_point(generate) workflow.add_edge(generate, critique) workflow.add_conditional_edges(critique, should_continue, {generate: generate, END: END}) app workflow.compile()这里有个细节add_conditional_edges的映射字典里END作为 key 时要用END常量本身不要写成字符串END否则会报找不到节点。我第一次写就栽在这报错信息是Node END not found排查了十几分钟。另外如果你用 Cline 或 CC Switch 这类工具调试配置里同样要写全三件套Base URL 填https://taotoken.net/apiKey 填你的Model ID 填模型列表里的准确名称。Codex 的auth.json也是同理字段名按官方文档来别自己造。4. 验证请求跑通全链路并对比输出质量配置写完跑一个真实任务验证。任务选一个容易暴露问题的生成一个带缓存和分页的用户查询接口。task 用 Flask 写一个 GET /users 接口要求 1. 支持 page 和 page_size 分页参数默认 page1, page_size10 2. 用 Redis 缓存查询结果缓存 60 秒 3. 参数非法时返回 400用户不存在返回 404 4. 返回 JSON 格式 result app.invoke({ task: task, draft: , feedback: , score: 0.0, history: [], iteration: 0, }) print(最终轮次, result[iteration]) print(最终评分, result[score]) print(评分轨迹, result[history]) print(最终输出\n, result[draft])实测下来第一轮评分通常在 5 到 6 分Critic 会指出“未校验 page_size 上限”“缓存未处理序列化异常”“404 分支缺失”。第二轮修正后评分到 7 到 8 分第三轮补上缓存一致性后到 8.5 以上条件边触发结束。整个过程 3 轮Token 消耗大概是单次生成的 4 倍左右但输出质量差距明显。对比一下开启前后的差异用同一任务单次生成single llm.invoke(task).content print(single)单次生成的典型问题分页参数直接int(request.args.get(page))没传就崩缓存 key 用fusers_{page}漏了page_size导致不同页大小命中同一缓存404 分支经常忘写。Self-Refine 版本这些问题基本都被 Critic 抓出来了。如果你想在模型对话页面手动对比不同模型的批评质量可以打开 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentchatutm_campaignrewrite 把同一段代码分别丢给不同模型审查看谁的反馈更具体。这一步对调优 Critic 提示词很有帮助。验证成功的标志有三个iteration在 2 到 5 之间自然停止、history评分呈上升趋势、最终输出里 Critic 提过的问题都消失了。如果iteration直接跑到 5 还没停说明评分阈值或停滞检测需要调如果第一轮就 8.5 分结束说明 Critic 太宽松要加严提示词。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth跑 Self-Refine 链路时报错集中在几个地方逐个说。401 Unauthorized。最常见的原因是ChatOpenAI没读到你的 Key去读了默认的OPENAI_API_KEY。解决方法是显式传api_key和base_url别依赖环境变量自动读取。还有一种情况是 Key 复制时带了空格或换行用print(repr(os.environ[TAOTOKEN_API_KEY]))检查一下首尾字符。local proxy failed。这个报错通常出现在你本机有网络代理配置、但代理没启动或端口不对的时候。检查HTTP_PROXY/HTTPS_PROXY环境变量如果不需要代理就清掉。注意这里说的是本机环境变量层面的问题不是让你去配什么特殊网络工具纯粹是排查环境变量污染。reading choices 相关报错。典型信息是KeyError: choices或list index out of range出现在解析响应时。原因一般是模型返回了非标准结构或者你的base_url末尾多了斜杠导致路径拼接错误。确认base_url写成https://taotoken.net/api不要写成https://taotoken.net/api/。另外检查模型 ID 是否拼错ID 不对时有些网关会返回错误结构而不是标准响应。OAuth 相关报错。如果你在 Claude Code 或类似工具里配置报 OAuth 失败通常是把 API Key 模式配成了 OAuth 模式。Claude Code 接入时用 Anthropic 兼容配置参考 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里的 ClaudeCodeAnthropic 章节Base URL 和 Key 按文档填别混用两种认证方式。JSON 解析失败。Critic 节点里json.loads报错是因为模型输出里带了 Markdown 代码块标记。我的处理是用正则re.search(r\{.*\}, text, re.S)先抠出 JSON 部分再解析。更稳的做法是在提示词里明确“不要用代码块包裹直接输出 JSON”。循环不停止。iteration一直涨到上限说明条件边逻辑有问题。检查should_continue的返回值是否和add_conditional_edges的映射 key 完全一致字符串大小写、空格都会导致路由失败失败时默认走第一个分支看起来就像“停不下来”。状态字段丢失。报KeyError: history是因为某个节点返回的 dict 没包含该字段。LangGraph 的状态更新是合并式的节点只返回要改的字段即可但首次进入图时初始 state 必须包含所有字段app.invoke时别漏。6. 语义一致 CTA把 Self-Refine 用到你的项目里Self-Refine 这套东西跑通 demo 只是第一步真正有价值的是接到你现有的生成链路里。我的建议是先从一个高频、易错的生成任务切入比如接口代码生成、SQL 生成、长文摘要别一上来就全量替换。接入时优先把 Critic 提示词打磨好它决定了整个循环的天花板。评分维度要具体到能指出行号反馈格式要结构化到能程序化解析。终止条件用复合判断别只靠固定轮次。如果你要长期跑这类多轮迭代任务Token 消耗会比单次生成高不少可以看下 Coding Plan 的套餐 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。需要新建 Key 或管理额度去控制台 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 。API Keys 管理页在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 接入参数有疑问查文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。最后留一个我踩过的坑别把temperature设太高。Self-Refine 的修正节点需要稳定复现问题temperature超过 0.5 后同一份反馈可能改出完全不同的结果评分轨迹会乱跳停滞检测也失效。生成和批评都控制在 0.2 到 0.3 之间输出质量最可控。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进