Add new workflow configuration for accident scene documentation

- Introduced a new JSON file containing detailed node configurations for the accident scene documentation workflow. - Included nodes for user guidance, workflow start, and AI chat integration, with comprehensive input and output specifications. - Enhanced the structure to support various input types and configurations for improved user interaction and data handling.
Start LangGraph backend migration foundation
2026-07-27 17:24:23 +08:00 · 2026-07-27 17:21:29 +08:00 · 2026-07-26 07:25:24 +08:00 · 2026-07-26 07:24:30 +08:00
40 changed files with 9131 additions and 208 deletions
--- a/.env
+++ b/.env
@@ -1,9 +0,0 @@
-DATABASE_URL=sqlite:///./test.db
-SECRET_KEY=your_secret_key
-DEBUG=True
-
-ANALYSIS_SERVICE_URL=http://127.0.0.1:3030
-ANALYSIS_AUTH_TOKEN=fastgpt-r13smJwPgXfGj1HDfc4SWAvIoNrL5Wc6o0BYnezqBs7hgzPdQ7Q34hVl2FJc0R
-APP_ID=6a310def7132e9f7d592dabb
-
-VOICE_CONFIG=config/voice-fastgpt-state-xfyunSuperTTS.json
--- a/.env.example
+++ b/.env.example
@@ -0,0 +1,19 @@
+DATABASE_URL=sqlite:///./test.db
+SECRET_KEY=replace-with-a-random-secret
+DEBUG=false
+
+ZNJJ_ENVIRONMENT=development
+AGENT_BACKEND=fastgpt
+LANGGRAPH_CHECKPOINTER=memory
+
+ANALYSIS_SERVICE_URL=http://127.0.0.1:3000
+ANALYSIS_AUTH_TOKEN=replace-with-local-fastgpt-token
+APP_ID=replace-with-fastgpt-app-id
+
+LLM_API_KEY=replace-with-llm-api-key
+LLM_BASE_URL=https://api.openai.com/v1
+LLM_MODEL=replace-with-model-name
+LLM_TIMEOUT_SECONDS=60
+LLM_MAX_RETRIES=2
+
+VOICE_CONFIG=config/voice-fastgpt-state-xfyunSuperTTS.json
--- a/.gitignore
+++ b/.gitignore
@@ -3,3 +3,5 @@ __pycache__/
 logs/
 *.log
 .env
