用状态机组织流程:LangGraph 最小骨架

「从零到 AI 应用工程师」专栏 · 第 21 篇


有了查表 Tool,下一步是固定流程:

START → ingest → classify → lookup → END

每个节点只做一件事;状态在节点间传递。LangGraph 很适合表达这种图。


一、状态里放什么

class AgentState(TypedDict):
    raw_input: str
    alarm_code: NotRequired[str]
    category: NotRequired[str]
    meaning: NotRequired[str | None]
    level: NotRequired[int | None]
    messages: Annotated[list[str], operator.add]  # 轨迹自动拼接

messages 用 Annotated[..., operator.add]:各节点返回的列表会追加,方便打印步骤。


二、三个节点

节点 职责
ingest 从原文抽 E001 这类码
classify 按前缀规则分类(如 E0→单体级)
lookup 调用 lookup_fault_code

分类先用规则,不上 LLM——快、稳、好测。

E0xx → cell(单体)
E1xx → pack
E2xx → insulation
E3xx → comms
其它 → unknown

三、构图

g = StateGraph(AgentState)
g.add_node("ingest", ingest_node)
g.add_node("classify", classify_node)
g.add_node("lookup", lookup_node)
g.add_edge(START, "ingest")
g.add_edge("ingest", "classify")
g.add_edge("classify", "lookup")
g.add_edge("lookup", END)
app = g.compile()

final = app.invoke({"raw_input": "E001", "messages": []})

验收:

python -m energy_agent diagnose E001
# 期望看到 classify: E001 → category=cell

四、为什么值得用图而不是一个 for 循环

  1. 后面要插入 rag、条件分支(未知码→人工)时,改边比改意大利面函数清晰;
  2. 每步可打点、可 trace;
  3. 讲解时画得清:ingest/classify/lookup/rag。

简单线性流程用图也不丢人——先跑通,再加条件边。


五、带走这三条

  1. 一节点一职责;状态字段显式。
  2. 分类可先规则,工具可先手写调用。
  3. 轨迹 messages 为后续 API/录屏留证据。

下一篇:把月 2 的 RAG 封装成 Tool,接到图里。

这是专栏第 21 篇。两到三天一更,RAG Tool 见。

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

友情链接更多精彩内容