首页 / 文章 / 实用提示:使用LangGraph构建LEPA支持代理

实用提示:使用LangGraph构建LEPA支持代理

《实用笔记》操作指南:使用 LangGraph 构建 LEPA 支持代理——面向采用该模式的团队提供的合同、校验规则及即插即用代码模块。

2082 词

可将此内容视为《利用LangGraph构建LEPA支持代理》一文中理念面向操作员的重构版本:清晰的阶段划分、有序的代码模块,以及能在交接过程中保留的恢复说明。 将“概览”阶段视为可度量的界面最为有效。在扩大范围之前,先记录一份最佳示例、一个故障案例以及回滚说明。 相较于庞大的脚本,应优先选择小型且可测试的单元。当某一步骤出现故障时,故障原因应能指向单一责任模块,而非复杂的流程链。

您希望第一个图表实现的功能

对于所需阶段,应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名,定义成功检测标准,并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作,必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。

User message
     ↓
Notice who is speaking (teacher / admin / unknown)
     ↓
Classify the topic (grades, login, …)
     ↓
Too vague? Ask a clarifying question
     ↓
Otherwise continue toward docs + an answer

项目设置(刻意保持简单)

由于项目设置刻意保持分阶段进行,因此在修改代码之前需明确输入参数、各步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。除了功能结果外,还需记录执行时间以及令牌或查询成本。提前了解这些成本可以避免在从演示环境过渡到共享环境时出现意外费用。对于那些会耗费资金或修改生产数据的操作,必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。

PRJ-02/
├── app/
│   ├── state.py      # SupportState
│   ├── graph.py      # StateGraph wiring
│   ├── agents/       # intake, knowledge, support
│   ├── nodes/        # classify, ask_clarification
│   └── tools/        # search_knowledge (next article)
├── knowledge/        # LEPA support Markdown
├── api/              # FastAPI (later article)
└── tests/

状态:在图中传递的对象

在修改代码之前,需为状态对象定义阶段、输入参数、步骤负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放,以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的步骤,需设置人工审批环节。编译时的连接方式并不等同于业务功能的完整性。 在修改代码之前,需为状态对象定义阶段、输入参数、步骤负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤,而无需猜测隐藏状态。 相较于庞大的脚本,应优先使用小型且可测试的单元。当某个步骤失败时,故障原因应能明确指向单一责任模块,而非复杂的流程结构。

e.

messages                 # conversation turns (add_messages reducer)
user_role                # teacher / admin / unknown
issue_category           # grades, authentication, …
clarification_needed     # should we ask for more detail?
clarification_question   # what we ask
retrieved_documents      # doc snippets (later step)
final_answer             # what we return to the user
conversation_summary     # reserved for later — unused in v1
messages: Annotated[list, add_messages]
{"user_role": "teacher"}
{"issue_category": "grades", "clarification_needed": False}

节点:每个阶段处理一项任务

在处理“每个阶段一个任务”的节点时,首先写下相关契约:所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名,定义成功检查标准,并杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时,恢复流程不应再次调用相同的大语言模型。

(state) → partial update

数据输入

在处理输入阶段时,首先写下相关契约:所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况,可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时,恢复流程不应再次收取相同的LLM调用费用。

分类

在处理“分类”阶段时,首先需明确相关规范:所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置,以便操作员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时,恢复流程不应再次调用相同的大型语言模型。 在处理“分类”阶段时,首先需明确相关规范:所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型且可测试的单元,而非结构复杂的脚本。当某个步骤失败时,故障应能指向具体的责任模块,而非整个混乱的流程。

请求进一步说明

将“需求澄清”阶段视为可度量的工作面,效果最佳。在扩大范围之前,需记录一份完美的实现文档、一个失败案例以及回滚说明。 把这个阶段视为输入与经过验证的输出之间的契约。为相关成果命名,明确成功标准,绝不允许默许不完整的处理结果。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段,还会在中断后导致无法继续处理。

知识与支持

将知识支持阶段视为可度量的对象时,其效果最佳。在扩大范围之前,先记录一份优秀的案例、一个故障实例以及回滚说明。在功能结果旁还需记录执行时间以及令牌或查询成本。提前了解这些成本,就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图结构的层次简单且类型明确,嵌套的数据块会掩盖是哪个节点修改了哪个字段,还会在出现中断后导致流程无法继续。