+.env.*
+!.env.example
--- a/docs/baselines/fastgpt-baseline-20260726.md
+++ b/docs/baselines/fastgpt-baseline-20260726.md
@@ -0,0 +1,101 @@
+# FastGPT 迁移前基线
+
+> 采集日期：2026-07-26
+> Git 基线：`5c719ed`
+> 环境：本地 macOS，项目 `.venv`，FakeBackend/FakeClient；未调用真实 FastGPT
+> 用途：冻结可复现工程基线，不冒充生产业务指标
+
+## 自动化基线
+
+执行：
+
+```bash
+.venv/bin/python -m pytest -q
+```
+
+Phase 0 开始前结果：
+
+```text
+32 passed in 0.32s
+```
+
+覆盖范围：
+
+- Pydantic 公共请求/响应 schema；
+- `/chat` 对 backend-neutral contract 的适配；
+- FastGPT backend 的 SDK/Event 转换；
+- state prefix 跨 chunk 的基础兼容；
+- 文本分句。
+
+未覆盖范围：
+
+- 真实 FastGPT 网络延迟和错误；
+- 真实 workflow 状态序列稳定性；
+- `/set_info`、`/get_info` 完整兼容行为；
+- 生产转人工比例和 prefix 失败率；
+- 数据库、并发恢复和 LangGraph。
+
+Phase 0 完成后的测试数量和耗时见本报告底部。
+
+## 本地 API 适配层耗时
+
+使用内存 FakeBackend，关闭日志后直接调用 endpoint 并完整消费流式响应。该数据只衡量 Python 适配、prefix/SSE 处理开销，不包含 HTTP、网络、FastGPT 或模型延迟。
+
+```text
+nonstream n=2000 p50=0.004ms p95=0.014ms p99=0.030ms
+stream-consume n=1000 p50=0.011ms p95=0.034ms p99=0.051ms
+```
+
+这组数据用于后续发现 API 适配层的明显性能回退，不能与生产端到端延迟混用。
+
+## 生产指标采集口径
+
+以下数据无法从仓库推导，必须由部署环境日志或监控采集。负责人应使用同一时间窗口、同一调用方集合，并排除压测流量。
+
+| 指标 | 计算方式 | 当前值 |
+|---|---|---|
+| `/chat` 请求数 | 成功与失败总请求 | 待生产采集 |
+| 非流式 P50/P95/P99 | endpoint 总耗时 | 待生产采集 |
+| 流式 TTFB P50/P95/P99 | 收到请求至首个 `text_delta` | 待生产采集 |
+| 响应体错误率 | `code != "200"` / 请求数 | 待生产采集 |
+| FastGPT 超时率 | timeout / FastGPT 调用数 | 待生产采集 |
+| Prefix 失败率 | 缺失、格式错误、未知码 / 模型回复数 | 当前未结构化记录 |
+| 转人工率 | `0001/0002/0003/0004/0005` / session 数 | 待生产采集 |
+| `formUpdate` 产生率 | 非空 patch / `needFormUpdate=true` 轮次 | 待生产采集 |
+| 平均轮次 | chat 轮次 / 完结 session 数 | 待生产采集 |
+| `/set_info`、`/get_info` 错误率 | `code != "200"` / 请求数 | 待生产采集 |
+
+## 当前可观测性限制
+
+当前日志虽然记录延迟，但也记录完整 `sessionId`、输入、输出和 `formUpdate`，不能直接作为长期生产基线方案。Phase 1/8 应先加入 request ID、session hash、结构化事件和脱敏，然后再持续采集。
+
+建议临时聚合时只输出：
+
+- 时间桶；
+- endpoint；
+- 成功/稳定错误类别；
+- 耗时；
+- stage code；
+- 是否产生 form patch；
+- 不可逆 session hash。
+
+不得导出原始对话、身份证、手机号、车牌、Token 或完整表单。
+
+## Phase 0 最终验证
+
+完成日期：2026-07-26
+
+```text
+48 passed in 0.49s
+```
+
+相对 Phase 0 开始前新增 16 个测试，覆盖：
+
+- `/set_info`、`/get_info` 的 FastGPT 辅助调用和兼容序列化；
+- SSE 成功事件顺序、唯一终止事件和文本拼接；
+- 当前缺失流式 prefix 的历史行为；
+- 状态/迁移/字段注册表闭合性；
+- 黄金场景唯一性和敏感号码扫描；
+- 源码、文档、测试和配置样例中的 FastGPT Token 扫描。
+
+生产业务指标仍标记为“待生产采集”。这是外部可观测数据依赖，不用估算值替代；最迟必须在 Phase 9 shadow 前完成采集。
--- a/docs/domain/README.md
+++ b/docs/domain/README.md
@@ -0,0 +1,40 @@
+# Phase 0 领域契约索引
+
+> 基线版本：2026-07-26
+> 适用范围：`/chat`、`/set_info`、`/get_info`
+> 规则来源优先级：V1.0.9 接口文档 > 2026-07-26 workflow/Prompt > 当前服务代码 > 历史 workflow/Prompt
+
+本目录冻结 LangGraph 迁移前的外部契约和业务规则。JSON 文件是后续代码生成、参数化测试和 Graph validator 的机器可读输入；Markdown 文件解释兼容行为及来源。
+
+## 交付物
+
+| 文件 | 用途 |
+|---|---|
+| `api-contract.md` | 三个接口的当前兼容行为和已知偏差 |
+| `stage-codes.json` | 状态码、内部/外部映射和产生方式 |
+| `stage-transitions.json` | 权威迁移矩阵和拍照顺序 |
+| `field-registry.json` | 字段、分组、类型、敏感性和写权限 |
+| `business-rules.md` | 从 workflow、Prompt 和 endpoint 提取的规则 |
+| `event-mapping.md` | 历史魔法字符串到结构化事件的映射 |
+| `../baselines/fastgpt-baseline-20260726.md` | 迁移前可复现测试/性能基线 |
+| `../../test/fixtures/golden/accident-scenarios.json` | 脱敏黄金场景 |
+
+## 已冻结的关键决定
+
+1. 对外继续使用 camelCase、字符串业务码和 JSON 编码的 `/get_info.value`。
+2. `3001`、`3002` 为内部信息确认状态，对外仍返回 `1002`。
+3. `2006`、`2017`、`2020` 仅作为历史别名接收，不作为新图的合法目标状态。
+4. `0004` 纳入正式状态表；当前 endpoint 状态名称映射缺失是待修复偏差，不代表删除该状态。
+5. 拍照完成、连续拍照失败、无回复和外部字段更新在新图中必须是结构化确定性事件。
+6. `sfzmwh1/2`、`sjwh1/2` 是只读兼容字段，不允许 `/set_info` 写入。
+7. `phase`、`stage_code`、计数器、版本和图路由字段禁止外部修改。
+
+## 变更规则
+
+Phase 0 冻结后，修改这里的状态、字段或外部契约必须同时：
+
+1. 说明业务原因和兼容影响；
+2. 更新机器可读 JSON；
+3. 更新黄金场景；
+4. 更新对应契约/领域测试；
+5. 获得接口调用方或产品确认。
--- a/docs/domain/api-contract.md
+++ b/docs/domain/api-contract.md
@@ -0,0 +1,115 @@
+# 兼容 API 契约基线
+
+## 通用约定
+
+- 路径保持 `/chat`、`/set_info`、`/get_info`。
+- 请求和响应字段保持 camelCase。
+- 业务成功/失败主要通过响应体字符串 `code` 表达；现有 endpoint 通常仍返回 HTTP 200。
+- `sessionId` 最大 64 字符，`timeStamp` 最大 32 字符。
+- Pydantic 校验失败由 FastAPI 返回 HTTP 422。
+- 本文冻结的是当前可观察行为；“目标行为”标记为后续迁移必须修复的已批准偏差。
+
+## `POST /chat`
+
+### 请求
+
+| 字段 | 类型 | 必填 | 默认值 |
+|---|---|---:|---|
+| `sessionId` | string | 是 | - |
+| `timeStamp` | string | 是 | - |
+| `text` | string | 是 | - |
+| `needFormUpdate` | boolean | 否 | `false` |
+| `useTextChunk` | boolean | 否 | `false` |
+
+### 非流式响应
+
+字段为 `sessionId`、`timeStamp`、`outputText`、`formUpdate`、`nextStage`、`nextStageCode`、`code`、`msg`。
+
+- 成功时 `code="200"`。
+- `<state>XXXX</state>` 从正文中移除。
+- `3001/3002/1002 → 1002`，`2006 → 2004`，`2017 → 2016`，`2020 → 0002`。
+- Prefix 缺失或正文不可解析时当前返回 `code="500"` 和“消息不完整”。
+- FastGPT 认证、限流和 API 异常分别映射为响应体 `401`、`429`、`500`。
+- `formUpdate` 保持无固定 schema 的 JSON 值，以兼容现有调用方。
+
+### SSE 响应
+
+事件名和数据：
+
+| 事件 | 数据 | 基数 |
+|---|---|---|
+| `stage_code` | `{"nextStageCode":"1002","nextStage":"通话中"}` | 成功轮最多一次 |
+| `formUpdate` | 表单 patch 对象 | 有更新时最多一次 |
+| `text_delta` | `{"text":"..."}` | 零到多次 |
+| `done` | `{"status":"completed"}` | 成功恰好一次 |
+| `error` | `{"msg":"...","code":"500"}` | 失败恰好一次且终止 |
+
+V1.0.9 文档要求 `stage_code` 先于 `text_delta`；`formUpdate` 的位置由 FastGPT `flowResponses` 到达时间决定，文档示例允许它出现在两个 `text_delta` 之间。LangGraph 迁移目标固定为：
+
+```text
+stage_code -> formUpdate(可选) -> text_delta* -> done
+```
+
+迁移后的错误路径不得同时产生 `done` 和 `error`。`useTextChunk=true` 只改变 `text_delta` 切分，不改变拼接后的文本。
+
+### 已知偏差
+
+- 当前流式 parser 会在整段文本中搜索标签，而不是强制标签位于开头。
+- 当前流式未知状态码仍可能发送空 `nextStage`。
+- 当前流式缺少 prefix 时仍可能输出正文和 `done`。
+- 当前流式内部事件处理异常会记录后继续，可能掩盖部分失败。
+- 当前实现记录完整输入、输出和 `formUpdate`，不符合数据保护目标。
+
+以上偏差被字符化测试记录，但不作为 LangGraph 新实现的目标行为；Phase 5–8 必须按迁移计划修正。
+
+## `POST /set_info`
+
+### 请求
+
+字段为 `sessionId`、`timeStamp`、`key`、`value`、`includeInputInfo`；其中 `includeInputInfo` 默认 `false`。
+
+当前实现：
+
+1. 通过一次 FastGPT 对话读取 `newVariables.state`；
+2. 删除辅助对话记录；
+3. 直接执行 `state[key] = value`；
+4. 再通过 FastGPT 对话写回并删除辅助记录。
+
+成功返回 `code="200"`；任一步失败返回响应体 `code="500"`。
+
+目标行为：
+
+- 仅接受 `field-registry.json` 中 `external_write=true` 的 key。
+- 进行类型转换和领域校验。
+- 禁止写内部状态字段。
+- 直接事务化写业务状态，不调用 LLM，不创建/删除辅助聊天记录。
+
+## `POST /get_info`
+
+请求字段为 `sessionId`、`timeStamp`、`key`、`includeInputInfo`；`includeInputInfo` 默认 `false`。
+
+支持：
+
+- `all`
+- `acdinfo`
+- `acdhuman1`
+- `acdhuman2`
+- 单个字段 key
+
+兼容序列化：
+
+- `value` 始终是 JSON 编码后的字符串。
+- boolean 转为字符串 `"1"` 或 `"0"`。
+- 缺失字段转为空字符串。
+- 未知单字段 key 当前返回 JSON 字符串 `""`，而不是报错。
+
+目标实现仍保留上述响应编码，但直接读取业务投影，不调用 FastGPT。
+
+## 契约来源
+
+- `src/schemas/models.py`
+- `src/api/endpoints.py`
+- `docs/视频快处智能信息采集机器人交互接口文档V1.0.9.docx`
+- `docs/chat-stream-mode.md`
+- `test/api/test_public_schema_contract.py`
+- `test/api/test_chat_backend_boundary.py`
--- a/docs/domain/business-rules.md
+++ b/docs/domain/business-rules.md
@@ -0,0 +1,116 @@
+# 事故采集业务规则基线
+
+## 规则来源
+
+本基线交叉比对以下来源：
+
+1. V1.0.9 交互接口文档；
+2. `workflow/20260726/事故信息采集20260726.json`；
+3. `prompts/20260723/单车拍照.txt`；
+4. `prompts/20260723/双车拍照.txt`；
+5. 当前 `src/api/endpoints.py`；
+6. 2025 版本 workflow/Prompt，仅用于识别历史兼容行为。
+
+发生冲突时采用接口文档和 2026-07-26 规则；历史状态别名只在 API 边界兼容。
+
+## 全局规则
+
+1. 每个自然语言回复必须以且仅以一个 `<state>四位数字</state>` 开头。
+2. 用户明确要求“转人工”“找人工”“人工客服”等时，立即进入 `0001`。
+3. 明确或高度可信的人伤、三辆及以上机动车、涉及行人/非机动车等复杂情况进入 `0003`。
+4. 明确否定人伤时不得因句中出现“受伤”“流血”等词误触发 `0003`。
+5. ASR 内容破碎或人伤语义矛盾时，用当前采集状态封闭确认，不能直接冒险放行。
+6. 当前问题没有有效答案时不得跳题。第一次澄清，第二次强制选择；仍失败进入 `0002`。
+7. 连续第一次无回复使用固定唤醒话术；连续第二次进入 `0004`。
+8. 用户有效回复后无回复计数清零。
+9. 状态候选必须经过枚举和迁移矩阵校验才能持久化。
+
+## 准备与事故信息采集
+
+1. 新 session 初始为 `1001`，提示撤离到安全区域、开启双闪、放置警告牌。
+2. `【开始】` 或 `【继续办理】` 后进入 `1002`。
+3. 采集顺序：
+   - 事故经过；
+   - 是否有人伤；
+   - 是否涉及非机动车/摩托车/自行车；
+   - 事故时间并校验不能晚于当前时间；
+   - 是否仍在现场；
+   - 机动车数量。
+4. 用户提前提供的字段用于填槽，但进入下一项前应做封闭式确认。
+5. 一辆机动车、无人伤且不涉及非机动车/行人：进入 `2000`。
+6. 两辆机动车、无人伤且不涉及非机动车/行人：进入 `2010`。
+7. 三辆及以上，或涉及非机动车/行人，或有人伤：进入 `0003`。
+
+## 单车拍照
+
+严格顺序：
+
+```text
+2000 车前/车牌
+ -> 2001 车辆碰撞部位
+ -> 2002 被撞物品
+ -> 2003 本人正面
+ -> 2004 确认或纠正车牌
+ -> 2005 确认车损位置
+ -> 3001 单车信息确认
+```
+
+- `2000`–`2003` 只有 `PhotoCompletedEvent` 可以正常推进；其他普通输入重复当前固定指令。
+- `2004` 肯定车牌或提供完整新车牌后进入 `2005`；仅否定但不提供号码时停留并追问。
+- `2005` 获得有效车损位置后进入 `3001`；连续两次无效回答进入 `0002`。
+- 任意单车照片状态收到拍照失败事件立即进入 `0005`。
+
+## 双车拍照
+
+严格顺序：
+
+```text
+2010 第一辆车侧前方/车牌
+ -> 2011 第一辆车碰撞部位
+ -> 2012 第二辆车碰撞部位
+ -> 2013 第二辆车侧后方/车牌
+ -> 2014 另一方驾驶人正面
+ -> 2015 本人正面
+ -> 2016 确认或纠正车牌
+ -> 3002 双车信息确认
+```
+
+- `2010`–`2015` 只有 `PhotoCompletedEvent` 可以正常推进。
+- `2016` 肯定或提供完整新车牌后进入 `3002`；无关或不完整回答停留，连续两次失败进入 `0002`。
+- 任意双车照片状态收到拍照失败事件立即进入 `0005`。
+
+## 当事人信息确认
+
+### 单车 `3001`
+
+依次确认：
+
+1. 是否为对应车辆车主/驾驶人；
+2. 姓名；
+3. 身份证后四位；不一致时采集完整号码并二次确认；
+4. 手机号后四位；不一致时采集完整号码并二次确认；
+5. 完成后进入 `0000`。
+
+### 双车 `3002`
+
+先完成第一位驾驶人上述信息，再要求将电话交给第二位驾驶人，重复相同步骤。第二位完成后进入 `0000`。
+
+身份证和手机号允许分段输入；中间态只保存已接收片段，不应把未完成号码写入已确认业务字段。日志、trace 和黄金数据不得包含真实号码。
+
+## 表单更新
+
+- `needFormUpdate=false` 时无需返回 `formUpdate`。
+- `needFormUpdate=true` 时只返回本轮相对当前表单发生变化的字段。
+- LLM 提取结果必须经过 `field-registry.json` 白名单和类型校验。
+- patch 之外的原字段保持不变。
+- 不允许 LLM 更新 phase、状态码、计数器或版本号。
+
+## 当前实现与目标规则的差异
+
+- FastGPT Prompt 承担了多数计数和迁移逻辑，服务端未校验迁移合法性。
+- 当前 prefix 正则不是开头锚定且接受任意位数字。
+- 当前 `/set_info` 可写任意 key。
+- 当前 `/get_info` 和 `/set_info` 通过辅助 LLM 对话访问状态。
+- 当前日志会记录完整用户输入、回复和表单。
+
+这些差异是 Phase 1–8 的明确改造项，不能被解释为本基线认可的目标行为。
--- a/docs/domain/event-mapping.md
+++ b/docs/domain/event-mapping.md
@@ -0,0 +1,37 @@
+# 历史输入到结构化事件的映射
+
+API 兼容层可以继续接收历史字符串，但进入 LangGraph 前必须转换为以下事件。图节点不得再通过自然语言字符串判断系统事件。
+
+| 历史输入 | 结构化事件 | 必要字段 | 确定性效果 |
+|---|---|---|---|
+| 普通用户文本 | `UserMessageEvent` | `event_id`, `text`, `need_form_update`, `use_text_chunk` | 进入当前阶段处理；自然语言轮次通常调用一次 LLM |
+| `【开始】` | `SessionStartedEvent` | `event_id` | `1001 → 1002`，开始事故描述采集 |
+| `【继续办理】` | `SessionStartedEvent` | `event_id` | 与 `【开始】` 相同 |
+| `【拍摄完成】` | `PhotoCompletedEvent` | `event_id`, `photo_step` | 仅按单车/双车严格顺序推进一步，0 次 LLM |
+| `【客户端连续3次拍摄识别失败：原因】` | `PhotoRecognitionFailedEvent` | `event_id`, `reason` | 任意照片阶段立即进入 `0005`，0 次 LLM |
+| `【用户无回复】` | `NoResponseEvent` | `event_id` | 第一次重复唤醒；连续第二次进入 `0004` |
+| `【用户未回复】` | `NoResponseEvent` | `event_id` | 历史别名，效果同上 |
+| `/set_info` 请求 | `SetInfoEvent` | `event_id`, `key`, `value` | 校验白名单/类型后幂等更新，不调用 LLM |
+
+## 优先级
+
+同一轮只允许一个输入事件。事件处理优先级为：
+
+```text
+显式人工请求
+  > 连续拍照失败
+  > 明确人伤/复杂情况
+  > 无回复
+  > 拍照完成
+  > 外部字段更新
+  > 普通用户文本
+```
+
+显式人工请求来自普通文本时允许确定性关键词 gate 先处理；语义模糊的人伤内容交由同一轮阶段 LLM 判断，但候选状态仍需迁移校验。
+
+## 计数重置
+
+- 收到有效用户回答后，`no_response_count` 清零。
+- 当前问题得到有效答案后，对应 `clarification_counts[question_id]` 清零。
+- 非连续的无回复不能累计到 `0004`。
+- 客户端已经负责累计三次拍照识别失败；服务端收到一次结构化失败事件就进入 `0005`，不得再次累计三次。
--- a/docs/domain/field-registry.json
+++ b/docs/domain/field-registry.json
@@ -0,0 +1,51 @@
+{
+  "version": "2026-07-26",
+  "groups": {
+    "acdinfo": ["ywrysw", "ywfjdc", "ywmtc", "bjrjs", "sgfssj", "sfsgxc", "jdcsl", "sgyy"],
+    "acdhuman1": ["xm1", "hpzl1", "hphm1", "sfzmhm1", "sfzmwh1", "sjhm1", "sjwh1", "csbw1"],
+    "acdhuman2": ["xm2", "hpzl2", "hphm2", "sfzmhm2", "sfzmwh2", "sjhm2", "sjwh2", "csbw2"]
+  },
+  "fields": [
+    {"key": "ywrysw", "group": "acdinfo", "type": "boolean", "description": "是否有人员伤亡", "sensitive": false, "external_write": true},
+    {"key": "ywfjdc", "group": "acdinfo", "type": "boolean", "description": "是否涉及非机动车", "sensitive": false, "external_write": true},
+    {"key": "ywmtc", "group": "acdinfo", "type": "boolean", "description": "是否涉及摩托车", "sensitive": false, "external_write": true},
+    {"key": "bjrjs", "group": "acdinfo", "type": "string", "description": "报警人角色/描述", "sensitive": false, "external_write": true},
+    {"key": "sgfssj", "group": "acdinfo", "type": "datetime", "description": "事故发生时间", "sensitive": false, "external_write": true},
+    {"key": "sfsgxc", "group": "acdinfo", "type": "boolean", "description": "是否在事故现场", "sensitive": false, "external_write": true},
+    {"key": "jdcsl", "group": "acdinfo", "type": "integer", "description": "事故机动车数量", "sensitive": false, "external_write": true},
+    {"key": "sgyy", "group": "acdinfo", "type": "string", "description": "事故原因/经过", "sensitive": false, "external_write": true},
+    {"key": "xm1", "group": "acdhuman1", "type": "string", "description": "驾驶员1姓名", "sensitive": true, "external_write": true},
+    {"key": "hpzl1", "group": "acdhuman1", "type": "string", "description": "驾驶员1号牌种类", "sensitive": false, "external_write": true},
+    {"key": "hphm1", "group": "acdhuman1", "type": "license_plate", "description": "驾驶员1车牌号", "sensitive": true, "external_write": true},
+    {"key": "sfzmhm1", "group": "acdhuman1", "type": "national_id", "description": "驾驶员1身份证号码", "sensitive": true, "external_write": true},
+    {"key": "sfzmwh1", "group": "acdhuman1", "type": "string", "description": "驾驶员1身份证尾号兼容字段", "sensitive": true, "external_write": false},
+    {"key": "sjhm1", "group": "acdhuman1", "type": "phone", "description": "驾驶员1手机号码", "sensitive": true, "external_write": true},
+    {"key": "sjwh1", "group": "acdhuman1", "type": "string", "description": "驾驶员1手机号尾号兼容字段", "sensitive": true, "external_write": false},
+    {"key": "csbw1", "group": "acdhuman1", "type": "string", "description": "驾驶员1车辆车损部位", "sensitive": false, "external_write": true},
+    {"key": "xm2", "group": "acdhuman2", "type": "string", "description": "驾驶员2姓名", "sensitive": true, "external_write": true},
+    {"key": "hpzl2", "group": "acdhuman2", "type": "string", "description": "驾驶员2号牌种类", "sensitive": false, "external_write": true},
+    {"key": "hphm2", "group": "acdhuman2", "type": "license_plate", "description": "驾驶员2车牌号", "sensitive": true, "external_write": true},
+    {"key": "sfzmhm2", "group": "acdhuman2", "type": "national_id", "description": "驾驶员2身份证号码", "sensitive": true, "external_write": true},
+    {"key": "sfzmwh2", "group": "acdhuman2", "type": "string", "description": "驾驶员2身份证尾号兼容字段", "sensitive": true, "external_write": false},
+    {"key": "sjhm2", "group": "acdhuman2", "type": "phone", "description": "驾驶员2手机号码", "sensitive": true, "external_write": true},
+    {"key": "sjwh2", "group": "acdhuman2", "type": "string", "description": "驾驶员2手机号尾号兼容字段", "sensitive": true, "external_write": false},
+    {"key": "csbw2", "group": "acdhuman2", "type": "string", "description": "驾驶员2车辆车损部位", "sensitive": false, "external_write": true}
+  ],
+  "read_keys": ["all", "acdinfo", "acdhuman1", "acdhuman2"],
+  "internal_write_denylist": [
+    "phase",
+    "stage_code",
+    "clarification_counts",
+    "no_response_count",
+    "handoff_reason",
+    "state_version",
+    "messages",
+    "input_event",
+    "photo_step"
+  ],
+  "legacy_serialization": {
+    "get_info_value": "json_encoded_string",
+    "boolean": {"true": "1", "false": "0"},
+    "missing": ""
+  }
+}
--- a/docs/domain/stage-codes.json
+++ b/docs/domain/stage-codes.json
@@ -0,0 +1,40 @@
+{
+  "version": "2026-07-26",
+  "sources": [
+    "src/api/endpoints.py",
+    "docs/视频快处智能信息采集机器人交互接口文档V1.0.9.docx",
+    "workflow/20260726/事故信息采集20260726.json",
+    "prompts/20260723/单车拍照.txt",
+    "prompts/20260723/双车拍照.txt"
+  ],
+  "codes": [
+    {"code": "0000", "name": "通话结束", "phase": "complete", "terminal": true, "external_code": "0000", "producer": "workflow"},
+    {"code": "0001", "name": "主动转人工", "phase": "handoff", "terminal": true, "external_code": "0001", "producer": "deterministic_or_llm"},
+    {"code": "0002", "name": "语义连续无法识别", "phase": "handoff", "terminal": true, "external_code": "0002", "producer": "deterministic"},
+    {"code": "0003", "name": "人伤或复杂情况转人工", "phase": "handoff", "terminal": true, "external_code": "0003", "producer": "deterministic_or_llm"},
+    {"code": "0004", "name": "长时间无回复", "phase": "handoff", "terminal": true, "external_code": "0004", "producer": "deterministic"},
+    {"code": "0005", "name": "连续拍照识别失败", "phase": "handoff", "terminal": true, "external_code": "0005", "producer": "deterministic"},
+    {"code": "1001", "name": "未准备好通话", "phase": "ready_gate", "terminal": false, "external_code": "1001", "producer": "deterministic"},
+    {"code": "1002", "name": "事故信息采集中", "phase": "collection", "terminal": false, "external_code": "1002", "producer": "llm"},
+    {"code": "2000", "name": "单车车前照片", "phase": "single_photo", "terminal": false, "external_code": "2000", "producer": "deterministic_or_llm"},
+    {"code": "2001", "name": "单车碰撞部位照片", "phase": "single_photo", "terminal": false, "external_code": "2001", "producer": "deterministic"},
+    {"code": "2002", "name": "被撞物品照片", "phase": "single_photo", "terminal": false, "external_code": "2002", "producer": "deterministic"},
+    {"code": "2003", "name": "本人正面照片", "phase": "single_photo", "terminal": false, "external_code": "2003", "producer": "deterministic"},
+    {"code": "2004", "name": "确认单车车牌", "phase": "single_photo", "terminal": false, "external_code": "2004", "producer": "llm"},
+    {"code": "2005", "name": "确认单车车损位置", "phase": "single_photo", "terminal": false, "external_code": "2005", "producer": "llm"},
+    {"code": "2010", "name": "第一辆车侧前方照片", "phase": "double_photo", "terminal": false, "external_code": "2010", "producer": "deterministic_or_llm"},
+    {"code": "2011", "name": "第一辆车碰撞部位照片", "phase": "double_photo", "terminal": false, "external_code": "2011", "producer": "deterministic"},
+    {"code": "2012", "name": "第二辆车碰撞部位照片", "phase": "double_photo", "terminal": false, "external_code": "2012", "producer": "deterministic"},
+    {"code": "2013", "name": "第二辆车侧后方车牌照片", "phase": "double_photo", "terminal": false, "external_code": "2013", "producer": "deterministic"},
+    {"code": "2014", "name": "另一方驾驶人正面照片", "phase": "double_photo", "terminal": false, "external_code": "2014", "producer": "deterministic"},
+    {"code": "2015", "name": "本人正面照片", "phase": "double_photo", "terminal": false, "external_code": "2015", "producer": "deterministic"},
+    {"code": "2016", "name": "确认双车车牌", "phase": "double_photo", "terminal": false, "external_code": "2016", "producer": "llm"},
+    {"code": "3001", "name": "单车当事人信息确认", "phase": "single_verification", "terminal": false, "external_code": "1002", "producer": "llm"},
+    {"code": "3002", "name": "双车当事人信息确认", "phase": "double_verification", "terminal": false, "external_code": "1002", "producer": "llm"}
+  ],
+  "legacy_aliases": {
+    "2006": "2004",
+    "2017": "2016",
+    "2020": "0002"
+  }
+}
--- a/docs/domain/stage-transitions.json
+++ b/docs/domain/stage-transitions.json
@@ -0,0 +1,46 @@
+{
+  "version": "2026-07-26",
+  "global_transitions": {
+    "explicit_handoff": "0001",
+    "injury_or_complex": "0003",
+    "two_invalid_clarifications": "0002",
+    "two_no_response_events": "0004",
+    "photo_recognition_failed_event": "0005"
+  },
+  "allowed": {
+    "1001": ["1001", "1002", "0001", "0003", "0004"],
+    "1002": ["1002", "2000", "2010", "0000", "0001", "0002", "0003", "0004"],
+    "2000": ["2000", "2001", "0001", "0003", "0004", "0005"],
+    "2001": ["2001", "2002", "0001", "0003", "0004", "0005"],
+    "2002": ["2002", "2003", "0001", "0003", "0004", "0005"],
+    "2003": ["2003", "2004", "0001", "0003", "0004", "0005"],
+    "2004": ["2004", "2005", "0001", "0002", "0003", "0004", "0005"],
+    "2005": ["2005", "3001", "0001", "0002", "0003", "0004", "0005"],
+    "2010": ["2010", "2011", "0001", "0003", "0004", "0005"],
+    "2011": ["2011", "2012", "0001", "0003", "0004", "0005"],
+    "2012": ["2012", "2013", "0001", "0003", "0004", "0005"],
+    "2013": ["2013", "2014", "0001", "0003", "0004", "0005"],
+    "2014": ["2014", "2015", "0001", "0003", "0004", "0005"],
+    "2015": ["2015", "2016", "0001", "0003", "0004", "0005"],
+    "2016": ["2016", "3002", "0001", "0002", "0003", "0004", "0005"],
+    "3001": ["3001", "0000", "0001", "0002", "0003", "0004"],
+    "3002": ["3002", "0000", "0001", "0002", "0003", "0004"],
+    "0000": [],
+    "0001": [],
+    "0002": [],
+    "0003": [],
+    "0004": [],
+    "0005": []
+  },
+  "photo_sequences": {
+    "single": ["2000", "2001", "2002", "2003", "2004", "2005", "3001"],
+    "double": ["2010", "2011", "2012", "2013", "2014", "2015", "2016", "3002"]
+  },
+  "notes": [
+    "同状态迁移表示无效输入后重复当前问题或固定指令。",
+    "3001 和 3002 是内部信息确认状态，对外兼容码统一为 1002。",
+    "终止状态不允许继续迁移；新请求必须创建或显式重置 session。",
+    "0003 在照片和信息确认阶段仍保留全局安全优先级。",
+    "迁移矩阵是 LangGraph 实现的权威基线；FastGPT 当前不会在服务端执行该校验。"
+  ]
+}
--- a/docs/langgraph-backend-migration-plan.md
+++ b/docs/langgraph-backend-migration-plan.md
--- a/docs/langgraph-minimal-slice.md
+++ b/docs/langgraph-minimal-slice.md
@@ -0,0 +1,82 @@
+# LangGraph 最小开发纵切
+
+> 状态：仅用于本地开发和测试
+> 日期：2026-07-27
+
+## 启用
+
+在被 Git 忽略的 `.env.local` 中配置：
+
+```text
+ZNJJ_ENVIRONMENT=development
+AGENT_BACKEND=langgraph
+LANGGRAPH_CHECKPOINTER=memory
+
+LLM_API_KEY=...
+LLM_BASE_URL=https://api.openai.com/v1
+LLM_MODEL=...
+LLM_TIMEOUT_SECONDS=60
+LLM_MAX_RETRIES=2
+```
+
+不设置 `AGENT_BACKEND` 时默认使用 FastGPT。
+
+## 当前调用链
+
+```text
+/chat
+  -> ChatBackend
+  -> LangGraphBackend
+  -> StateGraph
+  -> generate_response
+  -> OpenAI-compatible LLM
+  -> ChatResult
+  -> 现有非流式/SSE API Adapter
+```
+
+`InMemorySaver` 使用 `sessionId` 作为 `thread_id`，同一进程内同一 session
+可以恢复 `turn_count` 等图状态；不同 session 相互隔离。
+
+## 当前已有能力
+
+- FastGPT/LangGraph 后端配置切换。
+- 最小 Pydantic Settings 和条件化启动校验。
+- 实际使用 LangGraph `StateGraph`。
+- 开发/测试使用 `InMemorySaver`。
+- OpenAI-compatible LLM 节点。
+- 非流式 `/chat` 适配。
+- 通过完整结果桥接现有 SSE；`formUpdate` 先于文本发送。
+- Fake LLM 下的多轮 thread 隔离测试。
+
+## 明确限制
+
+当前图只有一个模型节点，目的是尽早建立可执行骨架。以下尚未实现：
+
+- 领域状态、输入事件和确定性路由节点；
+- 状态码枚举及迁移合法性校验；
+- 单车/双车拍照状态机；
+- 字段提取、验证和真实 `formUpdate`；
+- 模型 token 级流式输出；
+- PostgreSQL checkpointer；
+- session 并发、版本和幂等；
+- `/set_info`、`/get_info` 的业务状态迁移。
+
+因此：
+
+- `LANGGRAPH_CHECKPOINTER=postgres` 当前会启动失败；
+- staging/production 禁止使用 memory checkpointer；
+- `/set_info`、`/get_info`、`/delete_session` 暂时仍需要 FastGPT 配置；
+- 不得把当前 LangGraph backend 接入生产流量。
+
+## 下一步
+
+直接在现有图中加入 Phase 5 的纯确定性骨架：
+
+1. `AccidentState`、输入事件和状态枚举；
+2. state prefix parser 与迁移 validator；
+3. `normalize_input`、`route_phase`；
+4. 单车/双车拍照事件推进；
+5. 无回复和澄清计数；
+6. 对应参数化 Graph 测试。
+
+这些节点完成后再接 PostgreSQL 和业务状态 repository。
--- a/docs/视频快处智能信息采集机器人交互接口文档V1.0.8.docx
+++ b/docs/视频快处智能信息采集机器人交互接口文档V1.0.8.docx
--- a/requirements.txt
+++ b/requirements.txt
@@ -10,7 +10,8 @@ pytest-asyncio>=0.21.0
 pytest-cov>=4.1.0
 pillow>=10.4.0
 paho-mqtt>=2.1.0
