「从零到 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 循环
- 后面要插入
rag、条件分支(未知码→人工)时,改边比改意大利面函数清晰; - 每步可打点、可 trace;
- 讲解时画得清:ingest/classify/lookup/rag。
简单线性流程用图也不丢人——先跑通,再加条件边。
五、带走这三条
- 一节点一职责;状态字段显式。
- 分类可先规则,工具可先手写调用。
- 轨迹 messages 为后续 API/录屏留证据。
下一篇:把月 2 的 RAG 封装成 Tool,接到图里。
这是专栏第 21 篇。两到三天一更,RAG Tool 见。