AI工作流搭建实战：从LangChain到LangGraph的完整指南-育师

AI 工具这两年最大的变化，不是模型本身又强了多少，而是大家开始认真琢磨"怎么把模型塞进一条能稳定跑起来的流水线里"。我身边不少朋友一开始都是打开对话框，问一句答一句，用得很开心；等到想把这件事变成每天自动跑、批量处理、还能接进自己业务系统的时候，就卡住了。卡住的地方往往不是模型能力，而是工作流——也就是把输入、模型调用、判断、外部工具、输出这几步串起来的那套骨架。这篇就围绕"用 AI 工具搭建工作流"这件事，把从环境准备到跑通第一条链路、再到接进真实业务的完整过程讲清楚，代码和操作指引都会给到，适合刚接触工作流编排的开发者，也适合已经会用大模型 API、但还没系统搭过流程的人。

我会以 Python 为主语言，核心编排用 LangChain 和 LangGraph，批处理调度用 Airflow 做对照，中间穿插 Dify、Coze 这类可视化平台和纯代码方案的取舍。整篇不堆概念，重点讲每一步为什么这么设计、参数怎么定、坑在哪。你跟着走一遍，应该能自己搭出一条"输入文档 → 模型处理 → 条件判断 → 调用外部工具 → 输出结果"的完整链路。

1. 先把"工作流"这个词拆开看

1.1 工作流到底在解决什么问题

很多人对工作流的理解停留在"把几个步骤连起来"，这个理解不算错，但太浅。真正让工作流有价值的是三件事：可重复、可观测、可干预。

可重复指的是同一份输入，今天跑和明天跑，结果结构一致、流程路径一致。你手动在对话框里问，今天模型心情好多给你两段，明天少给你一段，下游根本没法接。工作流的第一价值就是把这种不确定性收敛到可控范围。

可观测指的是每一步的输入输出都留痕。模型调用失败、外部接口超时、判断分支走错，这些在纯对话里你根本看不见，但在工作流里每一步都是一个节点，节点有日志、有状态、有耗时。出了问题能定位到具体哪一步，这是能不能上生产的分水岭。

可干预指的是流程跑到一半，人可以插进去看一眼、改一下再继续。这个在 LangGraph 里叫 human-in-the-loop，在可视化平台里叫人工审核节点。很多业务场景（比如合同审核、简历初筛）根本不敢让 AI 全自动跑完，必须留个人工确认的口子，工作流框架能不能优雅地支持这个，直接决定它能不能落地。

理解了这三点，你就明白为什么"工作流"不是简单的步骤拼接，而是一套围绕稳定性、可维护性、可控性设计的工程结构。

1.2 纯代码编排和可视化平台，怎么选

这是新手最容易纠结的问题。我的建议是先看你的团队构成和交付节奏。

可视化平台（Dify、Coze 这类）的优势是上手快，拖拽连线，非技术同学也能改流程，适合快速验证想法、做内部工具、或者流程本身不复杂且变动频繁的场景。缺点是复杂逻辑表达起来别扭，比如嵌套条件、循环、自定义状态管理，拖拽界面会越拖越乱，而且深度定制受平台能力限制。

纯代码编排（LangChain、LangGraph）的优势是逻辑表达自由，复杂分支、循环、状态机都能写，版本管理、测试、CI/CD 都能接进现有工程体系。缺点是门槛高一些，需要你会 Python，调试也更依赖日志。

我的实际做法是混合：用可视化平台做原型和给业务方演示，确认流程价值后，核心链路用代码重写，平台只保留给非技术同学做参数调整的入口。这样既快又不失控。

维度	可视化平台	纯代码编排
上手速度	快，拖拽即可	慢，需要编程基础
复杂逻辑	受限，嵌套多了很乱	自由，状态机随便写
版本管理	平台内管理，弱	Git 管理，强
调试能力	看节点日志	完整日志 + 断点
适合场景	原型、内部工具、简单流程	生产链路、复杂业务
团队协作	非技术可参与	需开发主导

1.3 一条典型 AI 工作流长什么样

在动手之前，先在脑子里画出一条标准链路。绝大多数 AI 工作流都逃不出这个骨架：

输入接收：从文件、数据库、接口拿到原始数据
预处理：清洗、分块、格式转换
模型调用：把处理好的内容送给大模型
结果解析：把模型返回的自然语言解析成结构化数据
条件判断：根据结果决定走哪条分支
外部工具调用：查数据库、调接口、写文件
人工审核（可选）：关键节点插入人工确认
输出落库：结果写入目标位置

