开发实践 01｜使用 LingxiGraph 实现 Stage 1 文本处理流水线

[lyyzka]

任务编号

W01

任务目标

使用 LingxiGraph 独立实现一个可运行、可观察、可调试的 StateGraph 文本处理流水线，完成从 State 定义、Node 编写、Edge 连接、Graph 编译，到 invoke 执行、Streaming 事件观察和 Studio 调试的完整开发流程。

完成任务后，应能够理解并实际验证：

State → Node → Edge → Compile → Invoke / Stream

重点掌握 State 如何在不同节点之间传递，以及 LingxiGraph Runtime 如何执行和观察整个状态图。

背景

培训中已经介绍了 LangGraph 的基本思想，以及基于状态图思想实现的 LingxiGraph，包括 State、Node、Edge、Runtime 等核心概念。

本次任务从“理解概念”进入“实际开发”阶段。

你需要使用 LingxiGraph 实现参考项目 lingxigraph-dev-task-demo 中 Stage 1 类似的效果，通过一个不依赖大模型的确定性 Python 流水线，实际理解状态如何被不同节点读取、修改并继续向后传递。

本任务不要求调用 LLM，重点不是 AI 能力，而是掌握 StateGraph 的基本开发与运行方式。

主要完成方式

实验 / Demo

是否需要单独提交成果

需要

任务要求

使用 LingxiGraph 独立完成一个 Stage 1 文本处理状态图。

一、状态图结构

至少实现以下 Graph：

START
↓
normalize_request
↓
structure_task
↓
finalize
↓
END

必须真正使用 LingxiGraph 的 StateGraph、START、END、add_node()、add_edge() 和 compile() 构建状态图。

不能只编写三个普通 Python 函数并手动依次调用。

二、State

自行使用 TypedDict 等方式定义 Graph State。

State 至少需要能够保存：

• 原始用户请求
• 规范化后的请求
• 任务分析结果
• 建议处理步骤
• 最终结果

字段名称可以自行设计。

三、normalize_request

实现输入规范化节点。

例如：

输入：
" 请帮我 分析这个 Python 接口为什么返回 500 "

处理后：
"请帮我 分析这个 Python 接口为什么返回 500"

要求保留原始请求，同时去除首尾空白、合并连续空白，并将规范化结果写入 State。

四、structure_task

读取规范化后的请求，通过确定性 Python 规则识别任务类型，并生成对应的处理步骤。

至少支持：

1. 故障排查
   例如：
   “请分析这个 Python 接口为什么返回 500”

2. 功能开发
   例如：
   “请为项目添加一个健康检查接口”

3. 概念解释
   例如：
   “请解释 Python 装饰器的执行原理”

可以使用字符串匹配、关键词和 if / elif 实现，不要求使用 NLP 或大模型。

例如故障排查任务可以生成：

1. 检查异常日志
2. 定位接口入口
3. 检查依赖调用
4. 构造最小复现

五、finalize

读取前面节点写入 State 的内容，将规范化请求、任务分析和处理步骤整理为统一的最终输出，并将结果重新写入 State。

六、invoke

Graph 编译完成后，必须能够通过 graph.invoke() 正常运行。

至少分别测试：

• 故障排查
• 功能开发
• 概念解释

三类输入。

七、Streaming

实现并验证 LingxiGraph Streaming。

首先使用：

stream_mode="updates"

观察三个节点依次产生的 State 更新。

然后在节点中尝试使用：

runtime.stream_writer(...)

并通过：

stream_mode="custom"

产生三个阶段对应的自定义运行事件，例如：

• 用户输入已完成规范化
• 任务结构化分析已完成
• 流水线执行完成

事件具体文字可以自行设计。

八、Studio

使用：

lingxigraph dev

或：

uv run lingxigraph dev

启动本地开发环境。

在 LingxiGraph Studio 中能够看到：

START
→ normalize_request
→ structure_task
→ finalize
→ END

并至少在 Studio 中执行一次 Graph。

九、代码提交

将完整项目提交到自己的 GitHub 仓库。

README 至少说明：

项目用途、Graph 结构、安装方法、运行方法、示例输入输出、Streaming 使用方式，并附一张 Studio 运行截图。

建议先独立实现，遇到问题后再查看参考仓库的 Stage 1 源码，不建议直接复制参考代码。

完成标准

满足以下条件即可认为核心任务完成：

• 项目使用 LingxiGraph StateGraph，而不是普通 Python 顺序调用。
• 自行定义了 Graph State。
• Graph 至少包含 normalize_request、structure_task、finalize 三个节点。
• 节点通过 Edge 正确连接为 START → 3 个节点 → END。
• Graph 可以成功 compile()。
• graph.invoke() 可以正常运行。
• normalize_request 能正确处理多余空白，并保留原始输入。
• structure_task 至少能够识别“故障排查、功能开发、概念解释”三类任务。
• 不同任务能够产生对应的处理步骤。
• finalize 能生成完整的最终输出。
• 至少完成三类不同输入的运行验证。
• stream_mode="updates" 可以观察三个节点的 State 更新。
• stream_mode="custom" 可以观察三个节点产生的自定义事件。
• lingxigraph dev 可以正常启动。
• LingxiGraph Studio 中可以看到完整 Graph，并能够实际执行。
• 成果已上传到 GitHub。
• README 包含运行说明和 Studio 截图。

提交时请在 Build-Lab/Submissions
新建discussion，根据模版要求提交你的实践内容

关联 Issue

No response

参考资料

LingxiGraph：
https://github.com/LingXi-Org/LingxiGraph

Stage 1 效果参考：
https://github.com/lyyzka/lingxigraph-dev-task-demo

重点参考参考仓库 README 中的 Stage 1 效果。

建议开发顺序：

State
→ START → normalize_request → END
→ 确认 invoke 可以运行
→ 添加 structure_task
→ 添加 finalize
→ 添加 updates Streaming
→ 添加 custom Streaming
→ 最后使用 Studio 调试

不要一开始同时实现全部功能。先保证最简单的 START → Node → END 能运行，再逐步增加节点和 Runtime 能力。

建议截止时间

2026-08-30

进阶挑战

完成核心任务后，可以继续探索以下任意内容：

1. 增加更多任务类型。
2. 使用 Conditional Edge，让不同类型任务进入不同节点。
3. 为 Graph 编写 pytest 自动化测试。
4. 尝试 Checkpoint 与状态恢复。
5. 尝试 Retry。
6. 尝试 Human-in-the-loop。
7. 尝试接入 LLM，将确定性规则节点替换为 Agent 节点。
8. 继续实现参考项目 Stage 2。

进阶挑战不影响 W01 核心任务的完成判定。

[github-actions[bot]]

@LingXi-Org/builders 新任务已发布：实验 / Demo。完成后请到 任务提交 提交成果。 关联 Issue：No response。