-pydantic-settings==2.1.0
+pydantic-settings==2.14.2
+langgraph==1.2.9
 python-multipart==0.0.6
 python-jose[cryptography]==3.3.0
 passlib[bcrypt]==1.7.4
--- a/src/.env.example
+++ b/src/.env.example
@@ -1,13 +0,0 @@
-DATABASE_URL=sqlite:///./test.db
-SECRET_KEY=your_secret_key
-DEBUG=True
-
-ANALYSIS_SERVICE_URL=http://127.0.0.1:3030
-ANALYSIS_AUTH_TOKEN=fastgpt-hSPnXMoBNGVAEpTLkQT3YfAnN26gQSyvLd4ABL1MRDoh68nL4RDlopFHXqmH8
-APP_ID=683ea1bc86197e19f71fc1ae
-DELETE_SESSION_URL=http://127.0.0.1:3030/api/core/chat/delHistory?chatId={chatId}&appId={appId}
-DELETE_CHAT_URL=http://127.0.0.1:3030/api/core/chat/item/delete?contentId={contentId}&chatId={chatId}&appId={appId}
-GET_CHAT_RECORDS_URL=http://127.0.0.1:3030/api/core/chat/getPaginationRecords
-
-# Voice demo (Pipecat /ws-product). Relative to project root, or an absolute path.
-VOICE_CONFIG=config/voice.json
--- a/src/agent/init.py
+++ b/src/agent/init.py
@@ -0,0 +1,11 @@
+"""LangGraph accident workflow package."""
+
+from .graph import build_accident_graph
+from .state import AccidentGraphState, GeneratedTurn, ResponseGenerator
+
+__all__ = [
+    "AccidentGraphState",
+    "GeneratedTurn",
+    "ResponseGenerator",
+    "build_accident_graph",
+]
--- a/src/agent/graph.py
+++ b/src/agent/graph.py
@@ -0,0 +1,37 @@
+"""First runnable LangGraph workflow.
+
+This intentionally contains one model node. Deterministic routing, transition
+validation and typed business events are added in Phase 5 without changing the
+public ChatBackend boundary.
+"""
+
+from typing import Any
+
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import END, START, StateGraph
+
+from .state import AccidentGraphState, ResponseGenerator
+
+
+def build_accident_graph(
+    response_generator: ResponseGenerator,
+    *,
+    checkpointer: Any | None = None,
+):
+    """Compile the minimal async graph with thread-scoped memory."""
+
+    async def generate_response(
+        state: AccidentGraphState,
+    ) -> AccidentGraphState:
+        generated = await response_generator.generate(state)
+        return {
+            "response_text": generated.content,
+            "form_update": generated.form_update,
+            "turn_count": state.get("turn_count", 0) + 1,
+        }
+
+    builder = StateGraph(AccidentGraphState)
+    builder.add_node("generate_response", generate_response)
+    builder.add_edge(START, "generate_response")
+    builder.add_edge("generate_response", END)
+    return builder.compile(checkpointer=checkpointer or InMemorySaver())
--- a/src/agent/state.py
+++ b/src/agent/state.py
@@ -0,0 +1,30 @@
+"""Minimal serializable state for the first LangGraph slice."""
+
+from dataclasses import dataclass, field
+from typing import Any, Protocol, TypedDict
+
+
+class AccidentGraphState(TypedDict, total=False):
+    session_id: str
+    input_text: str
+    need_form_update: bool
+    response_text: str
+    stage_code: str
+    form_update: dict[str, Any]
+    turn_count: int
+
+
+@dataclass(frozen=True)
+class GeneratedTurn:
+    content: str
+    form_update: dict[str, Any] = field(default_factory=dict)
+
+
+class ResponseGenerator(Protocol):
+    async def generate(self, state: AccidentGraphState) -> GeneratedTurn:
+        """Generate one state-prefixed response from graph state."""
+        ...
+
+    async def aclose(self) -> None:
+        """Release any owned network resources."""
+        ...
--- a/src/api/endpoints.py
+++ b/src/api/endpoints.py
@@ -1,12 +1,18 @@
 from fastapi import APIRouter, HTTPException, Depends
 from fastapi.responses import StreamingResponse
 from ..schemas.models import ProcessRequest_chat, ProcessResponse_chat, ProcessRequest_get, ProcessResponse_get, ProcessRequest_set, ProcessResponse_set, ProcessResponse_delete_session, ProcessRequest_delete_session
