在构建下一代 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）

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Optional
import asyncio

class AgentState(TypedDict):
    user_id: str
    transaction_amount: float
    location: str
    device_fingerprint: str
    risk_score: float
    decisions: List[str]
    final_action: Optional[str]
    retry_count: int

# 定义节点
def check_fraud_rules(state: AgentState) -> AgentState:
    score = 0
    if state["transaction_amount"] > 10000:
        score += 40
    if state["location"] not in ["US", "CA", "JP"]:
        score += 30
    if "emulator" in state["device_fingerprint"]:
        score += 50
    return {**state, "risk_score": score, "decisions": state["decisions"] + ["rules_checked"]}

def trigger_manual_review(state: AgentState) -> AgentState:
    return {**state, "final_action": "manual_review", "decisions": state["decisions"] + ["triggered_review"]}

def auto_approve(state: AgentState) -> AgentState:
    return {**state, "final_action": "approved", "decisions": state["decisions"] + ["auto_approved"]}

def block_transaction(state: AgentState) -> AgentState:
    return {**state, "final_action": "blocked", "decisions": state["decisions"] + ["blocked"]}

# 条件分支逻辑
def route_decision(state: AgentState) -> str:
    if state["risk_score"] >= 80:
        return "block_transaction"
    elif state["risk_score"] >= 50:
        return "trigger_manual_review"
    elif state["risk_score"] >= 20:
        return "auto_approve"
    else:
        return END

# 构建图
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"])  # 输出: blocked
print(result["decisions"])     # 输出: ['rules_checked', 'blocked']

1.4 并行处理实战：多源数据聚合

# 并行获取：新闻、天气、股票、汇率
def fetch_news(state: AgentState) -> AgentState:
    # 模拟异步调用
    return {**state, "news": ["Breaking: AI regulation passed"], "fetched_news": True}

def fetch_weather(state: AgentState) -> AgentState:
    return {**state, "weather": "Sunny in New York", "fetched_weather": True}

def fetch_stocks(state: AgentState) -> AgentState:
    return {**state, "stocks": {"AAPL": 198.5, "TSLA": 245.3}, "fetched_stocks": True}

def fetch_exchange(state: AgentState) -> AgentState:
    return {**state, "exchange": {"USD/EUR": 0.92, "USD/JPY": 151.2}, "fetched_exchange": True}

def combine_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）

from langgraph.func import entrypoint, task
import re

@task
def clean_input(text: str) -> dict:
    return {
        "cleaned": re.sub(r'\s+', ' ', text.strip()),
        "length": len(text),
        "has_numbers": any(c.isdigit() for c in text)
    }

@task
def classify_intent(cleaned: dict) -> str:
    keywords = ["help", "support", "refund", "cancel"]
    if any(k in cleaned["cleaned"].lower() for k in keywords):
        return "customer_service"
    return "general"

@entrypoint(checkpointer="sqlite:///langgraph_checkpoints.db")
def process_user_query(user_input: str) -> dict:
    cleaned = clean_input(user_input).result()
    intent = classify_intent(cleaned).result()
    
    if intent == "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（原生）

@task
async def fetch_api_async(endpoint: str):
    response = await httpx.AsyncClient().get(endpoint)
    return response.json()

TypeScript（LangGraph.js 模拟）

import { entrypoint, task } from "@langgraph/functional";

@task()
async function fetchUser(id: string) {
  const res = await fetch(`/api/users/${id}`);
  return await res.json();
}

@entrypoint()
async function workflow(id: string) {
  const user = await fetchUser(id);
  const posts = await fetchPosts(user.id);
  return { user, posts };
}

Go（伪代码，模拟 Functional API）

// 伪代码：LangGraph Go 版本可能的形态
func ProcessUser(ctx context.Context, input string) (Output, error) {
    cleaned := CleanInput(ctx, input)
    intent := ClassifyIntent(ctx, cleaned)
    
    if intent == "support" {
        return SendSupportResponse(ctx, input)
    }
    return SendGenericResponse(ctx, input)
}
// 通过 LangGraph Go SDK，自动支持 checkpointing 和重试

💡 重点：Functional API 的“跨语言潜力”远超 Graph API。它本质上是可序列化的函数调用链，未来可支持 Rust、Java、C# 等语言无缝接入。

---

3. 混合使用：Graph + Functional 的“黄金组合”

场景：智能客服系统

• Graph API：负责多智能体协调（意图识别 → 情绪分析 → 工单生成 → 人工转接）
• Functional API：负责每一个子任务的内部处理（如：正则清洗、情感打分、模板生成）

from langgraph.graph import StateGraph
from langgraph.func import entrypoint, task

# Functional API：纯数据处理
@task
def extract_entities(text: str) -> dict:
    # 使用 spaCy/NLP 模型
    return {"entities": ["Apple", "iPhone 15"], "confidence": 0.92}

@task
def sentiment_score(text: str) -> float:
    return 0.87  # 假设为积极情绪

@task
def generate_ticket_body(entities: dict, sentiment: float) -> str:
    return f"User reported issue with {', '.join(entities['entities'])}. Sentiment: {sentiment}"

# Graph API：协调流程
class SupportState(TypedDict):
    user_message: str
    entities: dict
    sentiment: float
    ticket_body: str
    assigned_agent: str

def intent_router(state: SupportState) -> str:
    if state["sentiment"] < 0.3:
        return "escalate_to_human"
    else:
        return "auto_create_ticket"

def escalate_to_human(state: SupportState) -> SupportState:
    return {**state, "assigned_agent": "support_team_lead"}

def auto_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 细胞构成的有机体。

从今天开始，别再问“我该用哪个”，
问：“我这个任务，是该画图，还是该写函数？”

然后，两个都用。