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 pipPython 版本建议 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_NAMEgpt-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 扩展后把解释器指向刚才创建的虚拟环境CtrlShiftP→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( modelMODEL_NAME, api_keyOPENAI_API_KEY, base_urlOPENAI_BASE_URL, temperature0.2, ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的文本处理助手只输出要求的结构不要额外解释。), (human, 请把下面这段文本总结成一句话\n\n{text}), ]) chain prompt | llm result chain.invoke({text: 这里放你要处理的文本内容}) print(result.content)这里temperature0.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(description3到5个关键词) sentiment: str Field(description情感倾向正面/中性/负面) parser JsonOutputParser(pydantic_objectSummaryResult) prompt ChatPromptTemplate.from_messages([ (system, 你是一个文本分析助手。{format_instructions}), (human, 分析下面这段文本\n\n{text}), ]) prompt prompt.partial(format_instructionsparser.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 完了 BB 完了 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(modelMODEL_NAME, temperature0) 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_retries2): 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 什么时候需要 AirflowLangGraph 解决的是一次流程怎么跑Airflow 解决的是什么时候跑、跑多少次、失败了怎么办。当你的工作流需要每天定时跑、需要处理一批数据、需要监控每次运行的状态就该上调度器了。Airflow 的核心概念是 DAG有向无环图一个 DAG 就是一条工作流里面每个 task 是一个执行单元。它自带调度、重试、告警、历史记录这些是手写 cron 脚本给不了的。6.2 把 LangGraph 流程包成 Airflow Taskfrom 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(minutes5), } with DAG( dag_idai_workflow_daily, default_argsdefault_args, schedule0 2 * * *, # 每天凌晨2点 start_datedatetime(2024, 1, 1), catchupFalse, ) as dag: task PythonOperator( task_idrun_workflow, python_callablerun_ai_workflow, )catchupFalse很重要。Airflow 默认会补跑历史所有未执行的周期如果你的 start_date 设得很早一上线它会瞬间触发几百次运行。新手经常踩这个坑设成 False 就只跑当前及以后的周期。6.3 批处理的并发控制批量处理时别一股脑全并发出去。模型接口通常有速率限制并发太高会被限流甚至封禁。用 Airflow 的并发参数控制with DAG( dag_idai_batch_process, max_active_runs1, # 同一时间只跑一个 DAG 实例 concurrency5, # 最多 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(minutes2), retry_exponential_backoff: True, # 指数退避 on_failure_callback: send_alert, # 失败告警 }retry_exponential_backoffTrue让重试间隔逐次拉长避免短时间内反复冲击已经出问题的接口。on_failure_callback挂一个告警函数失败时发通知别等第二天才发现任务挂了。7. 那些文档里不会写的坑7.1 状态污染最隐蔽的 bugLangGraph 的状态在节点间传递如果你在节点里直接修改了传入的列表或字典可能污染上游数据。看这个例子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(rjson\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(maxsize1000) def cached_model_call(prompt_text: str) - str: return llm.invoke(prompt_text).contentlru_cache对纯函数式的调用很有效但要注意它按参数缓存如果参数里有不可哈希的对象会报错。8. 从能跑到好用几个进阶方向8.1 人工介入节点的正确姿势human-in-the-loop 不是简单加个等待确认而是要考虑确认期间状态怎么保存、确认后从哪继续、超时怎么办。LangGraph 支持在节点间中断并保存状态恢复时从断点继续from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile( checkpointermemory, 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(fnode{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 什么时候该考虑换方案工作流不是越复杂越好。如果你发现流程里节点越来越多、条件分支越来越绕、维护成本超过收益可能是时候重新审视了。几个信号改一个需求要动五六个节点、新人看不懂流程图、调试一次要跑半小时。这时候要么简化流程要么换更适合的编排方式。我个人的判断标准是如果一个流程的复杂度已经超过它带来的自动化收益就该砍掉重来。工作流是手段不是目的能稳定解决问题才是关键。最后分享一个我踩过好几次才养成的习惯任何工作流上线前先用异常输入跑一遍。空输入、超长输入、格式错误的输入、包含特殊字符的输入这些才是真实环境里最常见的。正常输入跑通不算本事异常输入不崩才是。我现在的习惯是每个工作流都配一组边界测试用例改完代码先跑这组通过了再上。这个习惯帮我挡掉了不少线上事故。