边:控制流的传递方式

将 Edge 控制流程视为可测量的表面时,其工作效果最佳。在扩大范围之前,先记录一个成功的案例、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一处,这样操作人员无需查看整个结构即可进行审计。 保持图结构的层次简单且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息,还会在中断后导致无法继续执行。

普通边

将“普通边”阶段视为可测量的表面来处理时,其效果最佳。在扩大范围之前,先记录一个成功的用例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品本身的一部分,而非后续的优化工作。要保持图结构的层次简单且类型明确,嵌套的数据块会掩盖哪个节点编写了哪个字段的信息,并且在中断后会导致流程无法继续。

graph.add_edge(START, "intake")
graph.add_edge("intake", "classify")
graph.add_edge("knowledge", "support")
graph.add_edge("support", END)
When this node finishes, always go there next.

条件边

将条件边阶段视为可测量的表面来处理时,其效果最佳。在扩大范围之前,先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元,而非庞大的脚本。当某一步骤失败时,故障应指向单一的责任主体,而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息,还会在中断后导致无法继续执行。

def route_after_classify(state: SupportState) -> Literal["clarify", "continue"]:
    if state.get("clarification_needed"):
        return "clarify"
    return "continue"
graph.add_conditional_edges(
    "classify",
    route_after_classify,
    {
        "clarify": "ask_clarification",
        "continue": "knowledge",
    },
)
"help"  → clarify → ask_clarification → END
"How do I enter grades?" → continue → knowledge → support → END

开始与结束

将START和END阶段视为可测量的界面时,其效果最佳。在扩大范围之前,先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名,明确成功标准,绝不允许默默地仅完成部分工作。 保持图结构的层次简单且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段,还会在中断后导致无法继续处理。

START = where the runtime begins
END   = where this run stops
START → intake → classify → …
…
ask_clarification → END
support → END

完整的拓扑结构(构建后的状态)

将整个拓扑结构视为可测量的表面来处理时,其在各阶段的表现最为理想。在扩大范围之前,先记录一份最佳案例、一个故障场景以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况,可避免在系统从演示环境过渡到共享环境时出现意外费用。需保持图结构的层次简单且类型明确,否则嵌套的数据块会掩盖具体是哪个节点修改了哪一字段,且在中断后还会导致流程无法继续。

START
  → intake
  → classify
  → conditional
        ├─ clarify  → ask_clarification → END
        └─ continue → knowledge → support → END

请求在图结构中的处理流程

将请求的处理流程视为可度量的对象,才能最有效地理解其运作方式。在扩大范围之前,先记录一份成功的处理案例、一个失败案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能开关应集中存放于一处,这样操作人员无需查看整个系统结构即可进行审计。 需保持系统状态的简洁性与类型化。嵌套的数据结构会掩盖是哪个节点修改了哪个字段,还会导致在进程中断后无法继续执行。

明确的问题

将“明确问题阶段”视为可度量的对象来处理时,其效果最佳。在扩大范围之前,先记录一个成功的用例、一个失败案例以及回滚说明。同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分,而非后续需要补充的内容。保持图表状态简洁且类型明确,嵌套的数据块会掩盖哪个节点修改了哪个字段的信息,还会在中断后导致流程无法继续。

User: "Why aren't grades showing?"
        ↓
intake        → user_role = unknown (unless they said teacher/admin)
        ↓
classify      → issue_category = grades, clarification_needed = False
        ↓
route         → "continue"
        ↓
knowledge     → search docs (next article)
        ↓
support       → final_answer
        ↓
END

模糊的问题

User: "help"
        ↓
intake
        ↓
classify      → unknown + clarification_needed = True
        ↓
route         → "clarify"
        ↓
ask_clarification → asks which LEPA area
        ↓
END

compile()与invoke():运行时时刻

return graph.compile(checkpointer=checkpointer)
from langchain_core.messages import HumanMessage
from app.graph import app_graph
result = app_graph.invoke(
    {"messages": [HumanMessage(content="How do I enter grades?")]}
)
print(result["final_answer"])

你刻意推迟的事项

要点总结

链接

操作检查清单