在构建下一代 AI 智能体系统时，工作流的表达方式直接决定了开发效率、系统可维护性和扩展能力。LangGraph 作为当前最成熟的智能体编排框架之一，提供了两种风格迥异但深度兼容的 API：Graph API与Functional API。它们不是“你选一个”的二选一，而是像“油画笔”与“喷枪”一样——在同一个画布上，你完全可以交替使用，甚至混合使用，以实现最精准的表达。
本文将彻底拆解这两种 API 的底层逻辑、适用边界、工程实践与真实组合方式，并提供多语言实现示例（Python、TypeScript、Go 模拟），助你从“会用”走向“精通”。
1. Graph API：声明式工作流的“指挥中心”
1.1 核心哲学：状态即图，边即决策
Graph API 的本质，是将程序逻辑映射为有向图。每个节点是原子任务，每条边是条件分支或依赖关系。它不关心你如何写循环，它只关心：在什么状态下，走哪条路，下一步执行谁。
🚀 关键洞察：Graph API 的“状态”不是变量，是全局上下文快照，它像一个黑板，所有节点都能读写，所有分支都能感知。
1.2 为什么它适合复杂系统？
| 问题 | 传统代码 | Graph API |
|---|---|---|
| 多分支逻辑（10+条件） | 嵌套 if-elif-else，难以维护 | 图结构清晰，分支一目了然 |
| 并行任务（5个独立 API 调用） | 手动写 asyncio 或线程池 | 用add_edge并行启动，自动合并 |
| 状态传递（A→B→C 传递中间结果） | 用参数传递，易出错 | 全局状态自动流转 |
| 团队协作 | 代码文档+注释 | 可视化图 + 导出 PNG/SVG |
1.3 实战示例：金融风控智能体（Python）
fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,List,OptionalimportasyncioclassAgentState(TypedDict):user_id:strtransaction_amount:floatlocation:strdevice_fingerprint:strrisk_score:floatdecisions:List[str]final_action:Optional[str]retry_count:int# 定义节点defcheck_fraud_rules(state:AgentState)->AgentState:score=0ifstate["transaction_amount"]>10000:score+=40ifstate["location"]notin["US","CA","JP"]:score+=30if"emulator"instate["device_fingerprint"]:score+=50return{**state,"risk_score":score,"decisions":state["decisions"]+["rules_checked"]}deftrigger_manual_review(state:AgentState)->AgentState:return{**state,"final_action":"manual_review","decisions":state["decisions"]+["triggered_review"]}defauto_approve(state:AgentState)->AgentState:return{**state,"final_action":"approved","decisions":state["decisions"]+["auto_approved"]}defblock_transaction(state:AgentState)->AgentState:return{**state,"final_action":"blocked","decisions":state["decisions"]+["blocked"]}# 条件分支逻辑defroute_decision(state:AgentState)->str:ifstate["risk_score"]>=80:return"block_transaction"elifstate["risk_score"]>=50:return"trigger_manual_review"elifstate["risk_score"]>=20:return"auto_approve"else:returnEND# 构建图workflow=StateGraph(AgentState)workflow.add_node("check_fraud_rules",check_fraud_rules)workflow.add_node("trigger_manual_review",trigger_manual_review)workflow.add_node("auto_approve",auto_approve)workflow.add_node("block_transaction",block_transaction)workflow.set_entry_point("check_fraud_rules")workflow.add_conditional_edges("check_fraud_rules",route_decision)workflow.add_edge("trigger_manual_review",END)workflow.add_edge("auto_approve",END)workflow.add_edge("block_transaction",END)# 编译并执行app=workflow.compile()result=app.invoke({"user_id":"user_123","transaction_amount":15000,"location":"NG","device_fingerprint":"android_emulator","risk_score":0,"decisions":[],"retry_count":0})print(result["final_action"])# 输出: blockedprint(result["decisions"])# 输出: ['rules_checked', 'blocked']1.4 并行处理实战：多源数据聚合
# 并行获取：新闻、天气、股票、汇率deffetch_news(state:AgentState)->AgentState:# 模拟异步调用return{**state,"news":["Breaking: AI regulation passed"],"fetched_news":True}deffetch_weather(state:AgentState)->AgentState:return{**state,"weather":"Sunny in New York","fetched_weather":True}deffetch_stocks(state:AgentState)->AgentState:return{**state,"stocks":{"AAPL":198.5,"TSLA":245.3},"fetched_stocks":True}deffetch_exchange(state:AgentState)->AgentState:return{**state,"exchange":{"USD/EUR":0.92,"USD/JPY":151.2},"fetched_exchange":True}defcombine_all(state:AgentState)->AgentState:return{**state,"summary":f"Today:{state['weather']}. Markets:{state['stocks']}. News:{state['news'][0]}. Rate:{state['exchange']}"}# 构建并行流workflow=StateGraph(AgentState)workflow.add_node("fetch_news",fetch_news)workflow.add_node("fetch_weather",fetch_weather)workflow.add_node("fetch_stocks",fetch_stocks)workflow.add_node("fetch_exchange",fetch_exchange)workflow.add_node("combine_all",combine_all)# 并行启动：从 START 同时触发4个节点workflow.add_edge("START","fetch_news")workflow.add_edge("START","fetch_weather")workflow.add_edge("START","fetch_stocks")workflow.add_edge("START","fetch_exchange")# 所有并行节点完成后，统一合并workflow.add_edge("fetch_news","combine_all")workflow.add_edge("fetch_weather","combine_all")workflow.add_edge("fetch_stocks","combine_all")workflow.add_edge("fetch_exchange","combine_all")workflow.set_entry_point("START")app=workflow.compile()result=app.invoke({"START":True})print(result["summary"])# 输出: Today: Sunny in New York. Markets: {'AAPL': 198.5, 'TSLA': 245.3}. News: Breaking: AI regulation passed. Rate: {'USD/EUR': 0.92, 'USD/JPY': 151.2}✅ 并行节点无需手动 await，LangGraph 自动调度，支持异步/同步混合。
2. Functional API：命令式的“快速脚本引擎”
2.1 核心哲学：你写什么，它就跑什么
Functional API 不是“图”，它是带检查点的函数包装器。它不强制你定义状态结构，不强制你写边，它只是让你的函数可恢复、可重试、可流式输出。
🚀 关键洞察：Functional API 是你现有 Python 函数的“增强版”。你写
def foo(x): return x+1，它变成@entrypoint def foo(x): return x+1，仅此而已。
2.2 为什么它适合快速迭代？
| 场景 | Graph API | Functional API |
|---|---|---|
| 3行逻辑处理数据 | 需定义 State、Node、Edge | @entrypoint一行搞定 |
| 改造已有业务逻辑 | 重构成本高 | 直接装饰原有函数 |
| 实验性原型 | 太重 | 5分钟上线 |
| 无状态任务 | 强制共享状态 | 函数内局部变量即可 |
2.3 实战示例：用户输入清洗器（Python）
fromlanggraph.funcimportentrypoint,taskimportre@taskdefclean_input(text:str)->dict:return{"cleaned":re.sub(r'\s+',' ',text.strip()),"length":len(text),"has_numbers":any(c.isdigit()forcintext)}@taskdefclassify_intent(cleaned:dict)->str:keywords=["help","support","refund","cancel"]ifany(kincleaned["cleaned"].lower()forkinkeywords):return"customer_service"return"general"@entrypoint(checkpointer="sqlite:///langgraph_checkpoints.db")defprocess_user_query(user_input:str)->dict:cleaned=clean_input(user_input).result()intent=classify_intent(cleaned).result()ifintent=="customer_service":response="Our team will reach out within 24 hours."else:response="Thanks for your message! We've logged your feedback."return{"original":user_input,"cleaned":cleaned["cleaned"],"intent":intent,"response":response,"processed_at":"2025-04-05T10:30:00Z"}# 调用result=process_user_query(" I want to CANCEL my subscription ")print(result["response"])# 输出: Our team will reach out within 24 hours.2.4 异步支持：Python + Node.js + Go 模拟
Python（原生）
@taskasyncdeffetch_api_async(endpoint:str):response=awaithttpx.AsyncClient().get(endpoint)returnresponse.json()TypeScript（LangGraph.js 模拟）
import{entrypoint,task}from"@langgraph/functional";@task()asyncfunctionfetchUser(id:string){constres=awaitfetch(`/api/users/${id}`);returnawaitres.json();}@entrypoint()asyncfunctionworkflow(id:string){constuser=awaitfetchUser(id);constposts=awaitfetchPosts(user.id);return{user,posts};}Go（伪代码，模拟 Functional API）
// 伪代码：LangGraph Go 版本可能的形态funcProcessUser(ctx context.Context,inputstring)(Output,error){cleaned:=CleanInput(ctx,input)intent:=ClassifyIntent(ctx,cleaned)ifintent=="support"{returnSendSupportResponse(ctx,input)}returnSendGenericResponse(ctx,input)}// 通过 LangGraph Go SDK，自动支持 checkpointing 和重试💡 重点：Functional API 的“跨语言潜力”远超 Graph API。它本质上是可序列化的函数调用链，未来可支持 Rust、Java、C# 等语言无缝接入。
3. 混合使用：Graph + Functional 的“黄金组合”
场景：智能客服系统
- Graph API：负责多智能体协调（意图识别 → 情绪分析 → 工单生成 → 人工转接）
- Functional API：负责每一个子任务的内部处理（如：正则清洗、情感打分、模板生成）
fromlanggraph.graphimportStateGraphfromlanggraph.funcimportentrypoint,task# Functional API：纯数据处理@taskdefextract_entities(text:str)->dict:# 使用 spaCy/NLP 模型return{"entities":["Apple","iPhone 15"],"confidence":0.92}@taskdefsentiment_score(text:str)->float:return0.87# 假设为积极情绪@taskdefgenerate_ticket_body(entities:dict,sentiment:float)->str:returnf"User reported issue with{', '.join(entities['entities'])}. Sentiment:{sentiment}"# Graph API：协调流程classSupportState(TypedDict):user_message:strentities:dictsentiment:floatticket_body:strassigned_agent:strdefintent_router(state:SupportState)->str:ifstate["sentiment"]<0.3:return"escalate_to_human"else:return"auto_create_ticket"defescalate_to_human(state:SupportState)->SupportState:return{**state,"assigned_agent":"support_team_lead"}defauto_create_ticket(state:SupportState)->SupportState:ticket=generate_ticket_body.invoke({"entities":state["entities"],"sentiment":state["sentiment"]})return{**state,"ticket_body":ticket["result"],"assigned_agent":"auto_system"}workflow=StateGraph(SupportState)workflow.add_node("extract_entities",extract_entities)workflow.add_node("sentiment_score",sentiment_score)workflow.add_node("auto_create_ticket",auto_create_ticket)workflow.add_node("escalate_to_human",escalate_to_human)workflow.set_entry_point("extract_entities")workflow.add_edge("extract_entities","sentiment_score")workflow.add_conditional_edges("sentiment_score",intent_router)workflow.add_edge("auto_create_ticket",END)workflow.add_edge("escalate_to_human",END)app=workflow.compile()result=app.invoke({"user_message":"I love my new iPhone 15! But the camera is blurry."})print(result["ticket_body"])# 输出: User reported issue with Apple, iPhone 15. Sentiment: 0.87✅ 你看到的
generate_ticket_body.invoke(...)，就是 Functional API 在 Graph 节点中的无缝嵌入。
4. API 迁移指南：何时该“升级”或“降级”？
| 你当前的痛点 | 推荐迁移方向 |
|---|---|
| 代码里有 7 层嵌套 if-else | → 用 Graph API 拆成可视化分支 |
| 多个函数之间靠全局变量传状态 | → 用 Graph API 定义 State 类 |
| 想并行调用 4 个 API，但怕出错 | → 用 Graph API 的并行边 + 自动重试 |
| 你只想把一个函数加个“重试+保存状态” | → 用@entrypoint一行搞定 |
| 你正在写一个 5 行的脚本，但想让它能恢复 | → Functional API 最轻量 |
| 你团队里有人看不懂代码，但能看图 | → Graph API 导出 PNG，直接上会议白板 |
| 你发现图越来越复杂，但逻辑其实很简单 | → 降级为 Functional API，删除所有节点定义 |
🔥经验法则：
- 如果你的流程能画在一张 A4 纸上，用 Functional API
- 如果你的流程需要打印出来贴墙上，用 Graph API
5. 总结：两种 API 的终极对比表
| 维度 | Graph API | Functional API |
|---|---|---|
| 编程范式 | 声明式（What） | 命令式（How） |
| 状态管理 | 全局共享、强类型 State | 函数局部、动态字典 |
| 分支逻辑 | 多条件、可视化路由 | 简单 if/else、循环 |
| 并行支持 | ✅ 原生支持，自动调度 | ❌ 需手动用 asyncio/threading |
| 学习曲线 | 中高（需理解图结构） | 极低（就是普通函数） |
| 调试体验 | 可视化流程 + 状态快照 | 传统 print/debugger |
| 团队协作 | ✅ 图可导出、可评审 | ✅ 代码即文档，适合单人 |
| 适用场景 | 多智能体、复杂决策、长期运行 | 快速原型、脚本封装、数据清洗 |
| 持久化 | ✅ 内置 Checkpointer | ✅ 同样支持 |
| 流式输出 | ✅ 支持 | ✅ 支持 |
| 人机交互 | ✅ 支持暂停/人工介入 | ✅ 支持 |
| 语言生态 | Python 主导 | 支持多语言（JS/Go/Rust 未来可期） |
✅ 最终建议：不要选，要搭配
Graph API 是你的“指挥官”—— 管理流程、协调智能体、处理分支。
Functional API 是你的“士兵”—— 执行具体任务、处理数据、调用 API。
你不需要在“用哪个”之间纠结。
你应该问：
- 这个任务复杂吗？→ 用 Graph
- 这个任务只是个函数？→ 用 Functional
- 这个任务要被多个流程复用？→ 用 Functional 包装，Graph 调用
- 我想让非工程师也能理解流程？→ 用 Graph + 导出 PNG
- 我只想给现有代码加个“断点恢复”？→ 一行
@entrypoint
📌 结语：智能体的未来，是“可组合的函数”与“可视化的流程”
LangGraph 的双 API 设计，不是冗余，而是工程哲学的胜利：
你不必为了“系统化”而牺牲“敏捷性”，也不必为了“快速”而放弃“可控性”。
在真实生产系统中，最强大的智能体，往往是一个 Graph 主干 + 数十个 Functional 细胞构成的有机体。
从今天开始，别再问“我该用哪个”，
问：“我这个任务，是该画图，还是该写函数？”
然后，两个都用。
