AI 业务流：用确定性状态机包裹概率性智能

# pip install langgraph langchain-openai
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json

class FlowState(TypedDict):
    order_id: str
    reason: str
    order: dict | None
    decision: str | None
    refund_id: str | None
    need_human: bool
    audit: list[str]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def load_order(state: FlowState):
    # 生产环境应查数据库，并做租户隔离
    order = {"id": state["order_id"], "amount": 299,
             "status": "paid", "days_since_purchase": 3}
    return {"order": order, "audit": state.get("audit", []) + ["load_order"]}

def classify(state: FlowState):
    prompt = f"""判断退款理由是否合规，只返回 JSON：
{{"decision":"approve|reject|human","confidence":0-1}}。
理由：{state['reason']}
订单：{json.dumps(state['order'], ensure_ascii=False)}"""
    resp = llm.invoke([HumanMessage(content=prompt)])
    data = json.loads(resp.content)
    need_human = data["decision"] == "human" or data["confidence"] < 0.75
    return {"decision": data["decision"], "need_human": need_human,
            "audit": state["audit"] + ["classify"]}

def approve(state: FlowState):
    # 幂等键：refund_{order_id}，重复执行不会重复退款
    refund_id = f"refund_{state['order_id']}"
    return {"refund_id": refund_id, "audit": state["audit"] + ["approve"]}

def reject(state: FlowState):
    return {"audit": state["audit"] + ["reject"]}

def human_review(state: FlowState):
    # 生产环境应挂起流程，写入人工工单，等待回调后 resume
    return {"decision": "human_pending",
            "audit": state["audit"] + ["human_review"]}

def route(state: FlowState) -> Literal["approve", "reject", "human_review"]:
    if state["need_human"]:
        return "human_review"
    return "approve" if state["decision"] == "approve" else "reject"

builder = StateGraph(FlowState)
builder.add_node("load_order", load_order)
builder.add_node("classify", classify)
builder.add_node("approve", approve)
builder.add_node("reject", reject)
builder.add_node("human_review", human_review)

builder.set_entry_point("load_order")
builder.add_edge("load_order", "classify")
builder.add_conditional_edges("classify", route, {
    "approve": "approve",
    "reject": "reject",
    "human_review": "human_review"
})
builder.add_edge("approve", END)
builder.add_edge("reject", END)
builder.add_edge("human_review", END)

graph = builder.compile()

if __name__ == "__main__":
    result = graph.invoke({
        "order_id": "A1001",
        "reason": "商品破损",
        "audit": []
    })
    print(result)