这条链路看着简单，但每一步都有讲究。比如预处理阶段的分块策略直接影响模型效果，结果解析阶段如果模型返回格式不稳定，整个下游都会崩。后面我会逐步展开。

2. 环境准备：别在这一步浪费时间

2.1 Python 环境与依赖管理

工作流项目对环境的依赖比普通脚本重，因为要同时装 LangChain、LangGraph、各种模型 SDK、还有可能用到 Airflow。我强烈建议用虚拟环境隔离，别往全局 Python 里装。

# 创建虚拟环境 python -m venv ai_workflow_env # 激活（Windows） ai_workflow_env\Scripts\activate # 激活（macOS / Linux） source ai_workflow_env/bin/activate # 升级 pip python -m pip install --upgrade pip

Python 版本建议 3.10 或 3.11。3.12 有些库的兼容性还在追，3.9 又偏老，3.10/3.11 是目前最稳的区间。装之前用python --version确认一下。

依赖安装分两批，核心的和可选的：

# 核心编排 pip install langchain langchain-core langgraph # 模型接入（以 OpenAI 兼容接口为例） pip install langchain-openai # 文档处理 pip install pypdf python-docx # 环境变量管理 pip install python-dotenv # 调度（可选，后面讲 Airflow 时再装） pip install apache-airflow

提示：LangChain 生态拆包很细，langchain、langchain-core、langchain-community、各家模型包是分开的。装的时候看清楚文档对应版本，不同版本 API 差异不小，尤其是 0.1 到 0.2 之间有不少破坏性变更。

2.2 密钥和配置怎么管

密钥绝对不能写死在代码里。用.env文件加python-dotenv是最省事的做法：

# .env 文件 OPENAI_API_KEY=你的密钥 OPENAI_BASE_URL=你的接口地址 MODEL_NAME=gpt-4o-mini

# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL") MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")

记得把.env加进.gitignore，这是基本纪律。我见过不止一次有人把密钥提交到仓库，然后被扫到盗刷，损失不小。

2.3 编辑器与调试环境

VS Code 是搭工作流最顺手的编辑器。装好 Python 扩展后，把解释器指向刚才创建的虚拟环境（Ctrl+Shift+P→Python: Select Interpreter）。这样代码补全、调试、运行都在同一个环境里，不会出现"命令行能跑、编辑器报错"的尴尬。

调试工作流有个技巧：每个节点单独可测。别一上来就跑整条链路，先把每个节点写成独立函数，单独喂输入验证输出，确认没问题再串起来。这样出问题时你能快速定位是哪个节点的问题，而不是面对一整条链路抓瞎。

3. 用 LangChain 搭第一条链路

3.1 为什么从 LangChain 入手

LangChain 的价值在于它把"模型调用"这件事标准化了。你不用关心底层是哪个厂商的接口，统一用ChatPromptTemplate组织提示词，用ChatModel调用模型，用OutputParser解析结果。这套抽象让你换模型时改动最小。

但要注意，LangChain 早期版本把太多东西塞进一个大包，导致又重又乱。现在拆包之后清爽多了，核心就是langchain-core提供抽象，各家模型包提供实现。搭工作流时，LangChain 负责"单步能力"，LangGraph 负责"多步编排"，这个分工要清楚。

3.2 一个最小可用的模型调用

先跑通最基础的调用，确认环境没问题：

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from config import OPENAI_API_KEY, OPENAI_BASE_URL, MODEL_NAME llm = ChatOpenAI( model=MODEL_NAME, api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL, temperature=0.2, ) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的文本处理助手，只输出要求的结构，不要额外解释。"), ("human", "请把下面这段文本总结成一句话：\n\n{text}"), ]) chain = prompt | llm result = chain.invoke({"text": "这里放你要处理的文本内容"}) print(result.content)

这里temperature=0.2是刻意调低的。工作流场景下，我们要的是稳定和可重复，不是创意。温度越高，同样输入输出越飘，下游解析越容易崩。除非你的场景明确需要多样性（比如生成多个候选文案），否则工作流里温度建议控制在 0.3 以下。

3.3 结构化输出：让模型返回能解析的数据