-from fastgpt_client import AsyncChatClient, aiter_stream_events
-from fastgpt_client.exceptions import (
-    APIError, AuthenticationError, RateLimitError, ValidationError
+from fastgpt_client import AsyncChatClient
+from ..backends.chat import (
+    ChatBackend,
+    ChatBackendAPIError,
+    ChatBackendAuthenticationError,
+    ChatBackendRateLimitError,
+    ChatInput,
+    FormUpdate,
+    TextDelta,
 )
-from ..core.fastgpt_client import get_fastgpt_client
-from ..core.config import Config
+from ..core.fastgpt_client import get_chat_backend, get_fastgpt_client
+from ..core.config import get_settings
 from ..utils.text_chunker import SentenceTextChunker, SentenceTextChunkerConfig
 from loguru import logger
 import json
@@ -14,7 +20,6 @@ import re
 import time

 router = APIRouter()
-FORM_EXTRACT_MODULE_NAME = "文本内容提取事故信息"
 STATE_TAG_PATTERN = re.compile(r"<state>\s*(\d+)\s*</state>", flags=re.DOTALL)
 STATUS_CODE_MAP = {
    '0000': '结束通话',
@@ -39,6 +44,13 @@ STATUS_CODE_MAP = {
    '2016': '确认双车中的车牌'
 }

+
+def get_fastgpt_app_id() -> str:
+    app_id = get_settings().fastgpt_app_id
+    if not app_id:
+        raise RuntimeError("FastGPT APP_ID is not configured")
+    return app_id
+
 def normalize_stage_code(stage_code: str) -> str:
    """Normalize FastGPT stage codes to external API stage codes."""
    if stage_code in ['3001', '3002', '1002']:
@@ -60,45 +72,6 @@ def extract_first_state_and_clean_content(text: str) -> tuple[str | None, str]:
    return match.group(1), STATE_TAG_PATTERN.sub("", text)


-def parse_json_value(value):
-    """Parse JSON string values when possible."""
-    parsed = value
-    for _ in range(3):
-        if not isinstance(parsed, str):
-            return parsed
-        parsed = parsed.strip()
-        if not parsed:
-            return {}
-        try:
-            parsed = json.loads(parsed)
-        except json.JSONDecodeError:
-            return parsed
-    return parsed
-
-
-def extract_form_update_from_flow_nodes(nodes):
-    """Extract form update data from the configured FastGPT content-extract node."""
-    if not isinstance(nodes, list):
-        return {}
-
-    for node in nodes:
-        if not isinstance(node, dict):
-            continue
-        if node.get("moduleName") != FORM_EXTRACT_MODULE_NAME:
-            continue
-
-        extract_result = node.get("extractResult", {})
-        if not isinstance(extract_result, dict):
-            return {}
-
-        form_update = extract_result.get("formUpdate", "")
-        if not form_update:
-            return {}
-        return parse_json_value(form_update)
-
-    return {}
-
-
 def format_set_info_input(payload: dict, include_input_info: bool) -> str:
    """Build optional setInfo input for FastGPT helper calls."""
    if not include_input_info:
@@ -113,7 +86,7 @@ async def delete_last_two_chat_records(
    try:
        # Get chat records using SDK
        response = await client.get_chat_records(
-            appId=Config.FASTGPT_APP_ID,
+            appId=get_fastgpt_app_id(),
            chatId=session_id,
            offset=0,
            pageSize=10
@@ -132,7 +105,7 @@ async def delete_last_two_chat_records(
        # Delete records using SDK
        for content_id in last_two_data_ids:
            delete_response = await client.delete_chat_record(
-                appId=Config.FASTGPT_APP_ID,
+                appId=get_fastgpt_app_id(),
                chatId=session_id,
                contentId=content_id
            )
@@ -156,13 +129,17 @@ def create_sse_event(event: str, data: dict) -> str:
 async def chat(
    request: ProcessRequest_chat,
    stream: bool = False,
-    client: AsyncChatClient = Depends(get_fastgpt_client)
+    backend: ChatBackend = Depends(get_chat_backend)
 ):
    """Handle chat completion request."""
    json_data = request.model_dump()
    need_form_update = json_data.get('needFormUpdate', False)
    use_text_chunk = json_data.get('useTextChunk', False)
-    chat_variables = {'needFormUpdate': need_form_update}
+    chat_input = ChatInput(
+        session_id=json_data['sessionId'],
+        text=json_data['text'],
+        need_form_update=need_form_update,
+    )
    request_started_at = time.perf_counter()
    logger.info(
        "Chat request received "
@@ -193,16 +170,8 @@ async def chat(
                else None
            )
            try:
-                # Use SDK's create_chat_completion with stream=True
-                response = await client.create_chat_completion(
-                    messages=[{"role": "user", "content": json_data['text']}],
-                    chatId=json_data['sessionId'],
-                    stream=True,
-                    detail=True,
-                    variables=chat_variables
-                )
                logger.info(
-                    "FastGPT stream response opened "
+                    "Chat backend stream opened "
                    f"sessionId={json_data['sessionId']} "
                    f"open_latency_ms={(time.perf_counter() - stream_started_at) * 1000:.1f}"
                )
@@ -279,41 +248,34 @@ async def chat(
                    state_filter_buffer = ""
                    return cleaned
                
-                async for event in aiter_stream_events(response):
+                async for event in backend.stream(chat_input):
                    try:
                        if not first_event_logged:
                            first_event_logged = True
                            logger.info(
-                                "FastGPT stream first event "
-                                f"sessionId={json_data['sessionId']} kind={event.kind} "
+                                "Chat backend stream first event "
+                                f"sessionId={json_data['sessionId']} "
+                                f"kind={type(event).__name__} "
                                f"ttfb_ms={(time.perf_counter() - stream_started_at) * 1000:.1f}"
                            )

-                        if event.kind == "flowResponses" and not module_form_sent:
-                            form_update = extract_form_update_from_flow_nodes(event.data)
-                            if form_update:
-                                form_update_payload = form_update
+                        if isinstance(event, FormUpdate) and not module_form_sent:
+                            if event.data:
+                                form_update_payload = event.data
                                logger.info(
-                                    "FastGPT stream formUpdate extracted "
+                                    "Chat backend stream formUpdate received "
                                    f"sessionId={json_data['sessionId']} "
-                                    f"type={type(form_update).__name__} "
-                                    f"formUpdate={form_update!r}"
+                                    f"type={type(event.data).__name__} "
+                                    f"formUpdate={event.data!r}"
                                )
-                                yield flush_form_update(form_update)
+                                yield flush_form_update(event.data)
                                module_form_sent = True
                            continue

-                        if event.kind not in {"answer", "fastAnswer", "data"}:
+                        if not isinstance(event, TextDelta):
                            continue

-                        data = event.data
-                        if not isinstance(data, dict):
-                            continue
-
-                        try:
-                            delta_content = data['choices'][0]['delta'].get('content', '')
-                        except (KeyError, IndexError):
-                            delta_content = ''
+                        delta_content = event.text
                        if not delta_content:
                            continue

@@ -400,23 +362,14 @@ async def chat(
        return StreamingResponse(event_generator(), media_type="text/event-stream")

    try:
-        # Use SDK's create_chat_completion
-        response = await client.create_chat_completion(
-            messages=[{"role": "user", "content": json_data['text']}],
-            chatId=json_data['sessionId'],
-            stream=False,
-            detail=True,
-            variables=chat_variables
-        )
-        response.raise_for_status()
-        data = response.json()
+        result = await backend.complete(chat_input)
        logger.info(
-            "FastGPT non-stream response received "
+            "Chat backend non-stream response received "
            f"sessionId={json_data['sessionId']} "
            f"latency_ms={(time.perf_counter() - request_started_at) * 1000:.1f}"
        )
        
-    except AuthenticationError as e:
+    except ChatBackendAuthenticationError as e:
        logger.error(f"Authentication error: {e}")
        return ProcessResponse_chat(
            sessionId=json_data['sessionId'],
@@ -427,7 +380,7 @@ async def chat(
            code="401",
            msg="认证失败"
        )
-    except RateLimitError as e:
+    except ChatBackendRateLimitError as e:
        logger.error(f"Rate limit error: {e}")
        return ProcessResponse_chat(
            sessionId=json_data['sessionId'],
@@ -438,7 +391,7 @@ async def chat(
            code="429",
            msg="请求过于频繁，请稍后重试"
        )
-    except APIError as e:
+    except ChatBackendAPIError as e:
        logger.error(f"API error: {e}")
        return ProcessResponse_chat(
            sessionId=json_data['sessionId'],
@@ -462,39 +415,10 @@ async def chat(
        )
    
    try:
-        # Extract content from FastGPT response
-        content = data['choices'][0]['message']['content']
-        logger.info(f"FastGPT服务返回信息content: {content}")
+        content = result.content
+        logger.info(f"Chat backend returned content: {content}")

-        finish_reason = data['choices'][0]['finish_reason']
-
-        # Extract state variables
-        state = data.get('newVariables', {}).get('state', {})
-        if isinstance(state, str):
-            state = json.loads(state)
-
-        transfer_to_human = state.get("transfer_to_human", False)
-        ywrysw = state.get("ywrysw", False)
-        ywfjdc = state.get("ywfjdc", False)
-        ywmtc = state.get("ywmtc", False)
-        jdcsl = state.get("jdcsl", 0)
-        accident_info_complete = state.get("accident_info_complete", False)
-        user_is_ready = state.get("user_is_ready", False)
-        if isinstance(user_is_ready, str):
-            user_is_ready = user_is_ready.lower() == 'true'
-        driver_info_complete = state.get("driver_info_complete", False)
-        drivers_info_complete = state.get("drivers_info_complete", False)
-        driver_info_check = state.get("drivers_info_check", False)
-        drivers_info_check = state.get("drivers_info_check", False)
-
-        logger.debug(f"State variables: {data.get('newVariables', {})}")
-
-        # Parse content - sometimes content is a string, sometimes it is a list
        content_stage_code = None
-        if isinstance(content, list):
-            logger.debug("content是一个list")
-            content = content[0]['text']['content']
-        
        if isinstance(content, str):
            logger.debug("content是一个str")
            content_stage_code, content = extract_first_state_and_clean_content(content)
@@ -509,10 +433,12 @@ async def chat(
            logger.error(f"content既不是list也不是str, type: {type(content)}")
            raise ValueError("大模型回复不是list也不是str")

-        nextStageCode = content_stage_code or data['newVariables']['status_code']
+        nextStageCode = content_stage_code or result.status_code
+        if not nextStageCode:
+            raise ValueError("大模型回复中缺少state")
        nextStageCode = normalize_stage_code(nextStageCode)
        nextStage = STATUS_CODE_MAP.get(nextStageCode, '')
-        form_update = extract_form_update_from_flow_nodes(data.get("responseData", []))
+        form_update = result.form_update
        logger.info(
            "Chat non-stream completed "
            f"sessionId={json_data['sessionId']} "
@@ -754,7 +680,7 @@ async def delete_session(
    try:
        # Use SDK's delete_chat_history
        response = await client.delete_chat_history(
-            appId=Config.FASTGPT_APP_ID,
+            appId=get_fastgpt_app_id(),
            chatId=chat_id
        )
        response.raise_for_status()
--- a/src/backends/init.py
+++ b/src/backends/init.py
@@ -0,0 +1,29 @@
+"""Backend adapters used by the public API layer."""
+
+from .chat import (
+    ChatBackend,
+    ChatBackendAPIError,
+    ChatBackendAuthenticationError,
+    ChatBackendError,
+    ChatBackendRateLimitError,
+    ChatInput,
+    ChatResult,
+    FormUpdate,
+    TextDelta,
+)
+from .fastgpt import FastGPTBackend
+from .langgraph import LangGraphBackend
+
+__all__ = [
+    "ChatBackend",
+    "ChatBackendAPIError",
+    "ChatBackendAuthenticationError",
+    "ChatBackendError",
+    "ChatBackendRateLimitError",
+    "ChatInput",
+    "ChatResult",
+    "FastGPTBackend",
+    "LangGraphBackend",
+    "FormUpdate",
+    "TextDelta",
+]
--- a/src/backends/chat.py
+++ b/src/backends/chat.py
@@ -0,0 +1,72 @@
+"""Backend-neutral chat contract."""
+
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+from typing import Any, Protocol
+
+
+@dataclass(frozen=True)
+class ChatInput:
+    """Input shared by all chat backend implementations."""
+
+    session_id: str
+    text: str
+    need_form_update: bool = False
+
+
+@dataclass(frozen=True)
+class TextDelta:
+    """A piece of raw model text.
+
+    The text may contain a partial ``<state>...</state>`` prefix. Parsing that
+    public protocol remains the responsibility of the FastAPI layer.
+    """
+
+    text: str
+
+
+@dataclass(frozen=True)
+class FormUpdate:
+    """A structured form update produced alongside model text."""
+
+    data: Any
+
+
+ChatStreamEvent = TextDelta | FormUpdate
+
+
+@dataclass(frozen=True)
+class ChatResult:
+    """Backend-neutral result for a non-streaming chat request."""
+
+    content: str
+    status_code: str | None = None
+    form_update: Any = field(default_factory=dict)
+
+
+class ChatBackend(Protocol):
+    """Contract implemented by FastGPT today and LangGraph later."""
+
+    def stream(self, chat_input: ChatInput) -> AsyncIterator[ChatStreamEvent]:
+        """Stream raw text and structured side-channel events."""
+        ...
+
+    async def complete(self, chat_input: ChatInput) -> ChatResult:
+        """Return one complete backend-neutral chat result."""
+        ...
+
+
+class ChatBackendError(Exception):
+    """Base error raised by a chat backend adapter."""
+
+
+class ChatBackendAuthenticationError(ChatBackendError):
+    """The backend rejected its configured credentials."""
+
+
+class ChatBackendRateLimitError(ChatBackendError):
+    """The backend rejected the request because of rate limiting."""
+
+
+class ChatBackendAPIError(ChatBackendError):
+    """The backend API failed or returned an invalid response."""
--- a/src/backends/fastgpt.py
+++ b/src/backends/fastgpt.py
@@ -0,0 +1,148 @@
+"""FastGPT implementation of the backend-neutral chat contract."""
+
+import json
+from collections.abc import AsyncIterator
+from typing import Any
+
+from fastgpt_client import AsyncChatClient, aiter_stream_events
+from fastgpt_client.exceptions import APIError, AuthenticationError, RateLimitError
+
+from .chat import (
+    ChatBackendAPIError,
+    ChatBackendAuthenticationError,
+    ChatBackendRateLimitError,
+    ChatInput,
+    ChatResult,
+    FormUpdate,
+    TextDelta,
+)
+
+
+FORM_EXTRACT_MODULE_NAME = "文本内容提取事故信息"
+
+
+def _parse_json_value(value: Any) -> Any:
+    parsed = value
+    for _ in range(3):
+        if not isinstance(parsed, str):
+            return parsed
+        parsed = parsed.strip()
+        if not parsed:
+            return {}
+        try:
+            parsed = json.loads(parsed)
+        except json.JSONDecodeError:
+            return parsed
+    return parsed
+
+
+def _extract_form_update(nodes: Any) -> Any:
+    if not isinstance(nodes, list):
+        return {}
+
+    for node in nodes:
+        if not isinstance(node, dict):
+            continue
+        if node.get("moduleName") != FORM_EXTRACT_MODULE_NAME:
+            continue
+
+        extract_result = node.get("extractResult", {})
+        if not isinstance(extract_result, dict):
+            return {}
+
+        form_update = extract_result.get("formUpdate", "")
+        return _parse_json_value(form_update) if form_update else {}
+
+    return {}
+
+
+def _extract_content(data: dict[str, Any]) -> str:
+    try:
+        content = data["choices"][0]["message"]["content"]
+    except (KeyError, IndexError, TypeError) as exc:
+        raise ChatBackendAPIError("FastGPT response is missing message content") from exc
+
+    if isinstance(content, list):
+        try:
+            content = content[0]["text"]["content"]
+        except (KeyError, IndexError, TypeError) as exc:
+            raise ChatBackendAPIError(
+                "FastGPT response contains invalid list content"
+            ) from exc
+
+    if not isinstance(content, str):
+        raise ChatBackendAPIError("FastGPT message content is not text")
+    return content
+
+
+class FastGPTBackend:
+    """Translate FastGPT SDK calls and events into the neutral chat contract."""
+
+    def __init__(self, client: AsyncChatClient):
+        self._client = client
+
+    async def stream(self, chat_input: ChatInput) -> AsyncIterator[TextDelta | FormUpdate]:
+        try:
+            response = await self._client.create_chat_completion(
+                messages=[{"role": "user", "content": chat_input.text}],
+                chatId=chat_input.session_id,
+                stream=True,
+                detail=True,
+                variables={"needFormUpdate": chat_input.need_form_update},
+            )
+
+            async for event in aiter_stream_events(response):
+                if event.kind == "flowResponses":
+                    form_update = _extract_form_update(event.data)
+                    if form_update:
+                        yield FormUpdate(form_update)
+                    continue
+
+                if event.kind not in {"answer", "fastAnswer", "data"}:
+                    continue
+                if not isinstance(event.data, dict):
+                    continue
+
+                try:
+                    content = event.data["choices"][0]["delta"].get("content", "")
+                except (KeyError, IndexError, TypeError, AttributeError):
+                    content = ""
+                if content:
+                    yield TextDelta(content)
+        except AuthenticationError as exc:
+            raise ChatBackendAuthenticationError(str(exc)) from exc
+        except RateLimitError as exc:
+            raise ChatBackendRateLimitError(str(exc)) from exc
+        except APIError as exc:
+            raise ChatBackendAPIError(str(exc)) from exc
+
+    async def complete(self, chat_input: ChatInput) -> ChatResult:
+        try:
+            response = await self._client.create_chat_completion(
+                messages=[{"role": "user", "content": chat_input.text}],
+                chatId=chat_input.session_id,
+                stream=False,
+                detail=True,
+                variables={"needFormUpdate": chat_input.need_form_update},
+            )
+            response.raise_for_status()
+            data = response.json()
+        except AuthenticationError as exc:
+            raise ChatBackendAuthenticationError(str(exc)) from exc
+        except RateLimitError as exc:
+            raise ChatBackendRateLimitError(str(exc)) from exc
+        except APIError as exc:
+            raise ChatBackendAPIError(str(exc)) from exc
+
+        if not isinstance(data, dict):
+            raise ChatBackendAPIError("FastGPT response body is not an object")
+
+        status_code = data.get("newVariables", {}).get("status_code")
+        if status_code is not None:
+            status_code = str(status_code)
+
+        return ChatResult(
+            content=_extract_content(data),
+            status_code=status_code,
+            form_update=_extract_form_update(data.get("responseData", [])),
+        )
--- a/src/backends/langgraph.py
+++ b/src/backends/langgraph.py
@@ -0,0 +1,42 @@
+"""Adapt the compiled accident LangGraph to the neutral chat contract."""
+
+from typing import Any
+
+from .chat import ChatInput, ChatResult, FormUpdate, TextDelta
+from ..agent.state import ResponseGenerator
+
+
+class LangGraphBackend:
+    def __init__(self, graph: Any, response_generator: ResponseGenerator):
+        self._graph = graph
+        self._response_generator = response_generator
+
+    async def complete(self, chat_input: ChatInput) -> ChatResult:
+        state = await self._graph.ainvoke(
+            {
+                "session_id": chat_input.session_id,
+                "input_text": chat_input.text,
+                "need_form_update": chat_input.need_form_update,
+            },
+            config={
+                "configurable": {
+                    "thread_id": chat_input.session_id,
+                }
+            },
+        )
+        return ChatResult(
+            content=state["response_text"],
+            status_code=state.get("stage_code"),
+            form_update=state.get("form_update", {}),
+        )
+
+    async def stream(self, chat_input: ChatInput):
+        """Bridge the first graph slice to SSE; token streaming comes later."""
+
+        result = await self.complete(chat_input)
+        if result.form_update:
+            yield FormUpdate(result.form_update)
+        yield TextDelta(result.content)
+
+    async def aclose(self) -> None:
+        await self._response_generator.aclose()
--- a/src/core/config.py
+++ b/src/core/config.py
@@ -1,30 +1,152 @@
-import os
-from dotenv import load_dotenv
+"""Minimal typed settings needed to run FastGPT or the first LangGraph slice."""

-load_dotenv()
+from functools import lru_cache
+from typing import Literal

-class Config:
-    PROJECT_NAME = "Flexible Employment Analysis API"
-    API_V1_STR = "/api/v1"
-    SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")
-    DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
-    DEBUG = os.getenv("DEBUG", "false").lower() in ("true", "1", "t")
+from pydantic import AliasChoices, Field, SecretStr, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict

-    # FastGPT Configuration
-    FASTGPT_API_KEY = os.getenv("ANALYSIS_AUTH_TOKEN")
-    FASTGPT_BASE_URL = os.getenv("ANALYSIS_SERVICE_URL")
-    FASTGPT_APP_ID = os.getenv("APP_ID")
-    DELETE_SESSION_URL = os.getenv("DELETE_SESSION_URL")
-    DELETE_CHAT_URL = os.getenv("DELETE_CHAT_URL")
-    GET_CHAT_RECORDS_URL = os.getenv("GET_CHAT_RECORDS_URL")

-    @classmethod
-    def validate(cls):
-        """Validate required configuration."""
-        required = [
-            ("FASTGPT_API_KEY", cls.FASTGPT_API_KEY),
-            ("FASTGPT_APP_ID", cls.FASTGPT_APP_ID),
-        ]
-        missing = [name for name, value in required if not value]
-        if missing:
-            raise ValueError(f"Missing required environment variables: {', '.join(missing)}")
+Environment = Literal["development", "test", "staging", "production"]
+AgentBackendName = Literal["fastgpt", "langgraph"]
+CheckpointerName = Literal["memory", "postgres"]
+
+
+class Settings(BaseSettings):
+    """Runtime configuration with legacy FastGPT environment aliases."""
+
+    model_config = SettingsConfigDict(
+        env_file=(".env", ".env.local"),
+        env_file_encoding="utf-8",
+        extra="ignore",
+        populate_by_name=True,
+    )
+
+    environment: Environment = Field(
+        default="development",
+        validation_alias=AliasChoices("ZNJJ_ENVIRONMENT", "ENVIRONMENT"),
+    )
+    agent_backend: AgentBackendName = Field(
+        default="fastgpt",
+        validation_alias=AliasChoices("AGENT_BACKEND", "ZNJJ_AGENT_BACKEND"),
+    )
+    langgraph_checkpointer: CheckpointerName = Field(
+        default="memory",
+        validation_alias=AliasChoices(
+            "LANGGRAPH_CHECKPOINTER",
+            "ZNJJ_LANGGRAPH_CHECKPOINTER",
+        ),
+    )
+    debug: bool = Field(default=False, validation_alias="DEBUG")
+
+    fastgpt_api_key: SecretStr | None = Field(
+        default=None,
+        validation_alias=AliasChoices(
+            "ANALYSIS_AUTH_TOKEN",
+            "ZNJJ_FASTGPT_API_KEY",
+        ),
+    )
+    fastgpt_base_url: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices(
+            "ANALYSIS_SERVICE_URL",
+            "ZNJJ_FASTGPT_BASE_URL",
+        ),
+    )
+    fastgpt_app_id: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices("APP_ID", "ZNJJ_FASTGPT_APP_ID"),
+    )
+
+    llm_api_key: SecretStr | None = Field(
+        default=None,
+        validation_alias=AliasChoices("LLM_API_KEY", "ZNJJ_LLM_API_KEY"),
+    )
+    llm_base_url: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices("LLM_BASE_URL", "ZNJJ_LLM_BASE_URL"),
+    )
+    llm_model: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices("LLM_MODEL", "ZNJJ_LLM_MODEL"),
+    )
+    llm_timeout_seconds: float = Field(
+        default=60.0,
+        gt=0,
+        validation_alias=AliasChoices(
+            "LLM_TIMEOUT_SECONDS",
+            "ZNJJ_LLM_TIMEOUT_SECONDS",
+        ),
+    )
+    llm_max_retries: int = Field(
+        default=2,
+        ge=0,
+        le=5,
+        validation_alias=AliasChoices(
+            "LLM_MAX_RETRIES",
+            "ZNJJ_LLM_MAX_RETRIES",
+        ),
+    )
+
+    database_dsn: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices("DATABASE_URL", "ZNJJ_DATABASE_DSN"),
+    )
+    secret_key: SecretStr | None = Field(
+        default=None,
+        validation_alias=AliasChoices("SECRET_KEY", "ZNJJ_SECRET_KEY"),
+    )
+
+    @property
+    def has_fastgpt_config(self) -> bool:
+        return bool(
+            self.fastgpt_api_key
+            and self.fastgpt_base_url
+            and self.fastgpt_app_id
+        )
+
+    @model_validator(mode="after")
+    def validate_selected_backend(self) -> "Settings":
+        if self.agent_backend == "fastgpt" and not self.has_fastgpt_config:
+            raise ValueError(
+                "FastGPT backend requires ANALYSIS_AUTH_TOKEN, "
+                "ANALYSIS_SERVICE_URL and APP_ID"
+            )
+
+        if self.agent_backend == "langgraph":
+            missing = []
+            if not self.llm_api_key:
+                missing.append("LLM_API_KEY")
+            if not self.llm_model:
+                missing.append("LLM_MODEL")
+            if missing:
+                raise ValueError(
+                    "LangGraph backend requires " + ", ".join(missing)
+                )
+            if self.langgraph_checkpointer != "memory":
+                raise ValueError(
+                    "Only the memory checkpointer is implemented in the "
+                    "minimal LangGraph slice"
+                )
+
+        if (
+            self.environment in {"staging", "production"}
+            and self.agent_backend == "langgraph"
+            and self.langgraph_checkpointer == "memory"
+        ):
+            raise ValueError(
+                "Staging and production LangGraph cannot use the memory "
+                "checkpointer"
+            )
+        return self
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()
+
+
+def secret_value(value: SecretStr | None) -> str | None:
+    """Return a secret only at the integration boundary."""
+
+    return value.get_secret_value() if value is not None else None
--- a/src/core/fastgpt_client.py
+++ b/src/core/fastgpt_client.py
@@ -1,39 +1,115 @@
-"""FastGPT client dependency injection."""
+"""Backend selection and resource lifecycle.
+
+The module name is kept for compatibility with existing endpoint imports.
+"""
+
 from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from fastgpt_client import AsyncChatClient
-from .config import Config

-# Global client instance
+from ..agent.graph import build_accident_graph
+from ..agent.state import ResponseGenerator
+from ..backends.chat import ChatBackend
+from ..backends.fastgpt import FastGPTBackend
+from ..backends.langgraph import LangGraphBackend
+from ..services.llm import OpenAIResponseGenerator
+from .config import Settings, get_settings, secret_value
+
+
 _fastgpt_client: AsyncChatClient | None = None
+_chat_backend: ChatBackend | None = None
+
+
+def create_chat_backend(
+    settings: Settings,
+    *,
+    fastgpt_client: AsyncChatClient | None = None,
+    response_generator: ResponseGenerator | None = None,
+) -> ChatBackend:
+    """Create the selected backend without changing the public API boundary."""
+
+    if settings.agent_backend == "fastgpt":
+        if fastgpt_client is None:
+            raise RuntimeError("FastGPT client is required for fastgpt backend")
+        return FastGPTBackend(fastgpt_client)
+
+    generator = response_generator
+    if generator is None:
+        api_key = secret_value(settings.llm_api_key)
+        if not api_key or not settings.llm_model:
+            raise RuntimeError("Validated LangGraph LLM settings are missing")
+        generator = OpenAIResponseGenerator(
+            api_key=api_key,
+            base_url=settings.llm_base_url,
+            model=settings.llm_model,
+            timeout=settings.llm_timeout_seconds,
+            max_retries=settings.llm_max_retries,
+        )
+
+    graph = build_accident_graph(generator)
+    return LangGraphBackend(graph, generator)
+
+
+def _create_fastgpt_client(settings: Settings) -> AsyncChatClient:
+    api_key = secret_value(settings.fastgpt_api_key)
+    if not api_key or not settings.fastgpt_base_url:
+        raise RuntimeError("Validated FastGPT settings are missing")
+    return AsyncChatClient(
+        api_key=api_key,
+        base_url=settings.fastgpt_base_url,
+        timeout=60.0,
+        max_retries=3,
+        retry_delay=1.0,
+        enable_logging=settings.debug,
+    )


@asynccontextmanager
 async def lifespan(app: FastAPI):
-    """Manage FastGPT client lifecycle."""
-    global _fastgpt_client
-    Config.validate()
+    """Initialize the selected chat backend and optional legacy info client."""

-    # Initialize client
-    _fastgpt_client = AsyncChatClient(
-        api_key=Config.FASTGPT_API_KEY,
-        base_url=Config.FASTGPT_BASE_URL,
-        timeout=60.0,
-        max_retries=3,
-        retry_delay=1.0,
-        enable_logging=Config.DEBUG,
+    global _chat_backend, _fastgpt_client
+    settings = get_settings()
+
+    if settings.has_fastgpt_config:
+        _fastgpt_client = _create_fastgpt_client(settings)
+        await _fastgpt_client.__aenter__()
+
+    _chat_backend = create_chat_backend(
+        settings,
+        fastgpt_client=_fastgpt_client,
    )
-    await _fastgpt_client.__aenter__()
+    app.state.settings = settings
+    app.state.chat_backend = _chat_backend
+    app.state.fastgpt_client = _fastgpt_client

-    yield
+    try:
+        yield
+    finally:
+        backend = _chat_backend
+        _chat_backend = None
+        if isinstance(backend, LangGraphBackend):
+            await backend.aclose()

-    # Cleanup
-    if _fastgpt_client:
-        await _fastgpt_client.__aexit__(None, None, None)
+        client = _fastgpt_client
+        _fastgpt_client = None
+        if client is not None:
+            await client.__aexit__(None, None, None)


 def get_fastgpt_client() -> AsyncChatClient:
-    """Get the FastGPT client instance."""
+    """Return the legacy client used by set/get/delete until Phase 4."""
+
    if _fastgpt_client is None:
-        raise RuntimeError("FastGPT client not initialized")
+        raise RuntimeError(
+            "FastGPT client is not initialized; /set_info, /get_info and "
+            "/delete_session still require legacy FastGPT configuration"
+        )
    return _fastgpt_client
+
+
+def get_chat_backend() -> ChatBackend:
+    if _chat_backend is None:
+        raise RuntimeError("Chat backend not initialized")
+    return _chat_backend
--- a/src/core/logging_config.py
+++ b/src/core/logging_config.py
@@ -4,7 +4,9 @@ import logging
 import inspect
 from loguru import logger
 from pathlib import Path
-from .config import Config
+from .config import get_settings
+
+settings = get_settings()

 # Remove default logger
 logger.remove()
@@ -13,7 +15,7 @@ logger.remove()
 logger.add(
    sys.stderr,
    format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
-    level="DEBUG" if Config.DEBUG else "INFO",
+    level="DEBUG" if settings.debug else "INFO",
    colorize=True,
 )

--- a/src/services/init.py
+++ b/src/services/init.py
@@ -0,0 +1,5 @@
+"""Application services."""
+
+from .llm import OpenAIResponseGenerator
+
+__all__ = ["OpenAIResponseGenerator"]
--- a/src/services/llm.py
+++ b/src/services/llm.py
@@ -0,0 +1,57 @@
+"""Minimal OpenAI-compatible response generator for LangGraph."""
+
+from openai import AsyncOpenAI
+
+from ..agent.state import AccidentGraphState, GeneratedTurn
+
+
+SYSTEM_PROMPT = """你是无锡交警事故信息采集助手。
+每次回复必须严格以且仅以一个 <state>四位数字</state> 开头，随后输出用户可见正文。
+当前最小流程尚未接入完整状态机；除非用户明确要求转人工，否则使用 1002。
+用户明确要求转人工时使用 0001。
+不要输出 JSON、Markdown、分析过程或重复的 state 标签。"""
+
+
+class OpenAIResponseGenerator:
+    """Generate one response through an OpenAI-compatible chat endpoint."""
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        model: str,
+        base_url: str | None = None,
+        timeout: float = 60.0,
+        max_retries: int = 2,
+    ):
+        self._model = model
+        self._client = AsyncOpenAI(
+            api_key=api_key,
+            base_url=base_url,
+            timeout=timeout,
+            max_retries=max_retries,
+        )
+
+    async def generate(self, state: AccidentGraphState) -> GeneratedTurn:
+        response = await self._client.chat.completions.create(
+            model=self._model,
+            messages=[
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {
+                    "role": "system",
+                    "content": (
+                        "当前内部阶段："
+                        f"{state.get('stage_code', '1002')}；"
+                        f"当前为第 {state.get('turn_count', 0) + 1} 轮。"
+                    ),
+                },
+                {"role": "user", "content": state["input_text"]},
+            ],
+        )
+        content = response.choices[0].message.content
+        if not isinstance(content, str) or not content:
+            raise ValueError("LLM returned empty response content")
+        return GeneratedTurn(content=content)
+
+    async def aclose(self) -> None:
+        await self._client.close()
--- a/test/agent/test_langgraph_backend.py
+++ b/test/agent/test_langgraph_backend.py
@@ -0,0 +1,121 @@
+from copy import deepcopy
+
+import pytest
+
+from src.agent.state import AccidentGraphState, GeneratedTurn
+from src.backends.chat import ChatInput, FormUpdate, TextDelta
+from src.backends.fastgpt import FastGPTBackend
+from src.backends.langgraph import LangGraphBackend
+from src.core.config import Settings
+from src.core.fastgpt_client import create_chat_backend
+
+
+class FakeResponseGenerator:
+    def __init__(self):
+        self.states = []
+        self.closed = False
+
+    async def generate(self, state: AccidentGraphState) -> GeneratedTurn:
+        self.states.append(deepcopy(state))
+        turn_number = state.get("turn_count", 0) + 1
+        patch = {"turn": turn_number} if state["need_form_update"] else {}
+        return GeneratedTurn(
+            content=f"<state>1002</state>第{turn_number}轮",
+            form_update=patch,
+        )
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+def langgraph_settings():
+    return Settings(
+        _env_file=None,
+        environment="test",
+        agent_backend="langgraph",
+        langgraph_checkpointer="memory",
+        llm_api_key="test-key",
+        llm_model="test-model",
+    )
+
+
+@pytest.mark.asyncio
+async def test_langgraph_backend_preserves_thread_scoped_turn_state():
+    generator = FakeResponseGenerator()
+    backend = create_chat_backend(
+        langgraph_settings(),
+        response_generator=generator,
+    )
+    assert isinstance(backend, LangGraphBackend)
+
+    first = await backend.complete(
+        ChatInput("session-1", "第一轮", need_form_update=True)
+    )
+    second = await backend.complete(
+        ChatInput("session-1", "第二轮", need_form_update=True)
+    )
+    other_session = await backend.complete(
+        ChatInput("session-2", "独立会话", need_form_update=True)
+    )
+
+    assert first.content == "<state>1002</state>第1轮"
+    assert first.form_update == {"turn": 1}
+    assert second.content == "<state>1002</state>第2轮"
+    assert second.form_update == {"turn": 2}
+    assert other_session.content == "<state>1002</state>第1轮"
+    assert generator.states[0].get("turn_count", 0) == 0
+    assert generator.states[1]["turn_count"] == 1
+    assert generator.states[2].get("turn_count", 0) == 0
+
+
+@pytest.mark.asyncio
+async def test_langgraph_stream_bridges_form_update_before_text():
+    generator = FakeResponseGenerator()
+    backend = create_chat_backend(
+        langgraph_settings(),
+        response_generator=generator,
+    )
+
+    events = [
+        event
+        async for event in backend.stream(
+            ChatInput("session-stream", "开始", need_form_update=True)
+        )
+    ]
+
+    assert events == [
+        FormUpdate({"turn": 1}),
+        TextDelta("<state>1002</state>第1轮"),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_langgraph_backend_closes_owned_generator():
+    generator = FakeResponseGenerator()
+    backend = create_chat_backend(
+        langgraph_settings(),
+        response_generator=generator,
+    )
+
+    await backend.aclose()
+
+    assert generator.closed
+
+
+def test_factory_keeps_fastgpt_as_default_compatible_backend():
+    settings = Settings(
+        _env_file=None,
+        environment="test",
+        agent_backend="fastgpt",
+        fastgpt_api_key="test-key",
+        fastgpt_base_url="http://fastgpt.test",
+        fastgpt_app_id="test-app",
+    )
+    fake_client = object()
+
+    backend = create_chat_backend(
+        settings,
+        fastgpt_client=fake_client,
+    )
+
+    assert isinstance(backend, FastGPTBackend)
--- a/test/api/fastgpt.http
+++ b/test/api/fastgpt.http
@@ -1,8 +1,11 @@
-GET http://101.89.151.141:3000
+@fastgptBaseUrl = http://127.0.0.1:3000
+@fastgptApiKey = replace-with-local-api-key
+
+GET {{fastgptBaseUrl}}
 ###
-POST http://101.89.151.141:3000/api/v1/chat/completions
+POST {{fastgptBaseUrl}}/api/v1/chat/completions
 content-type: application/json
-Authorization: Bearer fastgpt-xCH4CaEoNEyVtq7fkBEI5UP3O6sABKdpGszTtSYk4R2TVW5VgrPp1YPfuLX1iH
+Authorization: Bearer {{fastgptApiKey}}

 {

--- a/test/api/test_chat_backend_boundary.py
+++ b/test/api/test_chat_backend_boundary.py
@@ -0,0 +1,70 @@
+import pytest
+
+from src.api.endpoints import chat
+from src.backends.chat import ChatInput, ChatResult, FormUpdate, TextDelta
+from src.schemas.models import ProcessRequest_chat
+
+
+class FakeBackend:
+    def __init__(self):
+        self.received = []
+
+    async def stream(self, chat_input: ChatInput):
+        self.received.append(chat_input)
+        yield TextDelta("<sta")
+        yield TextDelta("te>1002</state>你")
+        yield FormUpdate({"jdcsl": 2})
+        yield TextDelta("好")
+
+    async def complete(self, chat_input: ChatInput):
+        self.received.append(chat_input)
+        return ChatResult(
+            content="<state>1002</state>你好",
+            status_code="1002",
+            form_update={"jdcsl": 2},
+        )
+
+
+def make_request():
+    return ProcessRequest_chat(
+        sessionId="session-001",
+        timeStamp="20260725120000",
+        text="发生了交通事故",
+        needFormUpdate=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_non_stream_chat_uses_backend_neutral_result():
+    backend = FakeBackend()
+
+    response = await chat(make_request(), stream=False, backend=backend)
+
+    assert response.outputText == "你好"
+    assert response.nextStageCode == "1002"
+    assert response.formUpdate == {"jdcsl": 2}
+    assert backend.received == [
+        ChatInput(
+            session_id="session-001",
+            text="发生了交通事故",
+            need_form_update=True,
+        )
+    ]
+
+
+@pytest.mark.asyncio
+async def test_stream_chat_keeps_state_prefix_buffering_in_fastapi_layer():
+    backend = FakeBackend()
+
+    response = await chat(make_request(), stream=True, backend=backend)
+    chunks = []
+    async for chunk in response.body_iterator:
+        chunks.append(chunk.decode() if isinstance(chunk, bytes) else chunk)
+    body = "".join(chunks)
+
+    assert body.index("event: stage_code") < body.index("event: text_delta")
+    assert '"nextStageCode": "1002"' in body
+    assert '"text": "你"' in body
+    assert '"text": "好"' in body
+    assert "event: formUpdate" in body
+    assert "event: done" in body
--- a/test/api/test_chat_sse_contract.py
+++ b/test/api/test_chat_sse_contract.py
@@ -0,0 +1,116 @@
+import json
+
+import pytest
+
+from src.api.endpoints import chat
+from src.backends.chat import ChatInput, ChatResult, FormUpdate, TextDelta
+from src.schemas.models import ProcessRequest_chat
+
+
+def make_request(**overrides):
+    payload = {
+        "sessionId": "session-001",
+        "timeStamp": "20260726120000",
+        "text": "发生了交通事故",
+        "needFormUpdate": True,
+    }
+    payload.update(overrides)
+    return ProcessRequest_chat(**payload)
+
+
+async def response_text(response):
+    chunks = []
+    async for chunk in response.body_iterator:
+        chunks.append(chunk.decode() if isinstance(chunk, bytes) else chunk)
+    return "".join(chunks)
+
+
+def parse_sse(body):
+    events = []
+    for block in body.strip().split("\n\n"):
+        lines = block.splitlines()
+        event = lines[0].removeprefix("event: ")
+        data = json.loads(lines[1].removeprefix("data: "))
+        events.append((event, data))
+    return events
+
+
+class OrderedBackend:
+    async def stream(self, chat_input: ChatInput):
+        yield TextDelta("<sta")
+        yield TextDelta("te>1002</state>")
+        yield FormUpdate({"jdcsl": 2})
+        yield TextDelta("第一句。")
+        yield TextDelta("第二句。")
+
+    async def complete(self, chat_input: ChatInput):
+        return ChatResult("<state>1002</state>第一句。第二句。", "1002", {"jdcsl": 2})
+
+
+class MissingPrefixBackend:
+    async def stream(self, chat_input: ChatInput):
+        yield TextDelta("没有状态前缀")
+
+    async def complete(self, chat_input: ChatInput):
+        return ChatResult("没有状态前缀")
+
+
+@pytest.mark.asyncio
+async def test_sse_success_event_order_and_cardinality():
+    response = await chat(make_request(), stream=True, backend=OrderedBackend())
+    events = parse_sse(await response_text(response))
+    names = [name for name, _ in events]
+
+    assert names == [
+        "stage_code",
+        "formUpdate",
+        "text_delta",
+        "text_delta",
+        "done",
+    ]
+    assert names.count("stage_code") == 1
+    assert names.count("formUpdate") == 1
+    assert names.count("done") == 1
+    assert "error" not in names
+    assert "".join(data["text"] for name, data in events if name == "text_delta") == (
+        "第一句。第二句。"
+    )
+
+
+@pytest.mark.asyncio
+async def test_use_text_chunk_only_changes_delta_boundaries():
+    request = make_request(useTextChunk=True)
+    response = await chat(request, stream=True, backend=OrderedBackend())
+    events = parse_sse(await response_text(response))
+
+    assert "".join(data["text"] for name, data in events if name == "text_delta") == (
+        "第一句。第二句。"
+    )
+    assert [name for name, _ in events].count("done") == 1
+
+
+@pytest.mark.asyncio
+async def test_missing_stream_prefix_characterizes_current_legacy_behavior():
+    response = await chat(
+        make_request(needFormUpdate=False),
+        stream=True,
+        backend=MissingPrefixBackend(),
+    )
+    events = parse_sse(await response_text(response))
+
+    assert [name for name, _ in events] == ["text_delta", "done"]
+    assert events[0][1]["text"] == "没有状态前缀"
+
+
+@pytest.mark.asyncio
+async def test_missing_non_stream_prefix_returns_compatible_business_error():
+    response = await chat(
+        make_request(needFormUpdate=False),
+        stream=False,
+        backend=MissingPrefixBackend(),
+    )
+
+    assert response.code == "500"
+    assert response.outputText == ""
+    assert response.nextStageCode == ""
+    assert response.msg == "大模型服务返回消息不完整"
--- a/test/api/test_info_contract.py
+++ b/test/api/test_info_contract.py
@@ -0,0 +1,145 @@
+import json
+
+import pytest
+
+from src.api import endpoints
+from src.api.endpoints import get_info, set_info
+from src.schemas.models import ProcessRequest_get, ProcessRequest_set
+
+
+class FakeResponse:
+    def __init__(self, payload):
+        self._payload = payload
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return self._payload
+
+
+class FakeInfoClient:
+    def __init__(self, state):
+        self.state = state
+        self.completion_calls = []
+
+    async def create_chat_completion(self, **kwargs):
+        self.completion_calls.append(kwargs)
+        if "variables" in kwargs:
+            self.state = kwargs["variables"]["state"]
+        return FakeResponse({"newVariables": {"state": self.state}})
+
+
+@pytest.fixture
+def skip_helper_record_deletion(monkeypatch):
+    calls = []
+
+    async def fake_delete(client, session_id):
+        calls.append(session_id)
+
+    monkeypatch.setattr(endpoints, "delete_last_two_chat_records", fake_delete)
+    return calls
+
+
+def make_set_request(**overrides):
+    payload = {
+        "sessionId": "session-001",
+        "timeStamp": "20260726120000",
+        "key": "hphm1",
+        "value": "<PLATE_1>",
+    }
+    payload.update(overrides)
+    return ProcessRequest_set(**payload)
+
+
+def make_get_request(**overrides):
+    payload = {
+        "sessionId": "session-001",
+        "timeStamp": "20260726120000",
+        "key": "all",
+    }
+    payload.update(overrides)
+    return ProcessRequest_get(**payload)
+
+
+@pytest.mark.asyncio
+async def test_set_info_reads_then_writes_fastgpt_state(
+    skip_helper_record_deletion,
+):
+    client = FakeInfoClient({"hphm1": "<PLATE_OLD>", "jdcsl": 1})
+
+    response = await set_info(make_set_request(), client=client)
+
+    assert response.code == "200"
+    assert client.state == {"hphm1": "<PLATE_1>", "jdcsl": 1}
+    assert len(client.completion_calls) == 2
+    assert client.completion_calls[1]["variables"]["state"] == client.state
+    assert skip_helper_record_deletion == ["session-001", "session-001"]
+
+
+@pytest.mark.asyncio
+async def test_set_info_include_input_info_keeps_legacy_magic_payload(
+    skip_helper_record_deletion,
+):
+    client = FakeInfoClient({})
+
+    await set_info(make_set_request(includeInputInfo=True), client=client)
+
+    message = client.completion_calls[0]["messages"][0]["content"]
+    assert message == '<setInfo>{"key": "hphm1", "value": "<PLATE_1>"}</setInfo>'
+
+
+@pytest.mark.asyncio
+async def test_get_info_all_keeps_json_string_and_boolean_encoding(
+    skip_helper_record_deletion,
+):
+    client = FakeInfoClient(
+        {
+            "ywrysw": False,
+            "ywfjdc": True,
+            "jdcsl": 2,
+            "xm1": "<PERSON_1>",
+            "hphm1": "<PLATE_1>",
+            "xm2": "<PERSON_2>",
+        }
+    )
+
+    response = await get_info(make_get_request(), client=client)
+    value = json.loads(response.value)
+
+    assert response.code == "200"
+    assert isinstance(response.value, str)
+    assert value["acdinfo"]["ywrysw"] == "0"
+    assert value["acdinfo"]["ywfjdc"] == "1"
+    assert value["acdinfo"]["jdcsl"] == 2
+    assert value["acdhuman1"]["xm1"] == "<PERSON_1>"
+    assert value["acdhuman2"]["xm2"] == "<PERSON_2>"
+    assert skip_helper_record_deletion == ["session-001"]
+
+
+@pytest.mark.asyncio
+async def test_get_info_unknown_key_returns_json_encoded_empty_string(
+    skip_helper_record_deletion,
+):
+    client = FakeInfoClient({})
+
+    response = await get_info(
+        make_get_request(key="unknown_legacy_key"),
+        client=client,
+    )
+
+    assert response.code == "200"
+    assert response.value == '""'
+
+
+@pytest.mark.asyncio
+async def test_get_info_fastgpt_shape_error_keeps_legacy_business_error():
+    class InvalidClient:
+        async def create_chat_completion(self, **kwargs):
+            return FakeResponse({})
+
+    response = await get_info(make_get_request(), client=InvalidClient())
+
+    assert response.code == "500"
+    assert response.value == ""
+    assert response.msg == "大模型服务器无响应"
--- a/test/api/test_public_schema_contract.py
+++ b/test/api/test_public_schema_contract.py
@@ -0,0 +1,195 @@
+"""Characterization tests for the public HTTP API schemas.
+
+These tests freeze the current contract before the FastGPT backend is replaced.
+They should change only when the teams integrating with this service agree to a
+contract change.
+"""
+
+from collections.abc import Callable
+from typing import Any
+
+import pytest
+from pydantic import BaseModel, ValidationError
+
+from src.schemas.models import (
+    ProcessRequest_chat,
+    ProcessRequest_get,
+    ProcessRequest_set,
+    ProcessResponse_chat,
+    ProcessResponse_get,
+    ProcessResponse_set,
+)
+
+
+SESSION_ID = "session-001"
+TIMESTAMP = "20260725120000"
+
+
+def chat_request_payload() -> dict[str, Any]:
+    return {
+        "sessionId": SESSION_ID,
+        "timeStamp": TIMESTAMP,
+        "text": "发生了交通事故",
+    }
+
+
+def chat_response_payload() -> dict[str, Any]:
+    return {
+        "sessionId": SESSION_ID,
+        "timeStamp": TIMESTAMP,
+        "outputText": "请描述事故经过。",
+        "nextStage": "通话中",
+        "nextStageCode": "1002",
+        "code": "200",
+    }
+
+
+def get_request_payload() -> dict[str, Any]:
+    return {
+        "sessionId": SESSION_ID,
+        "timeStamp": TIMESTAMP,
+        "key": "acdinfo",
+    }
+
+
+def get_response_payload() -> dict[str, Any]:
+    return {
+        "sessionId": SESSION_ID,
+        "timeStamp": TIMESTAMP,
+        "value": '{"jdcsl": "2"}',
+        "code": "200",
+    }
+
+
+def set_request_payload() -> dict[str, Any]:
+    return {
+        "sessionId": SESSION_ID,
+        "timeStamp": TIMESTAMP,
+        "key": "hphm1",
+        "value": "沪A12345",
+    }
+
+
+def set_response_payload() -> dict[str, Any]:
+    return {
+        "sessionId": SESSION_ID,
+        "timeStamp": TIMESTAMP,
+        "code": "200",
+    }
+
+
+def test_chat_request_defaults_are_backward_compatible() -> None:
+    request = ProcessRequest_chat(**chat_request_payload())
+
+    assert request.model_dump() == {
+        **chat_request_payload(),
+        "needFormUpdate": False,
+        "useTextChunk": False,
+    }
+
+
+def test_chat_response_shape_is_backward_compatible() -> None:
+    response = ProcessResponse_chat(**chat_response_payload())
+
+    assert response.model_dump() == {
+        **chat_response_payload(),
+        "formUpdate": {},
+        "msg": None,
+    }
+
+
+def test_get_info_shapes_are_backward_compatible() -> None:
+    request = ProcessRequest_get(**get_request_payload())
+    response = ProcessResponse_get(**get_response_payload())
+
+    assert request.model_dump() == {
+        **get_request_payload(),
+        "includeInputInfo": False,
+    }
+    assert response.model_dump() == {
+        **get_response_payload(),
+        "msg": None,
+    }
+
+
+def test_set_info_shapes_are_backward_compatible() -> None:
+    request = ProcessRequest_set(**set_request_payload())
+    response = ProcessResponse_set(**set_response_payload())
+
+    assert request.model_dump() == {
+        **set_request_payload(),
+        "includeInputInfo": False,
+    }
+    assert response.model_dump() == {
+        **set_response_payload(),
+        "msg": None,
+    }
+
+
+@pytest.mark.parametrize(
+    ("model", "payload_factory", "required_field"),
+    [
+        (ProcessRequest_chat, chat_request_payload, "sessionId"),
+        (ProcessRequest_chat, chat_request_payload, "timeStamp"),
+        (ProcessRequest_chat, chat_request_payload, "text"),
+        (ProcessRequest_get, get_request_payload, "key"),
+        (ProcessRequest_set, set_request_payload, "key"),
+        (ProcessRequest_set, set_request_payload, "value"),
+        (ProcessResponse_chat, chat_response_payload, "nextStageCode"),
+        (ProcessResponse_get, get_response_payload, "value"),
+    ],
+)
+def test_required_fields_remain_required(
+    model: type[BaseModel],
+    payload_factory: Callable[[], dict[str, Any]],
+    required_field: str,
+) -> None:
+    payload = payload_factory()
+    payload.pop(required_field)
+
+    with pytest.raises(ValidationError):
+        model(**payload)
+
+
+@pytest.mark.parametrize(
+    ("model", "payload_factory", "field_name", "invalid_value"),
+    [
+        (ProcessRequest_chat, chat_request_payload, "sessionId", "s" * 65),
+        (ProcessRequest_chat, chat_request_payload, "timeStamp", "t" * 33),
+        (ProcessRequest_get, get_request_payload, "sessionId", "s" * 65),
+        (ProcessRequest_set, set_request_payload, "timeStamp", "t" * 33),
+        (ProcessResponse_chat, chat_response_payload, "nextStage", "n" * 33),
+        (ProcessResponse_chat, chat_response_payload, "nextStageCode", "10020"),
+        (ProcessResponse_chat, chat_response_payload, "code", "10000"),
+        (ProcessResponse_get, get_response_payload, "code", "10000"),
+        (ProcessResponse_set, set_response_payload, "code", "10000"),
+    ],
+)
+def test_public_length_limits_are_enforced(
+    model: type[BaseModel],
+    payload_factory: Callable[[], dict[str, Any]],
+    field_name: str,
+    invalid_value: str,
+) -> None:
+    payload = payload_factory()
+    payload[field_name] = invalid_value
+
+    with pytest.raises(ValidationError):
+        model(**payload)
+
+
+def test_get_info_value_remains_a_json_encoded_string() -> None:
+    response = ProcessResponse_get(**get_response_payload())
+
+    assert isinstance(response.value, str)
+    assert response.value == '{"jdcsl": "2"}'
+
+
+def test_chat_form_update_remains_unstructured_for_compatibility() -> None:
+    form_update = {"jdcsl": 2, "ywrysw": False}
+    response = ProcessResponse_chat(
+        **chat_response_payload(),
+        formUpdate=form_update,
+    )
+
+    assert response.formUpdate == form_update
--- a/test/backends/test_fastgpt_backend.py
+++ b/test/backends/test_fastgpt_backend.py
@@ -0,0 +1,123 @@
+from types import SimpleNamespace
+
+import pytest
+
+from src.backends.chat import ChatInput, FormUpdate, TextDelta
+from src.backends.fastgpt import FastGPTBackend
+
+
+class FakeResponse:
+    def __init__(self, data):
+        self._data = data
+        self.raise_for_status_called = False
+
+    def raise_for_status(self):
+        self.raise_for_status_called = True
+
+    def json(self):
+        return self._data
+
+
+class FakeClient:
+    def __init__(self, response):
+        self.response = response
+        self.calls = []
+
+    async def create_chat_completion(self, **kwargs):
+        self.calls.append(kwargs)
+        return self.response
+
+
+@pytest.mark.asyncio
+async def test_complete_translates_fastgpt_response_to_neutral_result():
+    response = FakeResponse(
+        {
+            "choices": [
+                {
+                    "message": {
+                        "content": "<state>1002</state>请描述事故经过。"
+                    }
+                }
+            ],
+            "newVariables": {"status_code": "1002"},
+            "responseData": [
+                {
+                    "moduleName": "文本内容提取事故信息",
+                    "extractResult": {
+                        "formUpdate": '{"jdcsl": 2, "ywrysw": false}'
+                    },
+                }
+            ],
+        }
+    )
+    client = FakeClient(response)
+    backend = FastGPTBackend(client)
+
+    result = await backend.complete(
+        ChatInput(
+            session_id="session-001",
+            text="发生了交通事故",
+            need_form_update=True,
+        )
+    )
+
+    assert response.raise_for_status_called is True
+    assert result.content == "<state>1002</state>请描述事故经过。"
+    assert result.status_code == "1002"
+    assert result.form_update == {"jdcsl": 2, "ywrysw": False}
+    assert client.calls == [
+        {
+            "messages": [{"role": "user", "content": "发生了交通事故"}],
+            "chatId": "session-001",
+            "stream": False,
+            "detail": True,
+            "variables": {"needFormUpdate": True},
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_stream_translates_fastgpt_events_to_neutral_events(monkeypatch):
+    response = object()
+    client = FakeClient(response)
+    backend = FastGPTBackend(client)
+
+    async def fake_aiter_stream_events(actual_response):
+        assert actual_response is response
+        yield SimpleNamespace(
+            kind="answer",
+            data={"choices": [{"delta": {"content": "<state>1002"}}]},
+        )
+        yield SimpleNamespace(
+            kind="flowResponses",
+            data=[
+                {
+                    "moduleName": "文本内容提取事故信息",
+                    "extractResult": {"formUpdate": '{"jdcsl": 2}'},
+                }
+            ],
+        )
+        yield SimpleNamespace(
+            kind="answer",
+            data={"choices": [{"delta": {"content": "</state>你好"}}]},
+        )
+        yield SimpleNamespace(kind="ignored", data={})
+
+    monkeypatch.setattr(
+        "src.backends.fastgpt.aiter_stream_events",
+        fake_aiter_stream_events,
+    )
+
+    events = [
+        event
+        async for event in backend.stream(
+            ChatInput(session_id="session-001", text="你好")
+        )
+    ]
+
+    assert events == [
+        TextDelta("<state>1002"),
+        FormUpdate({"jdcsl": 2}),
+        TextDelta("</state>你好"),
+    ]
+    assert client.calls[0]["stream"] is True
--- a/test/core/test_config.py
+++ b/test/core/test_config.py
@@ -0,0 +1,78 @@
+import pytest
+from pydantic import ValidationError
+
+from src.core.config import Settings
+
+
+def test_fastgpt_backend_accepts_legacy_environment_names():
+    settings = Settings(
+        _env_file=None,
+        ZNJJ_ENVIRONMENT="test",
+        AGENT_BACKEND="fastgpt",
+        ANALYSIS_AUTH_TOKEN="test-fastgpt-key",
+        ANALYSIS_SERVICE_URL="http://fastgpt.test",
+        APP_ID="test-app",
+    )
+
+    assert settings.environment == "test"
+    assert settings.agent_backend == "fastgpt"
+    assert settings.has_fastgpt_config
+    assert settings.fastgpt_api_key.get_secret_value() == "test-fastgpt-key"
+
+
+def test_fastgpt_backend_rejects_incomplete_configuration():
+    with pytest.raises(ValidationError, match="FastGPT backend requires"):
+        Settings(
+            _env_file=None,
+            environment="test",
+            agent_backend="fastgpt",
+            fastgpt_api_key="test-key",
+        )
+
+
+def test_langgraph_backend_requires_only_minimal_llm_configuration():
+    settings = Settings(
+        _env_file=None,
+        environment="test",
+        agent_backend="langgraph",
+        langgraph_checkpointer="memory",
+        llm_api_key="test-llm-key",
+        llm_model="test-model",
+    )
+
+    assert settings.agent_backend == "langgraph"
+    assert settings.langgraph_checkpointer == "memory"
+    assert not settings.has_fastgpt_config
+
+
+def test_langgraph_backend_rejects_missing_llm_configuration():
+    with pytest.raises(ValidationError, match="LLM_API_KEY, LLM_MODEL"):
+        Settings(
+            _env_file=None,
+            environment="test",
+            agent_backend="langgraph",
+        )
+
+
+def test_production_cannot_use_memory_checkpointer():
+    with pytest.raises(ValidationError, match="cannot use the memory checkpointer"):
+        Settings(
+            _env_file=None,
+            environment="production",
+            agent_backend="langgraph",
+            langgraph_checkpointer="memory",
+            llm_api_key="test-llm-key",
+            llm_model="test-model",
+        )
+
+
+def test_secret_values_are_masked_in_settings_repr():
+    settings = Settings(
+        _env_file=None,
+        environment="test",
+        agent_backend="langgraph",
+        llm_api_key="never-print-this-key",
+        llm_model="test-model",
+    )
+
+    assert "never-print-this-key" not in repr(settings)
--- a/test/fixtures/golden/accident-scenarios.json
+++ b/test/fixtures/golden/accident-scenarios.json
@@ -0,0 +1,162 @@
+{
+  "version": "2026-07-26",
+  "description": "LangGraph 迁移的脱敏黄金场景。占位符不是可用个人信息。",
+  "scenarios": [
+    {
+      "case_id": "single_vehicle_happy_path",
+      "category": "end_to_end",
+      "initial_stage": "1001",
+      "turns": [
+        {"event": "session_started", "input": "【继续办理】", "expected_stage": "1002"},
+        {"event": "user_message", "input": "车辆倒车时碰到了固定物体", "expected_stage": "1002", "expected_patch": {"sgyy": "倒车碰到固定物体"}},
+        {"event": "user_message", "input": "没有人员受伤", "expected_stage": "1002", "expected_patch": {"ywrysw": false}},
+        {"event": "user_message", "input": "没有非机动车", "expected_stage": "1002", "expected_patch": {"ywfjdc": false}},
+        {"event": "user_message", "input": "事故时间是十分钟前", "expected_stage": "1002"},
+        {"event": "user_message", "input": "我还在现场", "expected_stage": "1002", "expected_patch": {"sfsgxc": true}},
+        {"event": "user_message", "input": "只有一辆机动车", "expected_stage": "2000", "expected_patch": {"jdcsl": 1}},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2001"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2002"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2003"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2004"},
+        {"event": "user_message", "input": "车牌正确", "expected_stage": "2005"},
+        {"event": "user_message", "input": "车损在车辆前方", "expected_stage": "3001", "expected_patch": {"csbw1": "前方"}}
+      ],
+      "expected_handoff_reason": null
+    },
+    {
+      "case_id": "double_vehicle_happy_path",
+      "category": "end_to_end",
+      "initial_stage": "1002",
+      "turns": [
+        {"event": "user_message", "input": "两辆机动车发生追尾，没有人受伤，也没有非机动车", "expected_stage": "1002", "expected_patch": {"jdcsl": 2, "ywrysw": false, "ywfjdc": false, "sgyy": "追尾"}},
+        {"event": "user_message", "input": "时间正确，我还在现场", "expected_stage": "2010", "expected_patch": {"sfsgxc": true}},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2011"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2012"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2013"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2014"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2015"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2016"},
+        {"event": "user_message", "input": "车牌正确", "expected_stage": "3002"}
+      ],
+      "expected_handoff_reason": null
+    },
+    {
+      "case_id": "explicit_handoff_global",
+      "category": "handoff",
+      "initial_stage": "2002",
+      "turns": [
+        {"event": "user_message", "input": "请帮我转人工", "expected_stage": "0001"}
+      ],
+      "expected_handoff_reason": "user_requested"
+    },
+    {
+      "case_id": "injury_handoff",
+      "category": "safety",
+      "initial_stage": "1002",
+      "turns": [
+        {"event": "user_message", "input": "有人倒地并且不舒服", "expected_stage": "0003", "expected_patch": {"ywrysw": true}}
+      ],
+      "expected_handoff_reason": "injury_or_complex"
+    },
+    {
+      "case_id": "injury_negation_does_not_handoff",
+      "category": "safety",
+      "initial_stage": "1002",
+      "turns": [
+        {"event": "user_message", "input": "不是人受伤，是车受损，人没事", "expected_stage": "1002", "expected_patch": {"ywrysw": false}}
+      ],
+      "expected_handoff_reason": null
+    },
+    {
+      "case_id": "three_vehicle_complex_handoff",
+      "category": "safety",
+      "initial_stage": "1002",
+      "turns": [
+        {"event": "user_message", "input": "一共涉及三辆机动车", "expected_stage": "0003", "expected_patch": {"jdcsl": 3}}
+      ],
+      "expected_handoff_reason": "complex_accident"
+    },
+    {
+      "case_id": "photo_failure_single",
+      "category": "deterministic_event",
+      "initial_stage": "2001",
+      "turns": [
+        {"event": "photo_recognition_failed", "input": "【客户端连续3次拍摄识别失败：图片过于模糊】", "expected_stage": "0005"}
+      ],
+      "expected_handoff_reason": "photo_recognition_failed"
+    },
+    {
+      "case_id": "photo_failure_double_confirmation",
+      "category": "deterministic_event",
+      "initial_stage": "2016",
+      "turns": [
+        {"event": "photo_recognition_failed", "input": "【客户端连续3次拍摄识别失败：未识别到完整车牌】", "expected_stage": "0005"}
+      ],
+      "expected_handoff_reason": "photo_recognition_failed"
+    },
+    {
+      "case_id": "two_consecutive_no_responses",
+      "category": "deterministic_event",
+      "initial_stage": "1002",
+      "turns": [
+        {"event": "no_response", "input": "【用户无回复】", "expected_stage": "1002", "expected_no_response_count": 1},
+        {"event": "no_response", "input": "【用户无回复】", "expected_stage": "0004", "expected_no_response_count": 2}
+      ],
+      "expected_handoff_reason": "no_response"
+    },
+    {
+      "case_id": "photo_step_cannot_skip",
+      "category": "transition_guard",
+      "initial_stage": "2011",
+      "turns": [
+        {"event": "user_message", "input": "后面的照片我都拍好了", "expected_stage": "2011"},
+        {"event": "photo_completed", "input": "【拍摄完成】", "expected_stage": "2012"}
+      ],
+      "expected_handoff_reason": null
+    },
+    {
+      "case_id": "external_update_then_chat",
+      "category": "state_consistency",
+      "initial_stage": "1002",
+      "initial_form": {"hphm1": "<PLATE_1>"},
+      "turns": [
+        {"event": "set_info", "input": {"key": "hphm1", "value": "<PLATE_CORRECTED>"}, "expected_stage": "1002", "expected_patch": {"hphm1": "<PLATE_CORRECTED>"}},
+        {"event": "user_message", "input": "请继续办理", "expected_stage": "1002", "expected_state_contains": {"hphm1": "<PLATE_CORRECTED>"}}
+      ],
+      "expected_handoff_reason": null
+    },
+    {
+      "case_id": "invalid_external_internal_field",
+      "category": "state_consistency",
+      "initial_stage": "1002",
+      "turns": [
+        {"event": "set_info", "input": {"key": "stage_code", "value": "0000"}, "expected_error": "INVALID_FIELD", "expected_stage": "1002"}
+      ],
+      "expected_handoff_reason": null
+    },
+    {
+      "case_id": "prefix_split_across_chunks",
+      "category": "prefix_parser",
+      "initial_stage": "1002",
+      "stream_chunks": ["<sta", "te>1002", "</sta", "te>请描述事故经过"],
+      "expected_stage": "1002",
+      "expected_text": "请描述事故经过"
+    },
+    {
+      "case_id": "prefix_unknown_code",
+      "category": "prefix_parser",
+      "initial_stage": "1002",
+      "model_output": "<state>9999</state>继续处理",
+      "expected_error": "UNKNOWN_STAGE_CODE",
+      "expected_stage": "1002"
+    },
+    {
+      "case_id": "prefix_illegal_photo_jump",
+      "category": "prefix_parser",
+      "initial_stage": "2000",
+      "model_output": "<state>2004</state>请确认车牌",
+      "expected_error": "ILLEGAL_STAGE_TRANSITION",
+      "expected_stage": "2000"
+    }
+  ]
+}
--- a/test/test_phase0_artifacts.py
+++ b/test/test_phase0_artifacts.py
@@ -0,0 +1,119 @@
+import json
+import re
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DOMAIN = ROOT / "docs" / "domain"
+GOLDEN = ROOT / "test" / "fixtures" / "golden" / "accident-scenarios.json"
+
+
+def load_json(path):
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def test_stage_registry_and_transition_matrix_are_closed():
+    stage_registry = load_json(DOMAIN / "stage-codes.json")
+    transitions = load_json(DOMAIN / "stage-transitions.json")
+    codes = {entry["code"] for entry in stage_registry["codes"]}
+
+    assert len(codes) == len(stage_registry["codes"])
+    assert {"0000", "0004", "0005", "1001", "1002", "3001", "3002"} <= codes
+    assert set(transitions["allowed"]) == codes
+
+    for source, targets in transitions["allowed"].items():
+        assert set(targets) <= codes, source
+
+    for terminal in {"0000", "0001", "0002", "0003", "0004", "0005"}:
+        assert transitions["allowed"][terminal] == []
+
+
+def test_photo_sequences_cannot_skip_steps():
+    transitions = load_json(DOMAIN / "stage-transitions.json")
+
+    for sequence in transitions["photo_sequences"].values():
+        for current, following in zip(sequence, sequence[1:]):
+            assert following in transitions["allowed"][current]
+            later_steps = set(sequence[sequence.index(following) + 1 :])
+            assert later_steps.isdisjoint(transitions["allowed"][current])
+
+
+def test_field_registry_groups_are_complete_and_unique():
+    registry = load_json(DOMAIN / "field-registry.json")
+    fields = registry["fields"]
+    keys = [field["key"] for field in fields]
+
+    assert len(keys) == len(set(keys))
+    assert set(keys) == {
+        key for group_keys in registry["groups"].values() for key in group_keys
+    }
+    assert {"sfzmhm1", "sjhm1", "hphm1", "sfzmhm2", "sjhm2", "hphm2"} <= {
+        field["key"] for field in fields if field["sensitive"]
+    }
+    assert {"sfzmwh1", "sjwh1", "sfzmwh2", "sjwh2"} <= {
+        field["key"] for field in fields if not field["external_write"]
+    }
+
+
+def test_golden_scenarios_are_unique_and_use_known_stages():
+    stages = {
+        entry["code"]
+        for entry in load_json(DOMAIN / "stage-codes.json")["codes"]
+    }
+    scenarios = load_json(GOLDEN)["scenarios"]
+    case_ids = [case["case_id"] for case in scenarios]
+
+    assert len(scenarios) >= 12
+    assert len(case_ids) == len(set(case_ids))
+    for case in scenarios:
+        assert case["initial_stage"] in stages
+        if "expected_stage" in case:
+            assert case["expected_stage"] in stages
+        for turn in case.get("turns", []):
+            if "expected_stage" in turn:
+                assert turn["expected_stage"] in stages
+
+
+def test_golden_turn_sequences_follow_the_transition_matrix():
+    allowed = load_json(DOMAIN / "stage-transitions.json")["allowed"]
+    scenarios = load_json(GOLDEN)["scenarios"]
+
+    for case in scenarios:
+        current = case["initial_stage"]
+        for turn in case.get("turns", []):
+            expected = turn.get("expected_stage")
+            if expected is None:
+                continue
+            if "expected_error" not in turn:
+                assert expected in allowed[current], (
+                    case["case_id"],
+                    current,
+                    expected,
+                )
+            current = expected
+
+
+def test_golden_fixture_contains_no_realistic_phone_or_national_id():
+    raw = GOLDEN.read_text(encoding="utf-8")
+
+    assert not re.search(r"(?<!\d)1[3-9]\d{9}(?!\d)", raw)
+    assert not re.search(r"(?<!\d)\d{17}[\dXx](?!\d)", raw)
+
+
+def test_repository_sources_and_examples_do_not_embed_fastgpt_tokens():
+    token_pattern = re.compile(r"fastgpt-[A-Za-z0-9]{20,}")
+    candidates = [ROOT / ".env.example"]
+
+    for directory in ("src", "test", "docs"):
+        candidates.extend(
+            path
+            for path in (ROOT / directory).rglob("*")
+            if path.is_file() and "__pycache__" not in path.parts
+        )
+
+    for path in candidates:
+        try:
+            text = path.read_text(encoding="utf-8")
+        except UnicodeDecodeError:
+            continue
+        assert not token_pattern.search(text), path
--- a/workflow/20260726/事故信息采集20260726.json
+++ b/workflow/20260726/事故信息采集20260726.json
Author	SHA1	Message	Date
Xin Wang	5263192f92	Add new workflow configuration for accident scene documentation - Introduced a new JSON file containing detailed node configurations for the accident scene documentation workflow. - Included nodes for user guidance, workflow start, and AI chat integration, with comprehensive input and output specifications. - Enhanced the structure to support various input types and configurations for improved user interaction and data handling.	2026-07-27 17:24:23 +08:00
Xin Wang	1c8e9da486	Start LangGraph backend migration foundation	2026-07-27 17:21:29 +08:00
Eric Wang	5c719ed2ea	Refactor chat backend integration and introduce new backend-neutral architecture - Replaced direct FastGPT client usage with a backend-neutral chat interface, allowing for improved flexibility and maintainability. - Introduced a new `ChatBackend` protocol and implemented `FastGPTBackend` to handle chat operations. - Updated the chat endpoint to utilize the new backend structure, enhancing the handling of chat requests and responses. - Added comprehensive tests to ensure compatibility and functionality of the new backend integration.	2026-07-26 07:25:24 +08:00
Eric Wang	893eb2f2f7	Update environment variables and add LangGraph backend migration plan - Updated ANALYSIS_AUTH_TOKEN and APP_ID in the .env file for enhanced security and configuration. - Added a comprehensive migration plan document for transitioning the LangGraph backend, detailing the background, scope, goals, and architectural decisions for the project.	2026-07-26 07:24:30 +08:00