Temporal外层编排

Agent 单次 invoke 在分钟级可接受；跨小时、跨服务、需补偿的订单 Saga 需要 Temporal 等 durable workflow（持久化工作流）引擎。LangGraph 管单进程内细粒度图；Temporal 管外层长事务与重试策略——二者分层而非互替。

段末注释：Temporal = 开源/workflow-as-code 平台，Activity 失败可自动重试、进程重启后续跑。

1. 定位

维度	内容
角色	外层长事务编排；内层 LangGraph 作为一步 Activity
输入 → 输出	Workflow 调 Activity → `graph.invoke` → 结果回 Workflow
核心 API	Temporal Workflow/Activity；LangGraph `invoke`
依赖 LangChain	Activity 内可用 LangGraph + checkpointer

2. 分层拓扑

Temporal Workflow（外层）

步骤	职责
`validate_order`	Activity：校验
`run_agent`	Activity：LangGraph invoke
`notify`	Activity：通知

LangGraph 内层（Agent 图）

节点	职责
`plan`	mock 规划
`act`	mock 执行

边（内层）：START → plan → act → END

3. 生命周期

1. Temporal 启动 Workflow（订单 ID = business key）
2. Activity run_agent：构造 thread_id=order-123
3. langgraph.invoke → checkpoint 可写 Postgres
4. Activity 成功 → Workflow 继续 notify
5. Worker 崩溃：Temporal 重试 Activity；LangGraph 靠 checkpoint 同一 thread 续跑

Superstep 仍在 LangGraph 内；Temporal 粒度是 Activity 整次 invoke。

4. 原理

4.1 边界划分

层	负责
Temporal	跨服务、定时、Saga、全局 idempotency
LangGraph	LLM/工具环、HITL interrupt、消息 state

4.2 thread_id 映射

建议 thread_id = f"{workflow_id}:{run_id}" 与业务订单对齐。

4.3 HITL 注意

LangGraph interrupt 暂停时 Activity 应心跳/短超时或拆成「提交图 + 轮询 checkpoint」两 Activity，避免 Temporal 误判 Activity 超时。

5. 最小可运行示例（概念代码）

# --- LangGraph 内层（可独立单测）---
from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END


class AgentState(TypedDict):
    task: str
    outcome: str


def plan(state: AgentState) -> dict:
    return {"outcome": f"planned:{state['task']}"}


def act(state: AgentState) -> dict:
    return {"outcome": state["outcome"] + "|acted"}


b = StateGraph(AgentState)
b.add_node("plan", plan)
b.add_node("act", act)
b.add_edge(START, "plan")
b.add_edge("plan", "act")
b.add_edge("act", END)
agent_graph = b.compile(checkpointer=InMemorySaver())


def run_langgraph_activity(task: str, thread_id: str) -> str:
    cfg = {"configurable": {"thread_id": thread_id}}
    out = agent_graph.invoke({"task": task, "outcome": ""}, cfg)
    return out["outcome"]


# --- Temporal 外层（示意，需 temporalio SDK 与 Worker）---
# @workflow.defn
# class OrderWorkflow:
#     @workflow.run
#     async def run(self, order_id: str) -> str:
#         await workflow.execute_activity(validate_order, order_id, ...)
#         result = await workflow.execute_activity(
#             run_langgraph_activity, order_id, f"order-{order_id}", ...
#         )
#         await workflow.execute_activity(notify, result, ...)
#         return result

6. 执行追踪

层	步骤	关键状态
Temporal	validate	通过
LangGraph	plan	outcome=planned:pay
LangGraph	act	outcome=planned:pay\|acted
Temporal	notify	完成

重要配置参数

参数	类型 / 默认	作用与影响	参考起点	配置指导
Activity `start_to_close_timeout`	duration	含 LLM 时要够长	5–30 min	HITL 更长或拆分
LangGraph `thread_id`	str	与订单绑定	order-{id}	幂等
Postgres checkpointer	Saver	Activity 重试不丢态	11 专篇	与 Temporal 共库分离 schema
Activity 心跳	Temporal	长 LLM	interrupt 场景	防误杀
Workflow id	business key	去重	order_id	防双开
LangGraph interrupt	compile	人审	拆两 Activity	勿长阻塞 Activity

7. 易踩坑

在 Temporal Workflow 里直接 invoke LLM：Workflow 必须确定性；LLM 放 Activity。
Activity 超时小于 Agent 运行时间：反复重试放大成本。
无 checkpoint 期望 Activity 重试可续：会从头跑图。

小结

Temporal 外层 = 长事务、跨服务；LangGraph 内层 = Agent 图与 HITL。
Activity 内 invoke + checkpointer + thread_id；Workflow 保持确定性。
interrupt/HITL 需与 Activity 超时/心跳策略协同设计。