工作流里最怕的就是模型返回一段自由文本，你还得写正则去抠。正确做法是让模型直接返回 JSON，并用解析器兜底。

from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field class SummaryResult(BaseModel): summary: str = Field(description="一句话总结") keywords: list[str] = Field(description="3到5个关键词") sentiment: str = Field(description="情感倾向：正面/中性/负面") parser = JsonOutputParser(pydantic_object=SummaryResult) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个文本分析助手。{format_instructions}"), ("human", "分析下面这段文本：\n\n{text}"), ]) prompt = prompt.partial(format_instructions=parser.get_format_instructions()) chain = prompt | llm | parser result = chain.invoke({"text": "你的文本内容"}) print(result["summary"], result["keywords"])

用 Pydantic 定义输出结构有两个好处：一是给模型明确的格式约束，二是解析失败时能拿到清晰的错误信息。实测下来，加了格式说明之后，模型返回合法 JSON 的概率能到 95% 以上。剩下那 5% 怎么办？后面讲重试机制时会说。

注意：不同模型对 JSON 模式的支持程度不一样。有些模型有专门的response_format参数强制 JSON 输出，能进一步降低解析失败率。如果你的模型支持，务必用上。

3.4 提示词模板的工程化写法

提示词别散落在代码各处，集中管理。我习惯建一个prompts.py，把所有模板放一起：

# prompts.py from langchain_core.prompts import ChatPromptTemplate SUMMARY_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是文本分析专家，严格按格式输出。"), ("human", "分析文本：{text}"), ]) CLASSIFY_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是分类助手，只返回类别名称。"), ("human", "把下面内容归类到[技术/产品/运营/其他]之一：{text}"), ])

这样做的好处是改提示词不用翻遍代码，而且方便做 A/B 测试——同一份输入用两个版本的提示词跑，对比效果。提示词是工作流里最需要反复迭代的部分，把它工程化管理，长期收益很大。

4. 用 LangGraph 把单步串成流程

4.1 LangChain 和 LangGraph 到底什么关系

这是被问得最多的问题。一句话说清楚：LangChain 管"一步怎么做"，LangGraph 管"多步怎么连"。

LangChain 的 Chain 是线性的，A 完了 B，B 完了 C，遇到需要循环、需要根据中间结果动态决定下一步、需要保存状态跨多轮交互的场景，Chain 就力不从心了。LangGraph 把流程建模成图，节点是处理单元，边是流转关系，还支持条件边和状态持久化，这些能力正好补上 Chain 的短板。

所以不是二选一，而是配合用。LangGraph 的节点内部，照样可以用 LangChain 的 prompt、model、parser。

4.2 状态、节点、边三个核心概念

LangGraph 的心智模型很简单，就三个东西：

状态（State）：一个贯穿全流程的数据结构，通常用 TypedDict 或 Pydantic 定义。每个节点读它、改它，改完传给下一个节点。

节点（Node）：一个函数，接收状态，返回状态的更新部分。

边（Edge）：定义节点之间的流转。普通边是固定的，条件边是根据状态动态决定走哪。

from typing import TypedDict from langgraph.graph import StateGraph, END class WorkflowState(TypedDict): raw_text: str summary: str category: str need_review: bool def summarize_node(state: WorkflowState): # 调用模型做总结 summary = "这里是总结结果" return {"summary": summary} def classify_node(state: WorkflowState): category = "技术" return {"category": category} def review_check_node(state: WorkflowState): # 根据分类决定是否需要人工审核 need_review = state["category"] == "其他" return {"need_review": need_review}

状态设计有个原则：只放流程需要的数据，别把整个上下文都塞进去。状态越大，序列化和传递成本越高，调试时也越难看清。我见过有人把原始文档全文、所有中间结果、模型完整响应都塞进状态，结果状态膨胀到几 MB，跑起来又慢又乱。

4.3 条件分支：让流程会"拐弯"

条件边是 LangGraph 最实用的能力。比如根据分类结果决定走哪条处理路径：

def route_by_category(state: WorkflowState): if state["category"] == "技术": return "tech_path" elif state["category"] == "产品": return "product_path" else: return "manual_review" graph = StateGraph(WorkflowState) graph.add_node("summarize", summarize_node) graph.add_node("classify", classify_node) graph.add_node("tech_path", tech_handler) graph.add_node("product_path", product_handler) graph.add_node("manual_review", review_handler) graph.set_entry_point("summarize") graph.add_edge("summarize", "classify") graph.add_conditional_edges( "classify", route_by_category, { "tech_path": "tech_path", "product_path": "product_path", "manual_review": "manual_review", } ) graph.add_edge("tech_path", END) graph.add_edge("product_path", END) graph.add_edge("manual_review", END) app = graph.compile()

