179 lines
7.8 KiB
Python
179 lines
7.8 KiB
Python
# requires: langgraph, pydantic
|
|
"""工单处理图：结构化分类 → 起草回复（失败自动重试）→ 高风险人工审批（interrupt）→ 发送（幂等）。
|
|
|
|
演示：任务状态字段、流式进度（custom 流）、检查点续跑（SqliteSaver）、人机协作、异常恢复。
|
|
不调用真实大模型：classify / draft 用规则模拟，接口形状和真实做法一致。
|
|
运行：python ticket_flow.py
|
|
基于 langgraph 1.2 编写，API 以官方文档为准：https://docs.langchain.com/oss/python/langgraph/overview
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import operator
|
|
import os
|
|
import re
|
|
import tempfile
|
|
from typing import Annotated, Literal, TypedDict
|
|
|
|
from langgraph.checkpoint.sqlite import SqliteSaver
|
|
from langgraph.config import get_stream_writer
|
|
from langgraph.graph import END, START, StateGraph
|
|
from langgraph.types import Command, RetryPolicy, interrupt
|
|
from pydantic import BaseModel, Field, ValidationError
|
|
|
|
Status = Literal["new", "classified", "drafted", "waiting_approval", "approved", "rejected", "sent", "failed"]
|
|
|
|
|
|
class Triage(BaseModel):
|
|
"""分类结果：真实项目里是大模型的结构化输出，这里用 pydantic 校验它。"""
|
|
category: Literal["refund", "bug", "question"]
|
|
amount: float = Field(ge=0, default=0)
|
|
urgent: bool = False
|
|
|
|
|
|
class TicketState(TypedDict, total=False):
|
|
ticket_id: str
|
|
text: str
|
|
triage: dict
|
|
draft: str
|
|
status: Status
|
|
reviewer_note: str
|
|
events: Annotated[list[str], operator.add] # 追加式字段：每个节点追加一条，作为审计记录
|
|
|
|
|
|
class TransientError(Exception):
|
|
"""可重试的错误：超时、限流、5xx。"""
|
|
|
|
|
|
# ---------- 外部依赖（测试里可以替换） ----------
|
|
class FakeLLM:
|
|
def __init__(self, fail_times: int = 0):
|
|
self.fail_times = fail_times
|
|
self.calls = 0
|
|
|
|
def classify(self, text: str) -> str:
|
|
# 真实模型返回 JSON 字符串；这里也返回字符串，交给 pydantic 校验
|
|
if "退款" in text:
|
|
m = re.search(r"(\d+(?:\.\d+)?)\s*元", text)
|
|
amount = float(m.group(1)) if m else 0
|
|
return f'{{"category": "refund", "amount": {amount}, "urgent": {str("加急" in text).lower()}}}'
|
|
if "报错" in text:
|
|
return '{"category": "bug", "urgent": true}'
|
|
return '{"category": "question"}'
|
|
|
|
def draft(self, text: str, triage: Triage) -> str:
|
|
self.calls += 1
|
|
if self.calls <= self.fail_times:
|
|
raise TransientError("模型接口超时")
|
|
return f"[{triage.category}] 您好，关于“{text}”，我们已经处理，请留意通知。"
|
|
|
|
|
|
class Mailer:
|
|
"""发送通道。用 ticket_id 做幂等键：同一个工单重复执行也只发一次。"""
|
|
def __init__(self, crash_once: bool = False):
|
|
self.sent: dict[str, str] = {}
|
|
self.crash_once = crash_once
|
|
|
|
def send(self, ticket_id: str, body: str) -> None:
|
|
if ticket_id in self.sent:
|
|
return
|
|
if self.crash_once:
|
|
self.crash_once = False
|
|
raise RuntimeError("进程崩溃（模拟）")
|
|
self.sent[ticket_id] = body
|
|
|
|
|
|
APPROVAL_AMOUNT = 500
|
|
|
|
|
|
def needs_approval(triage: dict) -> bool:
|
|
return triage["category"] == "refund" and triage["amount"] >= APPROVAL_AMOUNT
|
|
|
|
|
|
def build(llm: FakeLLM, mailer: Mailer, checkpointer):
|
|
def classify(state: TicketState):
|
|
get_stream_writer()({"step": "classify", "msg": "正在分类"})
|
|
raw = llm.classify(state["text"])
|
|
try:
|
|
triage = Triage.model_validate_json(raw)
|
|
except ValidationError as e: # 结构化输出不合法：降级为人工处理，而不是让流程崩掉
|
|
return {"status": "failed", "events": [f"classify_invalid:{e.error_count()}"]}
|
|
return {"triage": triage.model_dump(), "status": "classified", "events": [f"classified:{triage.category}"]}
|
|
|
|
def draft(state: TicketState):
|
|
get_stream_writer()({"step": "draft", "msg": "正在起草回复"})
|
|
text = llm.draft(state["text"], Triage(**state["triage"]))
|
|
status = "waiting_approval" if needs_approval(state["triage"]) else "drafted"
|
|
return {"draft": text, "status": status, "events": ["drafted"]}
|
|
|
|
def need_approval(state: TicketState) -> str: # 条件边只做路由，不修改状态
|
|
return "approve" if needs_approval(state["triage"]) else "send"
|
|
|
|
def approve(state: TicketState):
|
|
# interrupt 之前的代码在恢复时会重新执行一次，所以这里不要放有副作用的操作
|
|
decision = interrupt({"ticket_id": state["ticket_id"], "draft": state["draft"],
|
|
"amount": state["triage"]["amount"]})
|
|
if decision.get("ok"):
|
|
return {"status": "approved", "reviewer_note": decision.get("note", ""), "events": ["approved"]}
|
|
return {"status": "rejected", "reviewer_note": decision.get("note", ""), "events": ["rejected"]}
|
|
|
|
def after_approve(state: TicketState) -> str:
|
|
return "send" if state["status"] == "approved" else END
|
|
|
|
def send(state: TicketState):
|
|
get_stream_writer()({"step": "send", "msg": "正在发送"})
|
|
mailer.send(state["ticket_id"], state["draft"])
|
|
return {"status": "sent", "events": ["sent"]}
|
|
|
|
g = StateGraph(TicketState)
|
|
g.add_node("classify", classify)
|
|
g.add_node("draft", draft, retry_policy=RetryPolicy(max_attempts=3, initial_interval=0.01,
|
|
retry_on=TransientError))
|
|
g.add_node("approve", approve)
|
|
g.add_node("send", send)
|
|
g.add_edge(START, "classify")
|
|
g.add_conditional_edges("classify", lambda s: END if s["status"] == "failed" else "draft")
|
|
g.add_conditional_edges("draft", need_approval, ["approve", "send"])
|
|
g.add_conditional_edges("approve", after_approve, ["send", END])
|
|
g.add_edge("send", END)
|
|
return g.compile(checkpointer=checkpointer)
|
|
|
|
|
|
def main() -> None:
|
|
db = os.path.join(tempfile.mkdtemp(), "checkpoints.sqlite")
|
|
llm, mailer = FakeLLM(fail_times=1), Mailer(crash_once=True)
|
|
cfg = {"configurable": {"thread_id": "T-1001"}}
|
|
ticket = {"ticket_id": "T-1001", "text": "订单 A1 申请退款 800 元", "status": "new", "events": []}
|
|
|
|
# 第一次运行：draft 第一次超时被自动重试；遇到 800 元退款，停在人工审批
|
|
with SqliteSaver.from_conn_string(db) as cp:
|
|
app = build(llm, mailer, cp)
|
|
for mode, chunk in app.stream(ticket, cfg, stream_mode=["custom", "updates"]):
|
|
if mode == "custom":
|
|
print("进度：", chunk["msg"])
|
|
snap = app.get_state(cfg)
|
|
print("等待节点：", snap.next, "状态：", snap.values["status"])
|
|
assert snap.values["status"] == "waiting_approval" and snap.values["triage"]["amount"] == 800
|
|
print("待审批内容：", snap.tasks[0].interrupts[0].value)
|
|
assert snap.next == ("approve",) and llm.calls == 2
|
|
|
|
# 审批人第二天才处理：新的进程、新的图对象，只要 thread_id 一样就能接着跑
|
|
with SqliteSaver.from_conn_string(db) as cp:
|
|
app = build(llm, mailer, cp)
|
|
try:
|
|
app.invoke(Command(resume={"ok": True, "note": "金额核对无误"}), cfg)
|
|
except RuntimeError as e:
|
|
print("发送时崩溃：", e)
|
|
snap = app.get_state(cfg)
|
|
print("崩溃后停在：", snap.next, "状态：", snap.values["status"])
|
|
|
|
# 崩溃恢复：传 None 表示从最近的检查点继续，已经完成的节点（起草、审批）不会重新执行
|
|
with SqliteSaver.from_conn_string(db) as cp:
|
|
app = build(llm, mailer, cp)
|
|
final = app.invoke(None, cfg)
|
|
print("最终状态：", final["status"], "事件：", final["events"])
|
|
assert final["status"] == "sent" and llm.calls == 2 and "T-1001" in mailer.sent
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|