add_conditional_edges的第二个参数是路由函数，它读状态返回一个字符串，第三个参数是字符串到节点的映射。这个设计很灵活，路由逻辑可以任意复杂，只要最终返回一个映射里存在的键就行。

4.4 循环与重试：处理不稳定的模型输出

模型偶尔返回格式不对，这是常态。与其在解析处写一堆 try-except，不如在流程层面做重试。LangGraph 支持把边连回上游节点形成循环：

def validate_node(state: WorkflowState): # 校验 summary 是否为空 is_valid = bool(state.get("summary")) return {"is_valid": is_valid} def route_after_validate(state: WorkflowState): if state["is_valid"]: return "continue" if state.get("retry_count", 0) >= 3: return "give_up" return "retry" graph.add_node("validate", validate_node) graph.add_conditional_edges( "validate", route_after_validate, { "continue": "next_step", "retry": "summarize", # 回到总结节点重试 "give_up": "error_handler", } )

这里有个关键点：重试必须设上限。不设上限的循环一旦遇到模型持续返回异常，会无限跑下去，烧钱又烧时间。我一般设 3 次，超过就转人工或走降级逻辑。

提示：重试时最好在状态里记录重试次数，并且每次重试可以微调参数（比如提高温度、换更明确的提示词），而不是原样重跑。原样重跑大概率还是同样的错误。

5. 接入外部工具与真实数据

5.1 工具调用的两种模式

工作流里调用外部工具，有两种典型模式。一种是流程内固定调用，比如流程走到某一步，必然要查一次数据库，这是流程设计时就定死的。另一种是模型自主决定调用，也就是常说的 function calling 或 tool use，模型根据当前任务自己判断要不要调工具、调哪个。

固定调用简单可控，适合流程明确的场景。模型自主调用灵活，适合任务开放、需要模型自己规划的场景。实际项目里两者经常混用：主干流程用固定调用保证稳定，局部环节给模型几个工具让它自己选。

5.2 定义一个可被模型调用的工具

LangChain 用装饰器就能把普通函数变成工具：

from langchain_core.tools import tool @tool def query_order_status(order_id: str) -> str: """根据订单号查询订单状态。输入订单号，返回状态描述。""" # 实际项目里这里查数据库或调接口 mock_data = { "A001": "已发货", "A002": "待付款", } return mock_data.get(order_id, "订单不存在") @tool def calculate_refund(amount: float, days: int) -> float: """计算退款金额。amount 是原价，days 是已使用天数。""" if days <= 7: return amount return amount * 0.8

工具函数的 docstring 非常重要，模型就是靠它判断这个工具是干什么的、什么时候该用。docstring 写得含糊，模型就会乱调或该调不调。我一般要求 docstring 里写清楚：这个工具做什么、参数是什么含义、返回什么。

5.3 把工具绑到模型上

from langchain_openai import ChatOpenAI llm = ChatOpenAI(model=MODEL_NAME, temperature=0) llm_with_tools = llm.bind_tools([query_order_status, calculate_refund]) response = llm_with_tools.invoke("订单 A001 现在什么状态？") print(response.tool_calls)

bind_tools之后，模型返回的就不只是文本，还可能包含tool_calls，里面写明要调哪个工具、传什么参数。你的工作流负责执行这些调用，把结果再喂回模型，让它生成最终回答。这个"调用 → 执行 → 回喂"的循环，就是 agent 的基本形态。

5.4 工具调用的错误处理

外部工具一定会失败：接口超时、数据不存在、参数格式错。工作流必须能扛住这些。

def safe_tool_call(tool_func, args, max_retries=2): for attempt in range(max_retries + 1): try: return tool_func.invoke(args) except Exception as e: if attempt == max_retries: return f"工具调用失败：{str(e)}" time.sleep(1)

关键原则：工具失败不能让整个流程崩掉。要么返回一个明确的错误信息让模型知道，要么走降级路径。我见过工作流因为一个查询接口挂了，整条链路全断，前面的处理全白费。把每个工具调用都包一层容错，是上生产前的必修课。

6. 调度与批处理：从单次运行到定时任务

6.1 什么时候需要 Airflow

LangGraph 解决的是"一次流程怎么跑"，Airflow 解决的是"什么时候跑、跑多少次、失败了怎么办"。当你的工作流需要每天定时跑、需要处理一批数据、需要监控每次运行的状态，就该上调度器了。

Airflow 的核心概念是 DAG（有向无环图），一个 DAG 就是一条工作流，里面每个 task 是一个执行单元。它自带调度、重试、告警、历史记录，这些是手写 cron 脚本给不了的。

6.2 把 LangGraph 流程包成 Airflow Task

from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta def run_ai_workflow(**context): from workflow import app # 你的 LangGraph 编译结果 input_data = context["dag_run"].conf.get("input_text", "默认输入") result = app.invoke({"raw_text": input_data}) return result default_args = { "owner": "data_team", "retries": 2, "retry_delay": timedelta(minutes=5), } with DAG( dag_id="ai_workflow_daily", default_args=default_args, schedule="0 2 * * *", # 每天凌晨2点 start_date=datetime(2024, 1, 1), catchup=False, ) as dag: task = PythonOperator( task_id="run_workflow", python_callable=run_ai_workflow, )

catchup=False很重要。Airflow 默认会补跑历史所有未执行的周期，如果你的 start_date 设得很早，一上线它会瞬间触发几百次运行。新手经常踩这个坑，设成 False 就只跑当前及以后的周期。

6.3 批处理的并发控制

批量处理时，别一股脑全并发出去。模型接口通常有速率限制，并发太高会被限流甚至封禁。用 Airflow 的并发参数控制：

with DAG( dag_id="ai_batch_process", max_active_runs=1, # 同一时间只跑一个 DAG 实例 concurrency=5, # 最多 5 个 task 同时跑 ... ) as dag: ...

或者在代码层面用信号量控制：

import asyncio semaphore = asyncio.Semaphore(5) async def process_one(item): async with semaphore: return await call_model(item)

并发数设多少合适？我的经验是从小往大试。先设 3 到 5，观察接口响应时间和错误率，稳定了再往上加。别一上来就设 50，大概率直接触发限流。

6.4 失败重试与告警

批处理最怕的是"跑了一半挂了，不知道挂在哪"。Airflow 的重试机制能自动处理偶发失败，但重试次数要合理：

default_args = { "retries": 3, "retry_delay": timedelta(minutes=2), "retry_exponential_backoff": True, # 指数退避 "on_failure_callback": send_alert, # 失败告警 }

retry_exponential_backoff=True让重试间隔逐次拉长，避免短时间内反复冲击已经出问题的接口。on_failure_callback挂一个告警函数，失败时发通知，别等第二天才发现任务挂了。

7. 那些文档里不会写的坑

7.1 状态污染：最隐蔽的 bug

LangGraph 的状态在节点间传递，如果你在节点里直接修改了传入的列表或字典，可能污染上游数据。看这个例子：

def bad_node(state): items = state["items"] items.append("new") # 直接改了原列表 return {"items": items}

如果items是可变对象，这个 append 会影响到所有引用它的地方。正确做法是返回新对象：

def good_node(state): new_items = state["items"] + ["new"] return {"items": new_items}

这个坑特别隐蔽，因为单次运行可能看不出问题，一旦流程有分支或循环，就会出现"数据莫名其妙多了几条"的诡异现象。我排查过一次，花了整整一下午才定位到是状态被就地修改了。

7.2 模型输出的"薛定谔格式"

即使你用了 JSON 解析器，模型偶尔还是会返回带 markdown 代码块包裹的 JSON，比如```json ... ```。解析器直接解析会失败。稳妥做法是先清洗：

import re import json def clean_json_output(text: str) -> dict: # 去掉 markdown 代码块标记 text = re.sub(r"```json\s*|\s*```", "", text).strip() try: return json.loads(text) except json.JSONDecodeError: # 尝试提取第一个完整 JSON 对象 match = re.search(r"\{.*\}", text, re.DOTALL) if match: return json.loads(match.group()) raise

这个清洗函数我几乎每个项目都会放一份。别指望模型永远听话，做好兜底才是工程思维。

7.3 上下文长度：不是越长越好

很多人以为把整篇文档塞给模型效果最好，其实不然。上下文太长有三个问题：成本高、速度慢、模型注意力被稀释导致关键信息被忽略。

正确做法是分块 + 检索。把长文档切成合适大小的块，用向量检索找出和当前任务最相关的几块，只把这几块喂给模型。这就是 RAG 的基本思路。分块大小一般 500 到 1000 字符，块之间留一点重叠（比如 100 字符），避免关键信息被切断。

7.4 成本失控：跑起来才知道贵

工作流一旦自动化，调用量会指数级上升。我见过有人测试时没注意，一个循环跑了几千次模型调用，账单出来吓一跳。几个控制成本的手段：

用便宜的小模型做预处理和分类，只在关键环节用大模型
缓存重复的调用结果，同样的输入别重复问
设调用次数上限，超过就熔断
记录每次调用的 token 消耗，定期复盘

from functools import lru_cache @lru_cache(maxsize=1000) def cached_model_call(prompt_text: str) -> str: return llm.invoke(prompt_text).content

lru_cache对纯函数式的调用很有效，但要注意它按参数缓存，如果参数里有不可哈希的对象会报错。

8. 从能跑到好用：几个进阶方向

8.1 人工介入节点的正确姿势

human-in-the-loop 不是简单加个"等待确认"，而是要考虑：确认期间状态怎么保存、确认后从哪继续、超时怎么办。LangGraph 支持在节点间中断并保存状态，恢复时从断点继续：

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() app = graph.compile( checkpointer=memory, interrupt_before=["manual_review"], # 在人工审核节点前中断 ) # 第一次运行，会在 manual_review 前停下 config = {"configurable": {"thread_id": "task_001"}} app.invoke({"raw_text": "..."}, config) # 人工确认后，继续执行 app.invoke(None, config)

thread_id是恢复执行的钥匙，同一个 thread_id 才能接上之前的状态。生产环境别用 MemorySaver，它存在内存里，进程重启就没了，要用数据库持久化的 checkpointer。

8.2 可观测性：让流程透明

工作流跑起来之后，你需要知道：每次运行走了哪条路径、每步耗时多少、模型调用花了多少 token、哪一步最容易失败。这些靠 print 是不够的，要接专门的追踪工具。LangSmith 是官方方案，能可视化整条链路。如果不想用外部服务，至少自己记录结构化日志：

import logging import time logger = logging.getLogger("workflow") def timed_node(func): def wrapper(state): start = time.time() result = func(state) elapsed = time.time() - start logger.info(f"node={func.__name__} elapsed={elapsed:.2f}s") return result return wrapper

给每个节点加个计时装饰器，跑一段时间后你就能看出瓶颈在哪。我一般会重点盯模型调用和外部接口这两类节点，它们通常是最慢的。

8.3 版本管理与灰度

工作流上线后，提示词改了、模型换了、逻辑调了，怎么保证不出事？答案是版本化 + 灰度。

提示词和流程配置都进 Git，每次改动有记录、可回滚。新版本先在小流量上跑，对比效果和成本，确认没问题再全量。别直接在生产上改，改完出问题连回滚都找不到旧版本。

# 用配置区分版本 PROMPT_VERSION = os.getenv("PROMPT_VERSION", "v1") PROMPTS = { "v1": PROMPT_V1, "v2": PROMPT_V2, } current_prompt = PROMPTS[PROMPT_VERSION]

这样切换版本只改环境变量，不用动代码，灰度时也方便按流量比例分配。

8.4 什么时候该考虑换方案

工作流不是越复杂越好。如果你发现流程里节点越来越多、条件分支越来越绕、维护成本超过收益，可能是时候重新审视了。几个信号：改一个需求要动五六个节点、新人看不懂流程图、调试一次要跑半小时。这时候要么简化流程，要么换更适合的编排方式。

我个人的判断标准是：如果一个流程的复杂度已经超过它带来的自动化收益，就该砍掉重来。工作流是手段不是目的，能稳定解决问题才是关键。

最后分享一个我踩过好几次才养成的习惯：任何工作流上线前，先用异常输入跑一遍。空输入、超长输入、格式错误的输入、包含特殊字符的输入，这些才是真实环境里最常见的。正常输入跑通不算本事，异常输入不崩才是。我现在的习惯是每个工作流都配一组边界测试用例，改完代码先跑这组，通过了再上。这个习惯帮我挡掉了不少线上事故。