From 97de142e6d7de2bdb3782ac3afb547cbf8ce2d3f Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Sat, 24 Oct 2026 13:41:54 +0800 Subject: [PATCH] feat(storage): add selectable SQLite and PostgreSQL persistence - Select one storage backend per instance via AGENT_STORAGE, retaining memory as the backward-compatible default. Manage the selected async repository, official checkpointer and connection pools in lifespan; fail startup on unavailable or uninitialized storage without fallback. - Add SQLite and PostgreSQL adapters, schema initialization CLI, sample PostgreSQL compose configuration and a Windows-compatible serve entry. SQLite enforces single-instance file ownership; PostgreSQL uses a dedicated pool for session advisory locks plus version/epoch fencing. - Make repository IO and commit callbacks awaitable. Persist business fields, a complete committed graph snapshot and turn result/version in one business transaction before success. Rebuild execution checkpoints from that snapshot, discarding uncommitted progress after interruption. - Support optional Idempotency-Key on durable chat requests. Replay committed results and return 409 on changed request content; delete removes business/history data and replay records while retaining only an epoch/version tombstone to reject stale commits. - Document migration, database switching boundaries, backup/deployment expectations and crash recovery. Ignore generated database files. Validation: 315 tests passed with real SQLite and isolated PostgreSQL, including abrupt child-process exits before/after business commit; both backends passed CLI migration and actual HTTP server restart smoke. Existing API tests were rechecked after isolating them from user storage. An initial Windows CreateProcess permission failure did not recur in the full diagnostic run; failure assertions were not skipped or relaxed. --- .env.example | 15 + .gitignore | 10 + README.md | 118 +++---- compose.agent-db.yml | 18 ++ docs/agent-storage.md | 141 +++++++++ docs/langgraph-backend-migration-plan.md | 13 +- requirements.txt | 4 + src/api/endpoints.py | 16 +- src/backends/chat.py | 9 +- src/backends/langgraph.py | 19 +- src/core/config.py | 19 ++ src/core/fastgpt_client.py | 104 +++--- src/serve.py | 23 ++ src/services/app_service.py | 298 ++++++++---------- src/services/repository.py | 40 +-- src/storage/__init__.py | 1 + src/storage/checkpoints.py | 16 + src/storage/database.py | 93 ++++++ src/storage/locks.py | 61 ++++ src/storage/migrate.py | 34 ++ src/storage/repository.py | 142 +++++++++ src/storage/runtime.py | 92 ++++++ test/api/test_assignment6_endpoints.py | 2 + .../services/test_delete_failure_semantics.py | 2 +- test/services/test_field_commit_and_retry.py | 4 +- test/storage/__init__.py | 1 + test/storage/crash_worker.py | 43 +++ test/storage/test_http_persistence.py | 54 ++++ test/storage/test_persistence.py | 221 +++++++++++++ test/storage/test_storage_config_and_locks.py | 88 ++++++ 30 files changed, 1368 insertions(+), 333 deletions(-) create mode 100644 compose.agent-db.yml create mode 100644 docs/agent-storage.md create mode 100644 src/serve.py create mode 100644 src/storage/__init__.py create mode 100644 src/storage/checkpoints.py create mode 100644 src/storage/database.py create mode 100644 src/storage/locks.py create mode 100644 src/storage/migrate.py create mode 100644 src/storage/repository.py create mode 100644 src/storage/runtime.py create mode 100644 test/storage/__init__.py create mode 100644 test/storage/crash_worker.py create mode 100644 test/storage/test_http_persistence.py create mode 100644 test/storage/test_persistence.py create mode 100644 test/storage/test_storage_config_and_locks.py diff --git a/.env.example b/.env.example index 96dc4e1..b554f28 100644 --- a/.env.example +++ b/.env.example @@ -9,6 +9,21 @@ APP_ID=replace-with-fastgpt-app-id # Voice demo (Pipecat /ws-product). Relative to project root, or an absolute path. VOICE_CONFIG=config/voice.json +# 会话持久化：每个应用实例只选一个，修改后重启。 +# 默认 memory；数据库不可用不会回退到 memory。 +# AGENT_STORAGE=sqlite +# AGENT_SQLITE_PATH=./data/agent.sqlite3 +# AGENT_POSTGRES_DSN=postgresql://agent:replace-me@127.0.0.1:5432/agent +# AGENT_DB_POOL_SIZE=10 +# AGENT_DB_LOCK_TIMEOUT_SECONDS=30 +# 开发环境可自动建表；默认 false，部署前运行 python -m src.storage.migrate。 +# AGENT_DB_AUTO_MIGRATE=false +# 可选本地 PostgreSQL compose 配置（密码须与 DSN 一致）： +# AGENT_POSTGRES_USER=agent +# AGENT_POSTGRES_DB=agent +# AGENT_POSTGRES_PASSWORD=replace-me +# AGENT_POSTGRES_PORT=5432 + # LangGraph chat backend（作业 2） # AGENT_BACKEND=langgraph # diff --git a/.gitignore b/.gitignore index 6636135..8eea6e2 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,16 @@ __pycache__/ logs/ *.log .env +*.db +*.db-shm +*.db-wal +*.sqlite +*.sqlite-shm +*.sqlite-wal +*.sqlite3 +*.sqlite3-shm +*.sqlite3-wal +*.instance.lock evals/results/ .agents/ .claude/ diff --git a/README.md b/README.md index f916479..5c5e623 100644 --- a/README.md +++ b/README.md @@ -1,53 +1,64 @@ # AI 事故信息采集 API（LangGraph 迁移分支） -本分支是“将对话后端从 FastGPT 工作流迁移到 LangGraph”课程的迁移准备分支。 +本分支把事故采集、拍照和信息核实从 FastGPT workflow 迁移到嵌入式 LangGraph。 +文本四接口通过 `AGENT_BACKEND` 选择 FastGPT 或 LangGraph，HTTP schema 与 SSE 事件保持兼容。 +迁移作业计划见 [docs/langgraph-backend-migration-plan.md](docs/langgraph-backend-migration-plan.md)， +字段兼容约定见 [docs/assignment6-compat.md](docs/assignment6-compat.md)。 -当前状态：对外 API 契约保持不变，`/chat` 的供应商细节已封装到后端适配层（FastGPT 仍 -为唯一实现），迁移计划见 [docs/langgraph-backend-migration-plan.md](docs/langgraph-backend-migration-plan.md)。 -本阶段交付说明与验证证据见 [docs/assignment-baseline.md](docs/assignment-baseline.md)。 +## 环境与安装 -## 环境要求 - -- Python 3.11（使用 Conda 环境 `py3.11`，见 [AGENTS.md](AGENTS.md)） -- `fastgpt-client` SDK（当前以本地可编辑方式安装，见下文） - -## 安装 +Python 3.11 使用 Conda 环境 `py3.11`（见 [AGENTS.md](AGENTS.md)）。 ```bash conda activate py3.11 - -# 项目依赖 pip install -r requirements.txt -# FastGPT SDK（本地可编辑安装；SDK 源码在独立仓库维护） +# 项目仍使用该 SDK 支持 FastGPT 路径；源码在独立仓库维护。 pip install -e /path/to/fastgpt-python-sdk ``` -## 配置 +## 配置与后端选择 -服务端通过环境变量读取 FastGPT 连接信息。复制示例配置并按环境填写： +复制 `.env.example` 为 `.env` 并填写配置；`.env` 不进入 Git。 -```bash -cp .env.example .env +- FastGPT：`AGENT_BACKEND=fastgpt`，填写 `ANALYSIS_SERVICE_URL`、`ANALYSIS_AUTH_TOKEN`、`APP_ID`。 +- LangGraph：`AGENT_BACKEND=langgraph`，按 `.env.example` 填写命名 LLM 配置， + 由 `LLM_PROVIDER` 选择模型；文本后端不要求 FastGPT 凭证。 +- LangGraph 存储：`AGENT_STORAGE=memory|sqlite|postgresql`。每个实例只选一个，修改后重启。 + `DATABASE_URL` 不控制 LangGraph 会话存储，切换数据库不自动搬迁旧数据。 + +本地 SQLite 示例： + +```dotenv +AGENT_BACKEND=langgraph +AGENT_STORAGE=sqlite +AGENT_SQLITE_PATH=./data/agent.sqlite3 +AGENT_DB_AUTO_MIGRATE=false ``` -需要设置的变量： - -| 变量 | 说明 | -|---|---| -| `ANALYSIS_SERVICE_URL` | FastGPT 服务地址 | -| `ANALYSIS_AUTH_TOKEN` | FastGPT API Key | -| `APP_ID` | FastGPT 应用 ID | - -## 运行 +首次初始化和运行： ```bash -python -m uvicorn src.main:app --host 127.0.0.1 --port 8000 +conda run -n py3.11 python -m src.storage.migrate --backend sqlite +conda run -n py3.11 python -m src.serve --port 8000 ``` -- `GET /` 存活检查 -- `POST /chat` 对话（`?stream=true` 为 SSE 流式） -- `POST /set_info`、`POST /get_info`、`DELETE /delete_session` 业务数据接口 +PostgreSQL 的连接池、迁移、本地 compose 与多 worker 部署见 +[docs/agent-storage.md](docs/agent-storage.md)。Windows 使用 `python -m src.serve`， +它为 psycopg 选择兼容的事件循环；Linux 也可用常规 uvicorn 命令启动。 + +## HTTP 接口 + +- `GET /`：存活检查。 +- `POST /chat`：对话；`?stream=true` 使用真实增量 SSE。 +- `POST /set_info`、`POST /get_info`：业务字段读写，不调用模型。 +- `DELETE /delete_session`：清理 checkpoint 和业务状态，真实存储错误返回失败。 +- SQLite/PostgreSQL 模式可给 `/chat` 添加 `Idempotency-Key` 请求头， + 重放已提交结果；复用 key 但请求内容不同返回业务码 409。 + +SSE 的 `error` 是失败终止，`done` 是成功结束。正文可以先流出， +调用方必须在收到 `done` 后才将本轮视为成功。`needFormUpdate=false` +只隐藏对外差量，不影响内部字段提交。 ## 测试 @@ -55,36 +66,35 @@ python -m uvicorn src.main:app --host 127.0.0.1 --port 8000 conda run -n py3.11 python -m pytest -q ``` -测试分三层： +测试覆盖公共 schema、两后端适配、流式校验/取消、会话串行、字段别名、错误码与持久化恢复。 +SQLite 集成测试使用临时真实文件；PostgreSQL 集成测试需要 +`AGENT_TEST_POSTGRES_DSN` 指向隔离测试库。未配置时 PostgreSQL 用例会明确跳过。 +两端都有直接退出子进程的恢复测试，不调用真实 LLM。 -- `test/api/test_public_schema_contract.py`：公共请求/响应 schema 的契约冻结测试 -- `test/api/test_chat_backend_boundary.py`：API 层与后端适配层的边界测试（fake 后端） -- `test/backends/test_fastgpt_backend.py`：FastGPT 适配器的翻译测试（fake SDK） +真实模型评测使用 `evals/run.py`，结果按每轮模型与代码版本单独保存， +历史轮次通过率不代替当前提交的业务验收。 ## 目录结构 ```text src/ - api/endpoints.py # HTTP 协议层：schema、state 前缀解析、SSE 事件 - backends/chat.py # 后端中立契约（ChatBackend Protocol、ChatInput/ChatResult/事件） - backends/fastgpt.py # FastGPT 适配器：SDK 调用与响应翻译 - core/fastgpt_client.py # 客户端生命周期与依赖注入（lifespan） - schemas/models.py # 公共 Pydantic 模型 -docs/langgraph-backend-migration-plan.md # 十阶段迁移作业计划 -evals/ # 调用真实服务的评测用例与 runner + api/endpoints.py # HTTP schema、阶段标签解析、SSE 协议 + agent/ # 流程图、节点、版本化提示词与模型服务 + backends/ # FastGPT / LangGraph 中立适配 + services/ # 四接口应用服务、字段注册表、会话锁 + storage/ # SQLite / PostgreSQL、轮次日志、migration + core/fastgpt_client.py # 后端与存储生命周期、依赖注入 + serve.py # 跨平台单 worker 启动入口 + schemas/models.py # 公共请求/响应模型 +test/storage/ # 两种数据库的真实集成与故障恢复测试 +evals/ # 真实模型评测用例与 runner ``` -## 当前状态与已知限制 +## 部署边界 -作业 1—5 已完成（LLM 提议 + 代码校验架构，提示词与 `workflow/20261020` -逐字对齐，全量测试 143 通过）。逐作业进度、真实模型评测结果与开放问题 -见 [docs/langgraph-backend-migration-plan.md](docs/langgraph-backend-migration-plan.md) -文末“当前进度与已知差异”。 - -- 真实模型黄金评测：`collection_skip_known_info` 12/12 通过； - `photo_skip_plate_confirm` 0/4、`no_false` 2/8，统一卡在短促回答“对” - 的分流语义上，待与 FastGPT 应用层配置对账（差异已留档，未叠代码守卫）。 -- `set_info` / `get_info` 仍直接依赖 FastGPT 状态与历史，迁移到共享 - 业务状态属于作业 6；`delete_session` 已按后端分流（langgraph 清 checkpoint）。 -- 服务启动（langgraph 模式）不需要 FastGPT 配置；需要 LLM provider 配置。 -- 流式为“完整答案切片”，真实增量输出属作业 8；`formUpdate` 差量属作业 6。 +- memory 是临时存储，重启丢失；SQLite 限定单实例、单 worker。 +- PostgreSQL 使用数据库级会话锁、版本和 epoch 校验；仍需按目标负载验证池容量与吞吐。 +- 业务快照是已提交状态的权威源，旧 checkpoint 执行进度在下一轮重建， + 不承诺从未提交的 LLM 调用中途续跑。 +- 语音 WebSocket 管线仍有独立配置，不由本次数据库选择自动迁移。 +- 未知字段写入、敏感字段访问边界等兼容事项继续按兼容清单处理。 diff --git a/compose.agent-db.yml b/compose.agent-db.yml new file mode 100644 index 0000000..3e30517 --- /dev/null +++ b/compose.agent-db.yml @@ -0,0 +1,18 @@ +services: + postgres: + image: postgres:17-alpine + ports: + - "127.0.0.1:${AGENT_POSTGRES_PORT:-5432}:5432" + environment: + POSTGRES_USER: ${AGENT_POSTGRES_USER:-agent} + POSTGRES_DB: ${AGENT_POSTGRES_DB:-agent} + POSTGRES_PASSWORD: ${AGENT_POSTGRES_PASSWORD:?Set AGENT_POSTGRES_PASSWORD in .env} + volumes: + - agent_postgres_data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U \"$${POSTGRES_USER}\" -d \"$${POSTGRES_DB}\""] + interval: 5s + timeout: 3s + retries: 20 +volumes: + agent_postgres_data: diff --git a/docs/agent-storage.md b/docs/agent-storage.md new file mode 100644 index 0000000..b591e7a --- /dev/null +++ b/docs/agent-storage.md @@ -0,0 +1,141 @@ +# LangGraph 会话存储：memory / SQLite / PostgreSQL + +每个应用实例只选择一个存储后端。两个数据库地址可以同时写在 `.env`， +启动时只连接 `AGENT_STORAGE` 指定的一套；切换后需要重启，配置切换不搬迁历史数据。 +现有 `DATABASE_URL` 不控制本模块。 + +| 模式 | 用途 | checkpoint | 会话互斥 | +|---|---|---|---| +| memory（默认） | 离线测试与临时开发，重启丢失 | InMemorySaver | 进程内会话锁 | +| sqlite | 单实例、单 worker，本地持久化 | AsyncSqliteSaver | 进程内会话锁 + 数据库文件的实例锁 | +| postgresql | 多 worker、多实例 | AsyncPostgresSaver | 进程内锁 + PostgreSQL session advisory lock | + +SQLite 实例锁由操作系统管理，第二个实例打开同一数据库会失败，进程退出自动释放。 +SQLite 文件必须放在本地持久化磁盘，不作为多主机共享文件使用。PostgreSQL 的 +锁连接与业务事务连接使用独立连接池；LLM 运行期间不持有业务行事务。 + +## SQLite 启用步骤 + +先安装 `requirements.txt`，Python 使用 Conda `py3.11` 环境。在 `.env` 中设置： + +```dotenv +AGENT_BACKEND=langgraph +AGENT_STORAGE=sqlite +AGENT_SQLITE_PATH=./data/agent.sqlite3 +AGENT_DB_AUTO_MIGRATE=false +``` + +首次启动前初始化两类表： + +```bash +conda run -n py3.11 python -m src.storage.migrate --backend sqlite +conda run -n py3.11 python -m src.serve --port 8000 +``` + +可用 `--sqlite-path` 指定迁移目标，但必须与服务配置相同。 +本地开发可以显式设 `AGENT_DB_AUTO_MIGRATE=true`，在启动阶段建表；请求处理不建表。 +默认关闭自动 migration，未初始化、版本不匹配或数据库不可用时启动失败，不回退内存。 + +## PostgreSQL 启用步骤 + +使用已有 PostgreSQL 数据库，或用 `compose.agent-db.yml` 启动本地开发库。 +建议为本服务使用独立数据库，业务表使用 `agent_` 前缀，checkpointer 管理自己的表。 + +```dotenv +AGENT_BACKEND=langgraph +AGENT_STORAGE=postgresql +AGENT_POSTGRES_DSN=postgresql://agent:replace-me@127.0.0.1:5432/agent +AGENT_DB_POOL_SIZE=10 +AGENT_DB_LOCK_TIMEOUT_SECONDS=30 +AGENT_DB_AUTO_MIGRATE=false + +# 以下四项供本地 compose 使用；密码必须与 DSN 一致。 +AGENT_POSTGRES_USER=agent +AGENT_POSTGRES_DB=agent +AGENT_POSTGRES_PASSWORD=replace-me +AGENT_POSTGRES_PORT=5432 +``` + +`replace-me` 是占位值，需要改为自己的密码；含特殊字符的 DSN 密码需 URL 编码。 + +```bash +docker compose -f compose.agent-db.yml up -d +conda run -n py3.11 python -m src.storage.migrate --backend postgresql +conda run -n py3.11 python -m src.serve --port 8000 +``` + +Linux 部署也可使用 `uvicorn src.main:app --host 0.0.0.0 --port 8000 --workers 4`。 +Windows 本地使用 `python -m src.serve`，该入口选择 psycopg 支持的 SelectorEventLoop。 +连接池在 lifespan 内创建、关闭；每个 worker 使用独立池。业务、checkpoint、 +会话锁各有一个池，每池最大连接数由 `AGENT_DB_POOL_SIZE` 控制，需计入数据库容量。 +同一组 worker 必须使用相同数据库，不能混用 SQLite 和 PostgreSQL。 + +## 数据结构与恢复规则 + +- `agent_sessions`：强类型业务字段、单调递增版本、会话 epoch、最后提交的完整图快照。 +- `agent_turns`：稳定 turn ID、pending/committed/aborted 状态、可选幂等键、请求摘要、已提交结果。 +- `agent_schema`：本模块 migration 版本。LangGraph saver 独立管理自己的 migration 和 checkpoint 表。 +- `/get_info` 以业务字段为权威读源。图快照保存历史、阶段和簿记；未确认的 checkpoint 不作为成功业务结果。 + +```mermaid +sequenceDiagram + participant A as 应用服务 + participant B as 业务数据库 + participant G as LangGraph/checkpointer + A->>B: 同会话锁内记录 pending turn + B-->>A: 已提交字段、版本、完整图快照 + A->>G: 清除旧执行 checkpoint，从提交快照执行本轮 + G-->>A: 合法前缀及暂定正文增量 + G->>G: 完成节点与 checkpoint 写入 + G->>A: await 提交回调（完整结果） + A->>B: 同一事务提交字段、快照、版本和 committed 结果 + B-->>A: 提交成功 + A-->>A: 允许成功响应 / SSE done +``` + +第一版每轮从业务提交快照重建图执行，丢弃旧 checkpoint 执行进度；不会从中断节点中途续跑。 +这会减少 saver 中的历史 checkpoint，但已提交的对话历史保存在业务图快照中， +轮次结果保存在业务日志中。LLM 决策与流程提示词保持不变。 + +| 故障窗口 | 下次请求的确定行为 | +|---|---| +| checkpoint 先写入，业务提交前退出 | pending 轮次转为 aborted，丢弃未提交执行进度，从上次业务快照开始 | +| 业务提交成功，checkpoint 缺失或滞后 | 业务快照是权威，下次重新构建 checkpoint，不重复应用字段差量 | +| 业务提交成功，成功响应发送前退出 | 同一 Idempotency-Key 重放原结果，不再调用 LLM 或推进版本 | + +set/get/delete 与 chat 使用同一个会话锁。字段写入带版本条件，过时 turn 或旧 epoch +不能覆盖新值；即使 advisory-lock 连接丢失，版本与 epoch 校验仍拒绝迟到提交。 +数据库不可用时返回失败，未提交的流不会发 done。客户端已收到的流式正文不可撤回， +必须把 done 当作本轮成功结束标记。 + +## 可选 Idempotency-Key + +SQLite/PostgreSQL 模式的 `/chat` 接受请求头 `Idempotency-Key`（1—128 字符）。 +同一会话 epoch 下，相同 key 和相同 text/clientMode/needFormUpdate 重放已提交结果； +相同 key 对应不同请求返回业务码 409，SSE 使用 error 事件。 +未提交请求可使用相同 key 重试；流式结果重放来自已提交结果，不会再次流式调用模型。 + +未提供 key 时，每次 POST 都是新轮次，不承诺网络重试自动去重。 +memory 模式收到此头返回 409，不假装提供持久化幂等能力。FastGPT 路径不提供本模块的幂等保证。 + +## 删除与数据库切换 + +删除先清理图 checkpoint，再删除业务数据与轮次日志。真实故障返回失败、允许重试。 +数据库保留不含业务字段/历史的 epoch 与版本墓碑，用于拒绝旧任务提交；重复删除成功。 +删除后新聊天进入新的 epoch，不会重放旧的幂等结果。 + +把 `AGENT_STORAGE=sqlite` 改为 `postgresql` 并重启，仅改变数据源。 +原 SQLite 文件仍保留，PostgreSQL 不会自动出现其中的会话。本版不提供自动跨数据库迁移。 +切换生产数据源需要单独的数据迁移与验证步骤。备份 SQLite 时先停止服务或使用 SQLite +备份机制；PostgreSQL 使用数据库备份工具。本地 compose 的 `down` 保留数据卷。 + +## 验证与测试资源 + +```bash +conda run -n py3.11 python -m pytest test/storage/ -q +``` + +默认跑真实 SQLite；PostgreSQL 测试需要 `AGENT_TEST_POSTGRES_DSN` 指向隔离测试库， +没有该变量时跳过 PostgreSQL 用例，不把跳过当作 PostgreSQL 已验证。 +测试使用 fake LLM，不消耗模型额度，覆盖：重启续聊、四接口、两种数据类型、 +幂等重放、版本冲突、checkpoint 缺失/超前、提交前后进程直接退出、数据库锁和删除恢复。 diff --git a/docs/langgraph-backend-migration-plan.md b/docs/langgraph-backend-migration-plan.md index 955a584..48e181f 100644 --- a/docs/langgraph-backend-migration-plan.md +++ b/docs/langgraph-backend-migration-plan.md @@ -357,7 +357,7 @@ docs/langgraph/ | FastGPT 应用层配置（应用级系统提示词、思考模式开关）与线上部署差异 | 作业 4/5 验收对账前 | 移植代码、离线评测与差异记录 | | 评测案例与工作流导出的分歧（“对”分流语义、2004 走向）的权威口径 | 作业 5 验收前 | 白名单双走向、逐字比对测试、失败证据留档 | -## 当前进度与已知差异（2026-10-21 更新） +## 当前进度与已知差异（数据库实现更新，模型评测为历史记录） ### 作业进度 @@ -368,15 +368,20 @@ docs/langgraph/ | 3. 领域状态与事件 | ✅ 完成 | 事件解析 14 例、白名单校验 24 例；`LEGAL_TRANSITIONS` 17 条均带 `source_ref` | | 4. 采集 + LLM | ✅ 完成 | 真实模型 `collection_skip_known_info` 12/12；单测含逐字比对 | | 5. 拍照与核实 | ✅ 完成 | 真实模型 4/4 + 12/12 + 2/2（20261021 基准） | -| 6—10 | 未开始 | — | +| 6. 四接口业务状态 | 已实现 | 应用服务、异步 repository、字段别名、会话串行与接口回归 | +| 7. 持久化 | SQLite/PostgreSQL 初版已实现 | `src/storage/`；真实两库集成、进程退出恢复、幂等和版本冲突；见 [agent-storage.md](agent-storage.md) | +| 8. 真流式 | 已实现核心路径 | 模型增量、前缀校验、取消与提交回调；流式终止事件回归 | +| 9—10 | 未完成 | 生产访问边界、容量验证与双后端等价性验收继续保留 | -全量单元测试 143 通过；七份提示词与 20261020 工作流逐字比对由参数化测试锁定。 +本轮全量测试 315 通过，包含真实 SQLite/PostgreSQL 集成测试。提示词来源和 +逐字比对见 `test/agent/test_workflow_prompts.py`。下方真实模型结果是此前轮次， +本次数据库开发未重新运行 46 条真实模型黄金评测。 ### 已落地的架构决策 1. **LLM 提议、代码校验（propose-validate）**：所有 stage_code 由节点内 LLM 产出（提示词即程序），代码只做前缀解析、白名单校验（含 clientMode 出口）和落库；曾尝试的代码强制推进/短促词守卫已按此分工移除。 2. **依赖注入走 `context_schema`**：`GraphDeps(llm=...)` 经 `Runtime` 注入调用期，不进 state/checkpoint；`build_graph()` 与 LLM 解耦。 -3. **与工作流逐字对齐**：提示词原文渲染（含 `{{$变量}}` 替换：系统时间按 Asia/Shanghai、人员预填字段缺失时显式标注“未提供”）；聊天历史按 user/assistant 消息传递（上限 50 条），对齐 chatNode `history=50`；welcome 话术为 Assigned Reply#3 原文并计入对话历史。 +3. **与工作流逐字对齐**：提示词原文渲染（含 `{{$变量}}` 替换：系统时间按 Asia/Shanghai、人员预填字段缺失时显式标注“未提供”）；聊天历史按 user/assistant 消息传递（`AGENT_HISTORY_MESSAGES` 默认 50，0 为不截断）；welcome 话术为 Assigned Reply#3 原文并计入对话历史。 4. **不引入 LangGraph Server / langgraph_sdk**：生产形态的 Agent Server 需要 LangSmith license key、外网 beacon 出网及 Postgres+Redis 全栈，与课程目标（作业 7 亲手实现持久化）和部署环境不符；`langgraph dev` 仅为内存态，无增量能力。自部署高并发方案 = 嵌入库模式：无状态 FastAPI 多副本 + AsyncPostgresSaver + 每会话 advisory lock 串行 + 连接池预算（社区参照：langgraph-agent-toolkit、agent-bakery）。 5. **2004 语义**：确认单车车牌；车牌已明确时可 2003→2005 跳过，未明确时 2003→2004→2005。两种走向均在白名单登记，由模型判断。 diff --git a/requirements.txt b/requirements.txt index 64f244a..fd24525 100644 --- a/requirements.txt +++ b/requirements.txt @@ -22,3 +22,7 @@ sqlalchemy pymysql PyYAML>=6.0 langgraph==1.2.9 +langgraph-checkpoint-sqlite==3.1.2 +langgraph-checkpoint-postgres==3.2.0 +psycopg[binary,pool]>=3.2,<4 +aiosqlite>=0.20,<1 diff --git a/src/api/endpoints.py b/src/api/endpoints.py index fb95f41..314375b 100644 --- a/src/api/endpoints.py +++ b/src/api/endpoints.py @@ -1,4 +1,5 @@ -from fastapi import APIRouter, HTTPException, Depends +from typing import Annotated +from fastapi import APIRouter, HTTPException, Depends, Header 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 @@ -6,6 +7,7 @@ from ..backends.chat import ( ChatBackend, ChatBackendAPIError, ChatBackendAuthenticationError, + ChatBackendConflictError, ChatBackendRateLimitError, ChatInput, FormUpdate, @@ -156,7 +158,8 @@ def create_sse_event(event: str, data: dict) -> str: async def chat( request: ProcessRequest_chat, stream: bool = False, - backend: ChatBackend = Depends(get_chat_backend) + backend: ChatBackend = Depends(get_chat_backend), + idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", min_length=1, max_length=128)] = None, ): """Handle chat completion request. @@ -172,6 +175,7 @@ async def chat( text=json_data['text'], need_form_update=need_form_update, client_mode=client_mode, + request_key=idempotency_key, ) # 应用服务：langgraph 模式下包装字段注入/写回；fastgpt 模式透传。 # 单测直调函数（未走 lifespan）时回退 backend 直调。 @@ -419,6 +423,8 @@ async def chat( code, msg = "401", "认证失败" elif isinstance(e, ChatBackendRateLimitError): code, msg = "429", "请求过于频繁，请稍后重试" + elif isinstance(e, ChatBackendConflictError): + code, msg = "409", "请求幂等键或会话版本冲突" else: code, msg = "500", "大模型服务器无响应" yield create_sse_event("error", {"msg": msg, "code": code}) @@ -458,6 +464,12 @@ async def chat( code="429", msg="请求过于频繁，请稍后重试" ) + except ChatBackendConflictError: + return ProcessResponse_chat( + sessionId=json_data['sessionId'], timeStamp=json_data['timeStamp'], + outputText="", nextStage="", nextStageCode="", code="409", + msg="请求幂等键或会话版本冲突", + ) except ChatBackendAPIError as e: logger.error(f"API error: {e}") return ProcessResponse_chat( diff --git a/src/backends/chat.py b/src/backends/chat.py index 3a3f916..a93a27c 100644 --- a/src/backends/chat.py +++ b/src/backends/chat.py @@ -1,6 +1,6 @@ """Backend-neutral chat contract.""" -from collections.abc import AsyncIterator, Callable +from collections.abc import AsyncIterator, Awaitable, Callable from dataclasses import dataclass, field from typing import Any, Protocol @@ -18,7 +18,8 @@ class ChatInput: # FastGPT 后端忽略该字段。 injected_fields: dict | None = None # 图提交后、公开结果之前同步业务字段；不进入图状态或 HTTP schema。 - on_commit: Callable[[dict], None] | None = field(default=None, repr=False, compare=False) + on_commit: Callable[..., Awaitable[None] | None] | None = field(default=None, repr=False, compare=False) + request_key: str | None = None @dataclass(frozen=True) @@ -69,6 +70,10 @@ class ChatBackendError(Exception): """Base error raised by a chat backend adapter.""" +class ChatBackendConflictError(ChatBackendError): + """请求幂等键或会话版本冲突。""" + + class ChatBackendAuthenticationError(ChatBackendError): """The backend rejected its configured credentials.""" diff --git a/src/backends/langgraph.py b/src/backends/langgraph.py index 2d8c785..22c3868 100644 --- a/src/backends/langgraph.py +++ b/src/backends/langgraph.py @@ -15,6 +15,7 @@ test/backends/test_langgraph_streaming.py """ import asyncio +import inspect from contextlib import suppress from collections.abc import AsyncIterator, Awaitable, Callable @@ -103,6 +104,10 @@ class LangGraphBackend(ChatBackend): self._graph = graph if graph is not None else build_graph() self._llm = llm if llm is not None else workflow_fake_llm() + async def reset_checkpoint(self, session_id): + from ..storage.checkpoints import delete_checkpoint + await delete_checkpoint(self._graph, session_id) + async def complete(self, chat_input: ChatInput) -> ChatResult: """非流式：执行一轮图并翻译结果。 @@ -135,15 +140,13 @@ class LangGraphBackend(ChatBackend): LLMAuthError -> ChatBackendAuthenticationError（API 层 401）； LLMRateLimitError -> ChatBackendRateLimitError（API 层 429）。 """ - graph_input: dict = { + graph_input: dict = dict(chat_input.injected_fields or {}) + # 恢复快照中的上一轮 input_text 不得覆盖本次请求。 + graph_input.update({ "session_id": chat_input.session_id, "input_text": chat_input.text, "client_mode": chat_input.client_mode, - } - # 作业 6：应用服务注入的最新业务字段覆盖本轮输入， - # 使 set_info 写入后本轮 chat 立即可见（输入覆盖 checkpoint 同键）。 - if chat_input.injected_fields: - graph_input.update(chat_input.injected_fields) + }) try: result = await self._graph.ainvoke( graph_input, @@ -152,7 +155,9 @@ class LangGraphBackend(ChatBackend): ) # 即使消费者在图完成后断开，业务 repository 仍与 checkpoint 一致。 if chat_input.on_commit is not None: - chat_input.on_commit(result.get("turn_patch") or {}) + pending = chat_input.on_commit(result.get("turn_patch") or {}, state=result) + if inspect.isawaitable(pending): + await pending return result except LLMAuthError as exc: raise ChatBackendAuthenticationError(str(exc)) from exc diff --git a/src/core/config.py b/src/core/config.py index cee3594..6a34b22 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -19,6 +19,14 @@ class Config: # "fastgpt" 保持现有行为；"langgraph" 使用内存图 fake 回复。 AGENT_BACKEND = os.getenv("AGENT_BACKEND", "fastgpt") + # 每个实例只选一个会话存储；与通用 DATABASE_URL 独立。 + AGENT_STORAGE = os.getenv("AGENT_STORAGE", "memory").strip().lower() + AGENT_SQLITE_PATH = os.getenv("AGENT_SQLITE_PATH", "./data/agent.sqlite3") + AGENT_POSTGRES_DSN = os.getenv("AGENT_POSTGRES_DSN", "") + AGENT_DB_AUTO_MIGRATE = os.getenv("AGENT_DB_AUTO_MIGRATE", "false").lower() in ("true", "1") + AGENT_DB_POOL_SIZE = int(os.getenv("AGENT_DB_POOL_SIZE", "10")) + AGENT_DB_LOCK_TIMEOUT_SECONDS = float(os.getenv("AGENT_DB_LOCK_TIMEOUT_SECONDS", "30")) + # 命名 LLM 提供商列表（均为 OpenAI 兼容接口），两种写法二选一： # ① JSON 字典（推荐，单变量管理）： # LLM_PROVIDERS={"deepseek": {"base_url": "https://...", "api_key": "...", @@ -138,6 +146,17 @@ class Config: raise ValueError( f"AGENT_BACKEND must be 'fastgpt' or 'langgraph', got {cls.AGENT_BACKEND!r}" ) + if cls.AGENT_BACKEND == "langgraph": + if cls.AGENT_STORAGE not in ("memory", "sqlite", "postgresql"): + raise ValueError("AGENT_STORAGE must be memory, sqlite or postgresql") + if cls.AGENT_STORAGE == "sqlite" and ( + not cls.AGENT_SQLITE_PATH.strip() or cls.AGENT_SQLITE_PATH == ":memory:" + ): + raise ValueError("AGENT_SQLITE_PATH must be a persistent file path") + if cls.AGENT_STORAGE == "postgresql" and not cls.AGENT_POSTGRES_DSN.strip(): + raise ValueError("AGENT_POSTGRES_DSN is required for postgresql storage") + if cls.AGENT_DB_POOL_SIZE < 1 or cls.AGENT_DB_LOCK_TIMEOUT_SECONDS <= 0: + raise ValueError("Database pool size and lock timeout must be positive") profiles = cls.llm_profiles() # JSON 非法时在此抛出明确错误 incomplete = [n for n in cls._declared_provider_names() if n not in profiles] if incomplete: diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py index 300631f..d073bfd 100644 --- a/src/core/fastgpt_client.py +++ b/src/core/fastgpt_client.py @@ -1,107 +1,79 @@ -"""FastGPT client and chat backend dependency injection.""" -from contextlib import asynccontextmanager +"""后端与单一会话存储的生命周期、依赖注入。""" +from contextlib import AsyncExitStack, asynccontextmanager + from fastapi import FastAPI from fastgpt_client import AsyncChatClient + from ..agent.graph import build_graph from ..backends.chat import ChatBackend from ..backends.fastgpt import FastGPTBackend from ..backends.langgraph import LangGraphBackend from ..services.app_service import AccidentAppService -from ..services.repository import InMemorySessionRepository +from ..storage.checkpoints import delete_checkpoint from .config import Config - _fastgpt_client: AsyncChatClient | None = None _chat_backend: ChatBackend | None = None _langgraph_graph = None _app_service: AccidentAppService | None = None -# 两后端缺省 60 秒、最多额外重试 3 次，1/2/4 秒退避。 -# FastGPT SDK 的 max_retries 实为总尝试数，注入时转换为 1 + 重试次数。 -# LLM 关闭 SDK 自动重试；流式发出文本后不再重试。 -# LLM 预算按单次模型调用计算，FastGPT timeout 为 HTTP SDK 超时， -# 均不承诺多次模型调用的整个聊天回合具有相同总期限。 FASTGPT_TIMEOUT_SECONDS = 60.0 FASTGPT_MAX_RETRIES = 3 @asynccontextmanager async def lifespan(app: FastAPI): - """Manage chat backend resource lifecycle.""" - global _chat_backend, _fastgpt_client + global _chat_backend, _fastgpt_client, _langgraph_graph, _app_service Config.validate() llm = None + async with AsyncExitStack() as resources: + try: + if Config.AGENT_BACKEND == 'langgraph': + from ..agent.llm import build_llm_service + from ..storage.runtime import StorageSettings, open_storage - if Config.AGENT_BACKEND == "langgraph": - # 作业 2：LangGraph fake 模式不要求 FastGPT 密钥。 - # TODO(作业 2): 若语音仍启用，需在此单独初始化并校验 FastGPT 配置。 - from ..agent.llm import build_llm_service - - global _langgraph_graph, _app_service - _langgraph_graph = build_graph() - llm = build_llm_service() - _chat_backend = LangGraphBackend(_langgraph_graph, llm=llm) - # 作业 6：应用服务 + 内存 repository（作业 7 换 PostgreSQL）。 - _app_service = AccidentAppService( - _chat_backend, InMemorySessionRepository() - ) - else: - _fastgpt_client = AsyncChatClient( - api_key=Config.FASTGPT_API_KEY, - base_url=Config.FASTGPT_BASE_URL, - timeout=FASTGPT_TIMEOUT_SECONDS, - max_retries=FASTGPT_MAX_RETRIES + 1, - retry_delay=1.0, - enable_logging=Config.DEBUG, - ) - await _fastgpt_client.__aenter__() - _chat_backend = FastGPTBackend(_fastgpt_client) - - try: - yield - finally: - _chat_backend = None - _langgraph_graph = None - _app_service = None - if llm is not None and hasattr(llm, "aclose"): - await llm.aclose() - if _fastgpt_client: - await _fastgpt_client.__aexit__(None, None, None) - _fastgpt_client = None + storage = await resources.enter_async_context(open_storage(StorageSettings.from_config(Config))) + _langgraph_graph = build_graph(storage.checkpointer) + llm = build_llm_service() + _chat_backend = LangGraphBackend(_langgraph_graph, llm=llm) + _app_service = AccidentAppService(_chat_backend, storage.repository, locks=storage.locks) + else: + _fastgpt_client = AsyncChatClient( + api_key=Config.FASTGPT_API_KEY, base_url=Config.FASTGPT_BASE_URL, + timeout=FASTGPT_TIMEOUT_SECONDS, max_retries=FASTGPT_MAX_RETRIES + 1, + retry_delay=1.0, enable_logging=Config.DEBUG, + ) + await _fastgpt_client.__aenter__() + _chat_backend = FastGPTBackend(_fastgpt_client) + yield + finally: + _chat_backend = None + _langgraph_graph = None + _app_service = None + if llm is not None and hasattr(llm, 'aclose'): + await llm.aclose() + if _fastgpt_client: + await _fastgpt_client.__aexit__(None, None, None) + _fastgpt_client = None async def delete_chat_session(thread_id: str) -> None: - """删除图 checkpoint；缺少清理能力或存储出错必须报告失败。""" - if _langgraph_graph is None: - raise RuntimeError("LangGraph backend not initialized") - checkpointer = getattr(_langgraph_graph, "checkpointer", None) - async_delete = getattr(checkpointer, "adelete_thread", None) - if callable(async_delete): - await async_delete(thread_id) - return - sync_delete = getattr(checkpointer, "delete_thread", None) - if callable(sync_delete): - sync_delete(thread_id) - return - raise RuntimeError("LangGraph checkpointer does not support thread deletion") + await delete_checkpoint(_langgraph_graph, thread_id) def get_fastgpt_client() -> AsyncChatClient: - """Get the initialized FastGPT client.""" if _fastgpt_client is None: - raise RuntimeError("FastGPT client not initialized") + raise RuntimeError('FastGPT client not initialized') return _fastgpt_client def get_app_service() -> AccidentAppService: - """Get the initialized application service（langgraph 模式）。""" if _app_service is None: - raise RuntimeError("Application service not initialized") + raise RuntimeError('Application service not initialized') return _app_service def get_chat_backend() -> ChatBackend: - """Get the backend-neutral chat service.""" if _chat_backend is None: - raise RuntimeError("Chat backend not initialized") + raise RuntimeError('Chat backend not initialized') return _chat_backend diff --git a/src/serve.py b/src/serve.py new file mode 100644 index 0000000..5f09955 --- /dev/null +++ b/src/serve.py @@ -0,0 +1,23 @@ +"""Portable single-worker entry point: python -m src.serve --port 8000.""" +import argparse +import asyncio +import os + +import uvicorn + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--host', default='127.0.0.1') + parser.add_argument('--port', default=8000, type=int) + args = parser.parse_args() + # psycopg 不支持 Windows ProactorEventLoop。直接运行 Server.serve + # 使用已创建的 Selector loop，不被 uvicorn 默认 loop factory 覆盖。 + factory = asyncio.SelectorEventLoop if os.name == 'nt' else None + config = uvicorn.Config('src.main:app', host=args.host, port=args.port, loop='asyncio') + with asyncio.Runner(loop_factory=factory) as runner: + runner.run(uvicorn.Server(config).serve()) + + +if __name__ == '__main__': + main() diff --git a/src/services/app_service.py b/src/services/app_service.py index ff68778..0d22def 100644 --- a/src/services/app_service.py +++ b/src/services/app_service.py @@ -1,65 +1,41 @@ -"""事故处理应用服务（作业 6-1）：四业务接口共享同一份业务状态。 +"""四接口应用服务：同会话串行，持久化模式按已提交快照恢复。""" -职责边界（对应计划作业 6）： -- set_info / get_info / delete_session **不调用 LLM、不创建辅助对话** - （对照旧 FastGPT 实现的“辅助对话 + 删记录”hack）； -- chat 每轮从 repository 读最新业务字段注入图（set_info 写入后 - 下一轮 chat 必须可见），图产出字段差量后写回 repository； -- chat / stream / set_info / get_info / delete_session 都在同会话锁内。 - 锁的引用计数包含等待者，取消和异常都会归还引用；没有使用者时回收。 - 不同会话独立执行，同会话 set_info 不会被在途聊天的旧结果覆盖。 -- 未知 key 兼容策略：get 对未知 key 返回 JSON 空串 '""'（现状兼容）； - set 对未知 key 拒绝写入（白名单保护，差异见兼容清单）； -- includeInputInfo 继续被 endpoint 接受，但本层不再使用 - ——新后端无需触发辅助对话。 +import hashlib +import json +from contextlib import aclosing, asynccontextmanager, suppress +from dataclasses import dataclass, replace -endpoint 只负责校验、协议（SSE / 响应模型）与依赖注入。 -""" - -from contextlib import aclosing -from dataclasses import replace - -from ..backends.chat import ChatBackend, ChatInput, ChatResult, FormUpdate +from ..backends.chat import ChatBackendConflictError, ChatInput, ChatResult, FormUpdate, TextDelta from .fields_registry import ( - FIELD_REGISTRY, - GROUP_KEYS, - BUSINESS_TO_FORM_KEY, - FieldGroup, - business_patch_from_form, - business_value, - canonical_key, - parse_field_value, - to_external_string, + FIELD_REGISTRY, GROUP_KEYS, BUSINESS_TO_FORM_KEY, FieldGroup, + business_patch_from_form, business_value, canonical_key, + parse_field_value, to_external_string, ) -from .repository import CheckpointNotFoundError, SessionRepository +from .repository import CheckpointNotFoundError from .session_locks import SessionLocks -# set_info 对未知/受保护 key 的拒绝码（与现有“入参不规范”文案对齐） -SET_INFO_INVALID_KEY_MSG = "入参不规范：未知或受保护的字段 key" +SET_INFO_INVALID_KEY_MSG = '入参不规范：未知或受保护的字段 key' + + +@dataclass +class _Turn: + injected: dict + ticket: object = None + committed: bool = False class AccidentAppService: - """四个业务接口的应用服务（langgraph 后端）。""" - - def __init__(self, backend: ChatBackend, repository: SessionRepository): + def __init__(self, backend, repository, *, locks=None): self._backend = backend self._repository = repository - self._locks = SessionLocks() + self._locks = locks if locks is not None else SessionLocks() + self._durable = hasattr(repository, 'begin_turn') - # ---- chat：图每轮读取最新业务状态，产出差量写回 ---- - - def _injected_fields(self, session_id: str) -> dict: - """repository 业务字段 -> 图输入的分组注入。 - - acd 业务键经 BUSINESS_TO_FORM_KEY 映射回图表单键 - （无表单来源的 bjrjs/ywmtc 不注入）；人员业务键与图的 - person_fields 同构，原样注入。 - """ - business = self._repository.load(session_id) + async def _injected_fields(self, session_id): + business = await self._repository.load(session_id) if not business: return {} - injected: dict = {"accident_fields": {}, "person_fields": {}} - # 注入同义键的一致视图，覆盖 checkpoint 里的旧值。 + injected = {'accident_fields': {}, 'person_fields': {}} for key, spec in FIELD_REGISTRY.items(): value = business_value(business, key) if value is None: @@ -67,141 +43,129 @@ class AccidentAppService: if spec.group is FieldGroup.ACD: form_key = BUSINESS_TO_FORM_KEY.get(key) if form_key is not None: - injected["accident_fields"][form_key] = value + injected['accident_fields'][form_key] = value else: - injected["person_fields"][key] = value + injected['person_fields'][key] = value return injected - def _writeback_patch(self, session_id: str, patch: dict) -> None: - """图产出的差量（表单键/人员键）经映射落 repository 业务键。""" + async def _writeback_patch(self, session_id, patch): business_patch = business_patch_from_form(patch) if business_patch: - self._repository.apply_update(session_id, business_patch) + await self._repository.apply_update(session_id, business_patch) - async def chat(self, chat_input: ChatInput) -> ChatResult: - """非流式一轮对话：注入最新业务字段 -> 执行图 -> 写回字段差量。""" - async with self._locks.hold(chat_input.session_id): - return await self._chat_locked(chat_input) - - async def _chat_locked(self, chat_input: ChatInput) -> ChatResult: - injected = self._injected_fields(chat_input.session_id) - committed = False - - def commit(patch): - nonlocal committed - self._writeback_patch(chat_input.session_id, patch) - committed = True - - result = await self._backend.complete( - replace(chat_input, injected_fields=injected or None, on_commit=commit) - ) - if not committed: - patch = result.committed_patch - if patch is None: - patch = result.form_update - if isinstance(patch, dict): - commit(patch) - return replace( - result, form_update=result.form_update if chat_input.need_form_update else {}, - ) - - async def stream(self, chat_input: ChatInput): - """流式代理：字段注入同 chat()；FormUpdate 事件捕获写回后透传。""" - async with self._locks.hold(chat_input.session_id): - async with aclosing(self._stream_locked(chat_input)) as events: - async for event in events: - yield event - - async def _stream_locked(self, chat_input: ChatInput): - injected = self._injected_fields(chat_input.session_id) - committed = False - - def commit(patch): - nonlocal committed - self._writeback_patch(chat_input.session_id, patch) - committed = True - - result_input = replace(chat_input, injected_fields=injected or None, on_commit=commit) - async with aclosing(self._backend.stream(result_input)) as events: - async for event in events: - if isinstance(event, FormUpdate): - if not committed and isinstance(event.data, dict): - commit(event.data) - if not chat_input.need_form_update: - continue - yield event - - # ---- set_info / get_info：零 LLM、零辅助对话 ---- - - async def set_info(self, session_id: str, key: str, value: str) -> tuple[str, str]: - async with self._locks.hold(session_id): - return self._set_info(session_id, key, value) - - def _set_info(self, session_id: str, key: str, value: str) -> tuple[str, str]: - """白名单校验后写入 repository。返回 (code, msg)。""" - spec = FIELD_REGISTRY.get(key) - if spec is None or not spec.writable: - return "500", SET_INFO_INVALID_KEY_MSG + @asynccontextmanager + async def _turn(self, request): + ticket = None + turn = None try: - parsed = parse_field_value(spec, value) - except ValueError: - return "500", "入参不规范：字段值无法解析" - self._repository.apply_update(session_id, {canonical_key(key): parsed}) - return "200", "" + if self._durable: + payload = {'text': request.text, 'client_mode': request.client_mode, + 'need_form_update': request.need_form_update} + digest = hashlib.sha256(json.dumps(payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest() + ticket = await self._repository.begin_turn(request.session_id, request.request_key, digest) + if ticket.replay is None: + # 官方 checkpoint 可能先写成功，业务事务却未提交。 + # 丢弃该执行进度；始终从业务表的提交快照恢复。 + await self._backend.reset_checkpoint(request.session_id) + elif request.request_key: + raise ChatBackendConflictError('Idempotency-Key requires persistent storage') + injected = dict(ticket.graph_state) if ticket else {} + injected.update(await self._injected_fields(request.session_id)) + if ticket: + injected['state_version'] = ticket.version + turn = _Turn(injected, ticket) + yield turn + except BaseException: + if ticket is not None and ticket.replay is None and not (turn and turn.committed): + # 清理失败时保留 pending；下次持锁访问会标记 aborted 并恢复快照。 + with suppress(Exception): + await self._repository.abort_turn(ticket) + raise - async def get_info(self, session_id: str, key: str) -> str: + async def _commit(self, request, turn, patch, state): + if turn.ticket: + if state is None: + raise RuntimeError('Persistent turns require a complete graph state') + await self._repository.commit_turn(turn.ticket, business_patch_from_form(patch), state) + else: + await self._writeback_patch(request.session_id, patch) + turn.committed = True + + @staticmethod + def _replay(request, state): + patch = state.get('turn_patch') or {} + return ChatResult(content=state['response_text'], status_code=state.get('stage_code'), + form_update=patch if request.need_form_update else {}, committed_patch=patch) + + async def chat(self, request: ChatInput): + async with self._locks.hold(request.session_id): + async with self._turn(request) as turn: + if turn.ticket and turn.ticket.replay is not None: + return self._replay(request, turn.ticket.replay) + async def commit(patch, *, state=None): + await self._commit(request, turn, patch, state) + result = await self._backend.complete( + replace(request, injected_fields=turn.injected or None, on_commit=commit)) + if not turn.committed: + patch = result.committed_patch if result.committed_patch is not None else result.form_update + await commit(patch if isinstance(patch, dict) else {}) + return replace(result, form_update=result.form_update if request.need_form_update else {}) + + async def stream(self, request: ChatInput): + async with self._locks.hold(request.session_id): + async with self._turn(request) as turn: + if turn.ticket and turn.ticket.replay is not None: + result = self._replay(request, turn.ticket.replay) + yield TextDelta(result.content) + if result.form_update: + yield FormUpdate(result.form_update) + return + async def commit(patch, *, state=None): + await self._commit(request, turn, patch, state) + forwarded = replace(request, injected_fields=turn.injected or None, on_commit=commit) + async with aclosing(self._backend.stream(forwarded)) as events: + async for event in events: + if isinstance(event, FormUpdate): + if not turn.committed and isinstance(event.data, dict): + await commit(event.data) + if not request.need_form_update: + continue + yield event + + async def set_info(self, session_id, key, value): async with self._locks.hold(session_id): - return self._get_info(session_id, key) + spec = FIELD_REGISTRY.get(key) + if spec is None or not spec.writable: + return '500', SET_INFO_INVALID_KEY_MSG + try: + parsed = parse_field_value(spec, value) + except ValueError: + return '500', '入参不规范：字段值无法解析' + await self._repository.apply_update(session_id, {canonical_key(key): parsed}) + return '200', '' - def _get_info(self, session_id: str, key: str) -> str: - """按分组或单键读出，返回 JSON 字符串（兼容现有 /get_info 契约）。""" - import json + async def get_info(self, session_id, key): + async with self._locks.hold(session_id): + business = await self._repository.load(session_id) + def group_value(group): + return {k: to_external_string(business_value(business, k)) for k in GROUP_KEYS[group]} + if key == 'all': + value = {group.value: group_value(group) for group in FieldGroup} + else: + try: + value = group_value(FieldGroup(key)) + except ValueError: + value = to_external_string(business_value(business, key)) if key in FIELD_REGISTRY else '' + return json.dumps(value, ensure_ascii=False) - business = self._repository.load(session_id) - if key == "all": - return json.dumps( - { - group.value: { - k: to_external_string(business_value(business, k)) - for k in GROUP_KEYS[group] - } - for group in FieldGroup - }, - ensure_ascii=False, - ) - try: - group = FieldGroup(key) - except ValueError: - group = None - if group is not None: - return json.dumps( - {k: to_external_string(business_value(business, k)) for k in GROUP_KEYS[group]}, - ensure_ascii=False, - ) - # 单键：未知 key 返回 JSON 空串（兼容现状）；已知 key 返回其外部表示 - if key not in FIELD_REGISTRY: - return json.dumps("", ensure_ascii=False) - return json.dumps(to_external_string(business_value(business, key)), ensure_ascii=False) - - # ---- delete_session：业务状态 + 图 checkpoint 一起清理 ---- - - async def delete_session(self, session_id: str, drop_checkpoint) -> None: - """删除会话全部数据。 - - drop_checkpoint：由 core 层注入的 checkpoint 清理函数 - （图生命周期归 lifespan 管理，本层不持有图引用）。 - 与会话锁互斥：删除不与在途 chat/set/get 交错。 - 释放锁引用时保留等待者使用的锁，最后一个使用者退出后回收。 - checkpoint 清理成功（或明确不存在）后才删除业务字段。 - 其他异常向上传播，由 endpoint 返回失败，保留业务字段供重试。 - """ + async def delete_session(self, session_id, drop_checkpoint): async with self._locks.hold(session_id): if drop_checkpoint is not None: try: await drop_checkpoint(session_id) except CheckpointNotFoundError: pass - self._repository.delete(session_id) + await self._repository.delete(session_id) -__all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"] +__all__ = ['AccidentAppService', 'SET_INFO_INVALID_KEY_MSG'] diff --git a/src/services/repository.py b/src/services/repository.py index 6b6ebb8..2663922 100644 --- a/src/services/repository.py +++ b/src/services/repository.py @@ -1,55 +1,39 @@ -"""会话业务状态 repository（作业 6-1/6-2）。 - -抽象出 repository 接口，使应用服务不依赖具体存储； -作业 6 用内存实现跑通四接口，作业 7 再替换为 PostgreSQL 实现 -（对话 checkpoint 已由 LangGraph checkpointer 承担，本层只管 -业务字段——单一事实来源，图每轮读取最新值）。 - -语义约定： -- 业务状态按外部业务键（hphm1、ywrysw...）扁平存储，值为内部强类型； -- 不存在的会话视为空状态（读取返回 {}），与首轮 checkpoint 无字段一致； -- delete 幂等：重复删除固定成功，不抛异常（行为进测试）。 -""" +"""异步业务字段存储接口；字段值保留 bool/int/str 的区别。""" from typing import Protocol class CheckpointNotFoundError(Exception): - """存储适配器明确确认线程不存在；不能用一般存储异常代替。""" + """存储适配器明确确认线程不存在。""" class SessionRepository(Protocol): - """业务字段存储接口（作业 7 由 PostgreSQL 实现替换）。""" - - def load(self, session_id: str) -> dict: - """读取会话业务字段；无记录时返回空 dict。""" + async def load(self, session_id: str) -> dict: + """读取业务字段；不存在返回空 dict。""" ... - def apply_update(self, session_id: str, patch: dict) -> dict: - """合并差量并返回合并后的完整业务字段。""" + async def apply_update(self, session_id: str, patch: dict) -> dict: + """提交差量后返回完整业务字段。""" ... - def delete(self, session_id: str) -> None: - """删除会话业务状态；幂等，重复删除不抛异常。""" + async def delete(self, session_id: str) -> None: + """幂等删除；真实存储错误必须传播。""" ... class InMemorySessionRepository: - """内存实现：字典存储，仅用于单进程开发与测试。""" - - def __init__(self) -> None: + def __init__(self): self._store: dict[str, dict] = {} - def load(self, session_id: str) -> dict: + async def load(self, session_id: str) -> dict: return dict(self._store.get(session_id) or {}) - def apply_update(self, session_id: str, patch: dict) -> dict: + async def apply_update(self, session_id: str, patch: dict) -> dict: state = self._store.setdefault(session_id, {}) state.update(patch) return dict(state) - def delete(self, session_id: str) -> None: - # 幂等：pop 的缺省值吞掉不存在的情况，重复删除行为固定为成功 + async def delete(self, session_id: str) -> None: self._store.pop(session_id, None) diff --git a/src/storage/__init__.py b/src/storage/__init__.py new file mode 100644 index 0000000..57e7339 --- /dev/null +++ b/src/storage/__init__.py @@ -0,0 +1 @@ +"""数据库适配器：短事务、连接池生命周期与 SQLite 单实例文件锁。""" diff --git a/src/storage/checkpoints.py b/src/storage/checkpoints.py new file mode 100644 index 0000000..841ee4e --- /dev/null +++ b/src/storage/checkpoints.py @@ -0,0 +1,16 @@ +"""图 checkpoint 清理的统一边界；不把缺少清理能力当成功。""" + + +async def delete_checkpoint(graph, session_id): + if graph is None: + raise RuntimeError('LangGraph backend not initialized') + saver = getattr(graph, 'checkpointer', None) + async_delete = getattr(saver, 'adelete_thread', None) + if callable(async_delete): + await async_delete(session_id) + return + sync_delete = getattr(saver, 'delete_thread', None) + if callable(sync_delete): + sync_delete(session_id) + return + raise RuntimeError('LangGraph checkpointer does not support thread deletion') diff --git a/src/storage/database.py b/src/storage/database.py new file mode 100644 index 0000000..89d5829 --- /dev/null +++ b/src/storage/database.py @@ -0,0 +1,93 @@ +import asyncio +import os +from contextlib import asynccontextmanager +from pathlib import Path + + +class SqliteInstanceLock: + """拒绝第二个进程/实例打开同一 SQLite 文件；锁由操作系统随退出释放。""" + + def __init__(self, path): + self.path = Path(str(path) + '.instance.lock') + self.file = None + + def __enter__(self): + self.path.parent.mkdir(parents=True, exist_ok=True) + self.file = self.path.open('a+b') + try: + self.file.seek(0, os.SEEK_END) + if self.file.tell() == 0: + self.file.write(b'0') + self.file.flush() + self.file.seek(0) + if os.name == 'nt': + import msvcrt + msvcrt.locking(self.file.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(self.file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + self.file.close() + self.file = None + raise RuntimeError('SQLite storage already has an active instance; use PostgreSQL for multiple workers') from exc + return self + + def __exit__(self, *_): + if self.file is not None: + self.file.seek(0) + if os.name == 'nt': + import msvcrt + msvcrt.locking(self.file.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + fcntl.flock(self.file.fileno(), fcntl.LOCK_UN) + self.file.close() + self.file = None + + +class Connection: + def __init__(self, raw, postgres=False): + self.raw = raw + self.postgres = postgres + + async def execute(self, sql, params=()): + if self.postgres: + sql = sql.replace('?', '%s') + return await self.raw.execute(sql, params) + + async def one(self, sql, params=()): + cursor = await self.execute(sql, params) + try: + row = await cursor.fetchone() + return dict(row) if row is not None else None + finally: + await cursor.close() + + +class SqliteDatabase: + def __init__(self, connection): + self.connection = connection + self.lock = asyncio.Lock() + + @asynccontextmanager + async def transaction(self): + # 不跨 LLM 调用持有事务；仅保护一次业务提交。 + async with self.lock: + await self.connection.execute('BEGIN IMMEDIATE') + try: + yield Connection(self.connection) + await self.connection.commit() + except BaseException: + await self.connection.rollback() + raise + + +class PostgresDatabase: + def __init__(self, pool): + self.pool = pool + + @asynccontextmanager + async def transaction(self): + async with self.pool.connection() as connection: + async with connection.transaction(): + yield Connection(connection, postgres=True) diff --git a/src/storage/locks.py b/src/storage/locks.py new file mode 100644 index 0000000..b0b8d07 --- /dev/null +++ b/src/storage/locks.py @@ -0,0 +1,61 @@ +import asyncio +import hashlib +from contextlib import asynccontextmanager + +from ..backends.chat import ChatBackendAPIError +from ..services.session_locks import SessionLocks + + +class PostgresSessionLocks: + """本进程锁 + 独立连接上的 session advisory lock；不跨 LLM 持有行事务。""" + + def __init__(self, pool, timeout=30): + self.pool = pool + self.timeout = timeout + self.local = SessionLocks() + + @staticmethod + def key(session_id): + digest = hashlib.blake2b(('znjj-agent:'+session_id).encode(), digest_size=8).digest() + return int.from_bytes(digest, 'big', signed=True) + + async def _release(self, connection, key): + try: + async with asyncio.timeout(5): + await connection.execute('SELECT pg_advisory_unlock(%s)', (key,)) + except Exception: + # 不把仍持锁或失联的连接放回池。 + await connection.close() + + @asynccontextmanager + async def hold(self, session_id): + async with self.local.hold(session_id): + async with self.pool.connection(timeout=self.timeout) as connection: + acquired = False + key = self.key(session_id) + try: + try: + async with asyncio.timeout(self.timeout): + while True: + cursor = await connection.execute('SELECT pg_try_advisory_lock(%s) AS locked', (key,)) + row = await cursor.fetchone() + await cursor.close() + if row['locked']: + acquired = True + break + await asyncio.sleep(0.05) + except TimeoutError as exc: + raise ChatBackendAPIError('Session is busy; retry later') from exc + yield + finally: + if not acquired: + # 查询取消时服务端可能已加锁但客户端尚未收到结果。 + # 关闭此连接，避免把潜在持锁连接放回池。 + await connection.close() + if acquired: + release = asyncio.create_task(self._release(connection, key)) + try: + await asyncio.shield(release) + except asyncio.CancelledError: + await release + raise diff --git a/src/storage/migrate.py b/src/storage/migrate.py new file mode 100644 index 0000000..554682a --- /dev/null +++ b/src/storage/migrate.py @@ -0,0 +1,34 @@ +"""python -m src.storage.migrate --backend sqlite|postgresql""" +import argparse +import asyncio +import os +from dataclasses import replace + +from ..core.config import Config +from .runtime import StorageSettings, open_storage + + +async def migrate(settings): + if settings.kind == 'memory': + raise ValueError('Select --backend sqlite or postgresql') + async with open_storage(replace(settings, auto_migrate=True)): + print(f'Initialized {settings.kind} business schema and LangGraph checkpoint tables.') + + +def main(): + parser = argparse.ArgumentParser(description='Initialize the selected agent storage (no data transfer between backends).') + parser.add_argument('--backend', choices=['sqlite', 'postgresql']) + parser.add_argument('--sqlite-path') + args = parser.parse_args() + settings = StorageSettings.from_config(Config) + if args.backend: + settings = replace(settings, kind=args.backend) + if args.sqlite_path: + settings = replace(settings, sqlite_path=args.sqlite_path) + if os.name == 'nt' and settings.kind == 'postgresql': + asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) + asyncio.run(migrate(settings)) + + +if __name__ == '__main__': + main() diff --git a/src/storage/repository.py b/src/storage/repository.py new file mode 100644 index 0000000..f8ca02f --- /dev/null +++ b/src/storage/repository.py @@ -0,0 +1,142 @@ +"""SQLite/PostgreSQL 共用的业务表、版本检查与轮次提交日志。""" + +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from uuid import uuid4 + +from ..backends.chat import ChatBackendConflictError + +SCHEMA_VERSION = 1 +DDL = [ + 'CREATE TABLE IF NOT EXISTS agent_schema (version INTEGER PRIMARY KEY)', + '''CREATE TABLE IF NOT EXISTS agent_sessions ( + session_id TEXT PRIMARY KEY, version BIGINT NOT NULL, epoch BIGINT NOT NULL, + fields TEXT NOT NULL, graph_state TEXT NOT NULL, deleted INTEGER NOT NULL, + updated_at TEXT NOT NULL)''', + '''CREATE TABLE IF NOT EXISTS agent_turns ( + turn_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, epoch BIGINT NOT NULL, + request_key TEXT, request_hash TEXT NOT NULL, base_version BIGINT NOT NULL, + status TEXT NOT NULL, result TEXT, created_at TEXT NOT NULL, + UNIQUE(session_id, epoch, request_key))''', + 'CREATE INDEX IF NOT EXISTS agent_turns_session ON agent_turns(session_id)', +] + + +def encode(value): + return json.dumps(value, ensure_ascii=False, separators=(',', ':')) + + +def now(): + return datetime.now(timezone.utc).isoformat() + + +@dataclass +class TurnTicket: + session_id: str + turn_id: str + version: int + epoch: int + graph_state: dict + replay: dict | None = None + + +class SqlSessionRepository: + def __init__(self, database): + self.database = database + + async def migrate(self): + async with self.database.transaction() as conn: + for sql in DDL: + await conn.execute(sql) + await conn.execute('INSERT INTO agent_schema(version) VALUES (?) ON CONFLICT DO NOTHING', (SCHEMA_VERSION,)) + + async def verify(self): + async with self.database.transaction() as conn: + row = await conn.one('SELECT MAX(version) AS version FROM agent_schema') + if not row or row['version'] != SCHEMA_VERSION: + raise RuntimeError('Unsupported agent storage schema; run the storage migration command') + + async def _ensure(self, conn, session_id): + await conn.execute('''INSERT INTO agent_sessions + (session_id,version,epoch,fields,graph_state,deleted,updated_at) + VALUES (?,0,0,'{}','{}',0,?) ON CONFLICT DO NOTHING''', (session_id, now())) + return await conn.one('SELECT * FROM agent_sessions WHERE session_id=?', (session_id,)) + + async def load(self, session_id): + async with self.database.transaction() as conn: + row = await conn.one('SELECT fields FROM agent_sessions WHERE session_id=?', (session_id,)) + return json.loads(row['fields']) if row else {} + + async def apply_update(self, session_id, patch): + async with self.database.transaction() as conn: + row = await self._ensure(conn, session_id) + fields = json.loads(row['fields']) + fields.update(patch) + cursor = await conn.execute('''UPDATE agent_sessions SET fields=?,version=version+1, + deleted=0,updated_at=? WHERE session_id=? AND version=?''', + (encode(fields), now(), session_id, row['version'])) + if cursor.rowcount != 1: + raise ChatBackendConflictError('Session version changed; retry the request') + return fields + + async def delete(self, session_id): + async with self.database.transaction() as conn: + row = await self._ensure(conn, session_id) + cursor = await conn.execute('''UPDATE agent_sessions SET fields='{}',graph_state='{}', + deleted=1,version=version+1,epoch=epoch+1,updated_at=? + WHERE session_id=? AND version=?''', (now(), session_id, row['version'])) + if cursor.rowcount != 1: + raise ChatBackendConflictError('Session version changed during deletion') + await conn.execute('DELETE FROM agent_turns WHERE session_id=?', (session_id,)) + + async def begin_turn(self, session_id, request_key, request_hash): + async with self.database.transaction() as conn: + row = await self._ensure(conn, session_id) + previous = None + if request_key: + previous = await conn.one('''SELECT * FROM agent_turns + WHERE session_id=? AND epoch=? AND request_key=?''', + (session_id, row['epoch'], request_key)) + if previous and previous['request_hash'] != request_hash: + raise ChatBackendConflictError('Idempotency-Key was used with a different request') + if previous and previous['status'] == 'committed': + return TurnTicket(session_id, previous['turn_id'], row['version'], row['epoch'], + json.loads(row['graph_state']), json.loads(previous['result'])) + # 同会话锁已持有：遗留 pending 只能来自中断，下一次从提交快照重建。 + await conn.execute("UPDATE agent_turns SET status='aborted' WHERE session_id=? AND status='pending'", (session_id,)) + turn_id = previous['turn_id'] if previous else uuid4().hex + if previous: + await conn.execute("UPDATE agent_turns SET status='pending',base_version=?,result=NULL WHERE turn_id=?", + (row['version'], turn_id)) + else: + await conn.execute('''INSERT INTO agent_turns + (turn_id,session_id,epoch,request_key,request_hash,base_version,status,result,created_at) + VALUES (?,?,?,?,?,?,'pending',NULL,?)''', + (turn_id, session_id, row['epoch'], request_key, request_hash, row['version'], now())) + return TurnTicket(session_id, turn_id, row['version'], row['epoch'], json.loads(row['graph_state'])) + + async def commit_turn(self, ticket, patch, graph_state): + async with self.database.transaction() as conn: + row = await conn.one('SELECT * FROM agent_sessions WHERE session_id=?', (ticket.session_id,)) + if not row or row['epoch'] != ticket.epoch or row['version'] != ticket.version: + raise ChatBackendConflictError('Session changed before turn commit') + fields = json.loads(row['fields']) + fields.update(patch) + state = dict(graph_state) + state['state_version'] = ticket.version + 1 + state['turn_patch'] = graph_state.get('turn_patch') or {} + cursor = await conn.execute('''UPDATE agent_sessions SET fields=?,graph_state=?,deleted=0, + version=version+1,updated_at=? WHERE session_id=? AND version=? AND epoch=?''', + (encode(fields), encode(state), now(), ticket.session_id, ticket.version, ticket.epoch)) + if cursor.rowcount != 1: + raise ChatBackendConflictError('Session version changed during commit') + result = {key: state.get(key) for key in ('response_text', 'turn_patch', 'stage_code')} + cursor = await conn.execute("UPDATE agent_turns SET status='committed',result=? WHERE turn_id=? AND status='pending'", + (encode(result), ticket.turn_id)) + if cursor.rowcount != 1: + raise ChatBackendConflictError('Turn is no longer pending') + + async def abort_turn(self, ticket): + async with self.database.transaction() as conn: + await conn.execute("UPDATE agent_turns SET status='aborted' WHERE turn_id=? AND status='pending'", (ticket.turn_id,)) diff --git a/src/storage/runtime.py b/src/storage/runtime.py new file mode 100644 index 0000000..857c519 --- /dev/null +++ b/src/storage/runtime.py @@ -0,0 +1,92 @@ +from contextlib import AsyncExitStack, asynccontextmanager +from dataclasses import dataclass +from pathlib import Path + +from langgraph.checkpoint.memory import InMemorySaver + +from ..services.repository import InMemorySessionRepository +from ..services.session_locks import SessionLocks +from .database import PostgresDatabase, SqliteDatabase, SqliteInstanceLock +from .repository import SqlSessionRepository + + +@dataclass(frozen=True) +class StorageSettings: + kind: str = 'memory' + sqlite_path: str = './data/agent.sqlite3' + postgres_dsn: str = '' + auto_migrate: bool = False + pool_size: int = 10 + lock_timeout: float = 30 + + @classmethod + def from_config(cls, config): + return cls(config.AGENT_STORAGE, config.AGENT_SQLITE_PATH, config.AGENT_POSTGRES_DSN, + config.AGENT_DB_AUTO_MIGRATE, config.AGENT_DB_POOL_SIZE, + config.AGENT_DB_LOCK_TIMEOUT_SECONDS) + + +@dataclass +class StorageRuntime: + repository: object + checkpointer: object + locks: object + + +@asynccontextmanager +async def open_storage(settings: StorageSettings): + if settings.kind not in ('memory', 'sqlite', 'postgresql'): + raise ValueError('Unknown AGENT_STORAGE') + async with AsyncExitStack() as stack: + if settings.kind == 'memory': + yield StorageRuntime(InMemorySessionRepository(), InMemorySaver(), SessionLocks()) + return + if settings.kind == 'sqlite': + import aiosqlite + from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver + + if not settings.sqlite_path.strip() or settings.sqlite_path == ':memory:': + raise ValueError('SQLite persistence requires a file path') + path = Path(settings.sqlite_path).resolve() + path.parent.mkdir(parents=True, exist_ok=True) + stack.enter_context(SqliteInstanceLock(path)) + business = await stack.enter_async_context(aiosqlite.connect(path, isolation_level=None)) + business.row_factory = aiosqlite.Row + await business.execute('PRAGMA journal_mode=WAL') + await business.execute('PRAGMA busy_timeout=5000') + checkpoint_connection = await stack.enter_async_context(aiosqlite.connect(path)) + await checkpoint_connection.execute('PRAGMA busy_timeout=5000') + checkpointer = AsyncSqliteSaver(checkpoint_connection) + database = SqliteDatabase(business) + locks = SessionLocks() + else: + from psycopg.rows import dict_row + from psycopg_pool import AsyncConnectionPool + from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver + from .locks import PostgresSessionLocks + + if not settings.postgres_dsn.strip(): + raise ValueError('AGENT_POSTGRES_DSN is required') + async def pool(): + value = AsyncConnectionPool( + settings.postgres_dsn, min_size=1, max_size=settings.pool_size, open=False, + timeout=settings.lock_timeout, + kwargs={'autocommit': True, 'prepare_threshold': 0, 'row_factory': dict_row, 'connect_timeout': 5}, + ) + await stack.enter_async_context(value) + await value.wait(timeout=15) + return value + business_pool = await pool() + checkpoint_pool = await pool() + lock_pool = await pool() + database = PostgresDatabase(business_pool) + checkpointer = AsyncPostgresSaver(checkpoint_pool) + locks = PostgresSessionLocks(lock_pool, settings.lock_timeout) + repository = SqlSessionRepository(database) + if settings.auto_migrate: + await repository.migrate() + await checkpointer.setup() + await repository.verify() + # 启动时同时检查 saver 表，不能仅检查业务表后偷偷用内存 saver。 + await checkpointer.aget_tuple({'configurable': {'thread_id': '__storage_health__'}}) + yield StorageRuntime(repository, checkpointer, locks) diff --git a/test/api/test_assignment6_endpoints.py b/test/api/test_assignment6_endpoints.py index ed7a1b3..5c269ec 100644 --- a/test/api/test_assignment6_endpoints.py +++ b/test/api/test_assignment6_endpoints.py @@ -20,6 +20,8 @@ from src.main import app def client(monkeypatch): """langgraph 后端 + fake LLM 的测试客户端（lifespan 完整初始化）。""" monkeypatch.setattr(Config, "AGENT_BACKEND", "langgraph") + # 通用 API 契约测试不连接用户 .env 中的真实持久化数据库。 + monkeypatch.setattr(Config, "AGENT_STORAGE", "memory") monkeypatch.setattr(Config, "LLM_PROVIDER", "fake") import src.agent.llm as llm_module diff --git a/test/services/test_delete_failure_semantics.py b/test/services/test_delete_failure_semantics.py index 27b89e5..759ccc6 100644 --- a/test/services/test_delete_failure_semantics.py +++ b/test/services/test_delete_failure_semantics.py @@ -59,7 +59,7 @@ async def test_cancelled_checkpoint_cleanup_keeps_business_fields(): @pytest.mark.asyncio async def test_business_delete_failure_is_not_reported_as_success(): class FailingRepository(InMemorySessionRepository): - def delete(self, _): + async def delete(self, _): raise ConnectionError('business storage offline') service = AccidentAppService(SimpleNamespace(), FailingRepository()) await service.set_info('s', 'hphm1', 'test-plate') diff --git a/test/services/test_field_commit_and_retry.py b/test/services/test_field_commit_and_retry.py index 9d9fdff..163c0a9 100644 --- a/test/services/test_field_commit_and_retry.py +++ b/test/services/test_field_commit_and_retry.py @@ -31,8 +31,8 @@ async def test_person_synonyms_share_storage_and_prompt_values(): await service.set_info('s', api, 'new') assert json.loads(await service.get_info('s', api)) == 'new' assert json.loads(await service.get_info('s', native)) == 'new' - assert api not in repo.load('s') - fields = service._injected_fields('s')['person_fields'] + assert api not in await repo.load('s') + fields = (await service._injected_fields('s'))['person_fields'] prompt = render_prompt(SINGLE_CONFIRM, {'person_fields': fields}) assert 'old' not in prompt assert 'new' in prompt diff --git a/test/storage/__init__.py b/test/storage/__init__.py new file mode 100644 index 0000000..b153f2f --- /dev/null +++ b/test/storage/__init__.py @@ -0,0 +1 @@ +"""SQLite/PostgreSQL durability integration tests.""" diff --git a/test/storage/crash_worker.py b/test/storage/crash_worker.py new file mode 100644 index 0000000..35e1d1b --- /dev/null +++ b/test/storage/crash_worker.py @@ -0,0 +1,43 @@ +"""Integration-test child: exit abruptly at a durable turn's commit boundary.""" +import asyncio +import os +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from src.agent.graph import build_graph +from src.backends.chat import ChatInput +from src.backends.langgraph import LangGraphBackend +from src.services.app_service import AccidentAppService +from src.storage.runtime import StorageSettings, open_storage + + +class LLM: + async def complete(self, system, user, *, history=None): + if '【字段提取】' in system: + return '{"has_injury":false}' + return '1002crash-boundary-reply' + + +async def main(): + kind, path, session, boundary = sys.argv[1:] + settings = StorageSettings(kind=kind, sqlite_path=path, + postgres_dsn=os.getenv('AGENT_TEST_POSTGRES_DSN',''), pool_size=2) + async with open_storage(settings) as storage: + service = AccidentAppService(LangGraphBackend(build_graph(storage.checkpointer), llm=LLM()), + storage.repository, locks=storage.locks) + original = storage.repository.commit_turn + async def crash(*args): + if boundary=='before': + os._exit(23) + await original(*args) + os._exit(24) + storage.repository.commit_turn = crash + await service.chat(ChatInput(session,'继续办理',request_key='crash-key')) + + +if __name__ == '__main__': + factory=asyncio.SelectorEventLoop if os.name=='nt' else None + with asyncio.Runner(loop_factory=factory) as runner: + runner.run(main()) diff --git a/test/storage/test_http_persistence.py b/test/storage/test_http_persistence.py new file mode 100644 index 0000000..cea9816 --- /dev/null +++ b/test/storage/test_http_persistence.py @@ -0,0 +1,54 @@ +import asyncio +from contextlib import asynccontextmanager +from dataclasses import replace + +import httpx +import pytest +from fastapi import FastAPI + +from src.api.endpoints import router +from src.core import fastgpt_client as wiring +from src.core.config import Config +from src.storage.runtime import open_storage +from .test_persistence import RecordingLLM, run, settings + + +def test_http_restart_replay_headers_and_switching(settings, monkeypatch): + import src.agent.llm as llm_module + monkeypatch.setattr(Config, 'AGENT_BACKEND', 'langgraph') + monkeypatch.setattr(Config, 'LLM_PROVIDER', 'fake') + monkeypatch.setattr(Config, 'AGENT_STORAGE', settings.kind) + monkeypatch.setattr(Config, 'AGENT_SQLITE_PATH', settings.sqlite_path) + monkeypatch.setattr(Config, 'AGENT_POSTGRES_DSN', settings.postgres_dsn) + monkeypatch.setattr(Config, 'AGENT_DB_POOL_SIZE', settings.pool_size) + monkeypatch.setattr(Config, 'AGENT_DB_AUTO_MIGRATE', True) + monkeypatch.setattr(llm_module, 'build_llm_service', lambda: RecordingLLM()) + app = FastAPI() + app.include_router(router) + @asynccontextmanager + async def client(): + async with wiring.lifespan(app): + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as value: + yield value + async def scenario(): + import uuid + payload = {'sessionId': uuid.uuid4().hex, 'timeStamp':'20261022120000'} + request = {**payload, 'text':'新对话'} + async with client() as http: + assert (await http.post('/set_info', json={**payload,'key':'hphm1','value':'http-plate'})).json()['code']=='200' + first = (await http.post('/chat', json=request, headers={'Idempotency-Key':'http-key'})).json() + assert first['nextStageCode']=='1001' + monkeypatch.setattr(Config, 'AGENT_DB_AUTO_MIGRATE', False) + async with client() as http: + assert (await http.post('/chat', json=request, headers={'Idempotency-Key':'http-key'})).json()==first + conflict = await http.post('/chat', json={**request,'text':'different'}, headers={'Idempotency-Key':'http-key'}) + assert conflict.json()['code']=='409' + stream = await http.post('/chat?stream=true', json={**payload,'text':'继续办理','needFormUpdate':False}) + assert 'event: done' in stream.text + assert 'event: formUpdate' not in stream.text + assert '"0"' == (await http.post('/get_info', json={**payload,'key':'jdcsl'})).json()['value'] + assert (await http.request('DELETE','/delete_session',json=payload)).json()['code']=='200' + async with client() as http: + assert (await http.post('/get_info',json={**payload,'key':'hphm1'})).json()['value']=='""' + assert (await http.post('/chat',json=request,headers={'Idempotency-Key':'http-key'})).json()['nextStageCode']=='1001' + run(scenario()) diff --git a/test/storage/test_persistence.py b/test/storage/test_persistence.py new file mode 100644 index 0000000..c7bb228 --- /dev/null +++ b/test/storage/test_persistence.py @@ -0,0 +1,221 @@ +import asyncio +import json +import os +import subprocess +import sys +from pathlib import Path +from contextlib import aclosing, asynccontextmanager +from dataclasses import replace +from uuid import uuid4 + +import pytest + +from src.agent.graph import build_graph +from src.backends.chat import ChatBackendConflictError, ChatInput, FormUpdate +from src.backends.langgraph import LangGraphBackend +from src.services.app_service import AccidentAppService +from src.storage.runtime import StorageSettings, open_storage + + +def run(coroutine): + # psycopg 在 Windows 要求 SelectorEventLoop；Linux 使用标准 runner。 + factory = asyncio.SelectorEventLoop if os.name == 'nt' else None + with asyncio.Runner(loop_factory=factory) as runner: + return runner.run(coroutine) + + +@pytest.fixture(params=['sqlite', 'postgresql']) +def settings(request, tmp_path): + if request.param == 'postgresql': + dsn = os.getenv('AGENT_TEST_POSTGRES_DSN') + if not dsn: + pytest.skip('Set AGENT_TEST_POSTGRES_DSN to an isolated PostgreSQL database') + return StorageSettings(kind='postgresql', postgres_dsn=dsn, auto_migrate=True, pool_size=2) + return StorageSettings(kind='sqlite', sqlite_path=str(tmp_path/'agent.sqlite3'), auto_migrate=True) + + +class RecordingLLM: + def __init__(self): + self.histories = [] + + async def complete(self, system, user, *, history=None): + if '【字段提取】' in system: + return '{"has_injury":false,"vehicle_count":0}' + self.histories.append(history or []) + return '1002已记录' + + async def stream_tokens(self, system, user, *, history=None): + self.histories.append(history or []) + yield '1002' + yield '已记录' + + +@asynccontextmanager +async def service_for(settings, llm=None): + async with open_storage(settings) as storage: + backend = LangGraphBackend(build_graph(storage.checkpointer), llm=llm or RecordingLLM()) + yield AccidentAppService(backend, storage.repository, locks=storage.locks), storage + + +async def session_row(storage, session): + async with storage.repository.database.transaction() as conn: + return await conn.one('SELECT * FROM agent_sessions WHERE session_id=?', (session,)) + + +def test_restart_keeps_fields_phase_and_history(settings): + async def scenario(): + session = uuid4().hex + async with service_for(settings) as (service, storage): + await service.set_info(session, 'hphm1', 'test-plate') + first = await service.chat(ChatInput(session, '新对话')) + assert first.content.startswith('1001') + await service.chat(ChatInput(session, '继续办理')) + values = await storage.repository.load(session) + assert values['ywrysw'] is False + assert type(values['jdcsl']) is int and values['jdcsl'] == 0 + llm = RecordingLLM() + async with service_for(replace(settings, auto_migrate=False), llm) as (service, storage): + assert json.loads(await service.get_info(session, 'hphm1')) == 'test-plate' + result = await service.chat(ChatInput(session, '重启后的新输入')) + assert result.content.startswith('1002') + assert any(m['content'] == '继续办理' for m in llm.histories[0]) + row = await session_row(storage, session) + graph = json.loads(row['graph_state']) + assert graph['turn_count'] == 3 + assert graph['dialogue'][-2]['content'] == '重启后的新输入' + assert graph['state_version'] == row['version'] + run(scenario()) + + +@pytest.mark.parametrize('stream', [False, True]) +def test_idempotency_replay_after_restart_and_conflict(settings, stream): + async def scenario(): + session = uuid4().hex + request = ChatInput(session, '继续办理', request_key='stable-key', need_form_update=True) + async with service_for(settings) as (service, storage): + await service.chat(ChatInput(session, '新对话')) + if stream: + first = [event async for event in service.stream(request)] + assert any(isinstance(event, FormUpdate) for event in first) + else: + first = await service.chat(request) + before = await session_row(storage, session) + llm = RecordingLLM() + async with service_for(settings, llm) as (service, storage): + result = await service.chat(request) + assert result.content == '1002已记录' + assert llm.histories == [] + assert (await session_row(storage, session))['version'] == before['version'] + with pytest.raises(ChatBackendConflictError): + await service.chat(replace(request, text='different')) + run(scenario()) + + +def test_checkpoint_ahead_of_business_commit_is_discarded(settings): + async def scenario(): + session = uuid4().hex + async with service_for(settings) as (service, storage): + await service.chat(ChatInput(session, '新对话')) + original = storage.repository.commit_turn + async def fail(*args): raise ConnectionError('business commit failed') + async def unavailable(*args): raise ConnectionError('journal cleanup failed') + storage.repository.commit_turn = fail + storage.repository.abort_turn = unavailable + with pytest.raises(ConnectionError): + await service.chat(ChatInput(session, '未提交的输入', request_key='retry')) + storage.repository.commit_turn = original + llm = RecordingLLM() + async with service_for(settings, llm) as (service, storage): + await service.chat(ChatInput(session, '未提交的输入', request_key='retry')) + assert all(m['content'] != '未提交的输入' for m in llm.histories[0]) + assert json.loads((await session_row(storage, session))['graph_state'])['turn_count'] == 2 + run(scenario()) + + +def test_committed_business_state_restores_missing_checkpoint(settings): + async def scenario(): + session = uuid4().hex + async with service_for(settings) as (service, storage): + await service.chat(ChatInput(session, '新对话')) + await service.chat(ChatInput(session, '继续办理')) + await storage.checkpointer.adelete_thread(session) + async with service_for(settings) as (service, storage): + reply = await service.chat(ChatInput(session, '再次输入')) + assert reply.content.startswith('1002') + assert json.loads((await session_row(storage, session))['graph_state'])['turn_count'] == 3 + run(scenario()) + + +def test_disconnect_after_commit_can_replay_without_advancing(settings): + async def scenario(): + session = uuid4().hex + request = ChatInput(session, '继续办理', request_key='disconnect') + async with service_for(settings) as (service, storage): + await service.chat(ChatInput(session, '新对话')) + initial = await session_row(storage, session) + async with aclosing(service.stream(request)) as events: + await anext(events) + async with asyncio.timeout(5): + while (await session_row(storage, session))['version'] == initial['version']: + await asyncio.sleep(0.01) + # 此时图及业务提交已完成，但消费者没有读到回合末尾。 + before = await session_row(storage, session) + llm = RecordingLLM() + async with service_for(settings, llm) as (service, storage): + await service.chat(request) + assert llm.histories == [] + assert (await session_row(storage, session))['version'] == before['version'] + run(scenario()) + + +def test_delete_then_restart_and_stale_commit_cannot_revive_data(settings): + async def scenario(): + session = uuid4().hex + async with service_for(settings) as (service, storage): + await service.set_info(session, 'hphm1', 'old-plate') + await service.chat(ChatInput(session, '新对话', request_key='old-key')) + ticket = await storage.repository.begin_turn(session, None, 'pending') + await service.delete_session(session, service._backend.reset_checkpoint) + with pytest.raises(ChatBackendConflictError): + await storage.repository.commit_turn(ticket, {'hphm1':'revived'}, {}) + assert await storage.checkpointer.aget_tuple({'configurable': {'thread_id':session}}) is None + async with service_for(settings) as (service, storage): + assert json.loads(await service.get_info(session, 'hphm1')) == '' + reply = await service.chat(ChatInput(session, '新会话', request_key='old-key')) + assert reply.content.startswith('1001') + assert json.loads((await session_row(storage, session))['graph_state'])['turn_count'] == 1 + run(scenario()) + + +def test_business_version_conflict_rejects_stale_turn(settings): + async def scenario(): + session = uuid4().hex + async with service_for(settings) as (service, storage): + ticket = await storage.repository.begin_turn(session, None, 'pending') + await storage.repository.apply_update(session, {'jdcsl':2}) + with pytest.raises(ChatBackendConflictError): + await storage.repository.commit_turn(ticket, {'jdcsl':1}, {}) + assert (await storage.repository.load(session))['jdcsl'] == 2 + run(scenario()) + + +@pytest.mark.parametrize('boundary', ['before', 'after']) +def test_abrupt_process_exit_at_commit_boundary(settings, boundary): + session = uuid4().hex + async def initialize(): + async with service_for(settings) as (service, storage): + await service.chat(ChatInput(session, '新对话')) + run(initialize()) + worker = Path(__file__).with_name('crash_worker.py') + result = subprocess.run([sys.executable, str(worker), settings.kind, settings.sqlite_path, session, boundary], + capture_output=True, timeout=45) + assert result.returncode == (23 if boundary=='before' else 24), result.stderr.decode(errors='replace') + async def recover(): + llm = RecordingLLM() + async with service_for(replace(settings,auto_migrate=False),llm) as (service,storage): + reply=await service.chat(ChatInput(session,'继续办理',request_key='crash-key')) + assert reply.content.startswith('1002') + assert bool(llm.histories) is (boundary=='before') + row=await session_row(storage,session) + assert json.loads(row['graph_state'])['turn_count']==2 + run(recover()) diff --git a/test/storage/test_storage_config_and_locks.py b/test/storage/test_storage_config_and_locks.py new file mode 100644 index 0000000..279533b --- /dev/null +++ b/test/storage/test_storage_config_and_locks.py @@ -0,0 +1,88 @@ +import asyncio +import os +from dataclasses import replace +from uuid import uuid4 + +import pytest + +from src.core.config import Config +from src.storage.runtime import StorageSettings, open_storage +from .test_persistence import run + + +@pytest.fixture +def postgres_settings(): + dsn = os.getenv('AGENT_TEST_POSTGRES_DSN') + if not dsn: + pytest.skip('Set AGENT_TEST_POSTGRES_DSN to run PostgreSQL integration tests') + return StorageSettings(kind='postgresql',postgres_dsn=dsn,auto_migrate=True,pool_size=2) + + +def test_sqlite_rejects_second_instance_and_releases_guard(tmp_path): + async def scenario(): + config = StorageSettings(kind='sqlite',sqlite_path=str(tmp_path/'single.db'),auto_migrate=True) + async with open_storage(config): + with pytest.raises(RuntimeError, match='active instance'): + async with open_storage(config): pass + async with open_storage(config): pass + run(scenario()) + + +def test_sqlite_requires_migration_and_does_not_fall_back(tmp_path): + async def scenario(): + config = StorageSettings(kind='sqlite',sqlite_path=str(tmp_path/'empty.db')) + with pytest.raises(Exception): + async with open_storage(config): + pytest.fail('Uninitialized durable storage must not start') + async with open_storage(replace(config,auto_migrate=True)): pass + async with open_storage(config) as runtime: + assert runtime.checkpointer.__class__.__name__=='AsyncSqliteSaver' + run(scenario()) + + +def test_postgresql_locks_serialize_independent_application_instances(postgres_settings): + settings = postgres_settings + async def scenario(): + session = uuid4().hex + async with open_storage(settings) as first, open_storage(settings) as second: + entered = asyncio.Event() + async def waiting(): + async with second.locks.hold(session): entered.set() + async with first.locks.hold(session): + task=asyncio.create_task(waiting()) + await asyncio.sleep(0.15) + assert not entered.is_set() + # 同一个锁池中，不同会话仍能执行。 + async with second.locks.hold(session+'-other'): + pass + await asyncio.wait_for(task,3) + assert entered.is_set() + run(scenario()) + + +def test_postgresql_cancelled_lock_holder_releases_server_lock(postgres_settings): + settings = postgres_settings + async def scenario(): + session=uuid4().hex + async with open_storage(settings) as first, open_storage(settings) as second: + ready=asyncio.Event() + async def holding(): + async with first.locks.hold(session): + ready.set() + await asyncio.Future() + task=asyncio.create_task(holding()) + await ready.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): await task + async with asyncio.timeout(3): + async with second.locks.hold(session): pass + run(scenario()) + + +@pytest.mark.parametrize('kind', ['unknown', 'sqlite', 'postgresql']) +def test_invalid_storage_config_is_rejected(monkeypatch,kind): + monkeypatch.setattr(Config,'AGENT_BACKEND','langgraph') + monkeypatch.setattr(Config,'AGENT_STORAGE',kind) + monkeypatch.setattr(Config,'AGENT_SQLITE_PATH',':memory:') + monkeypatch.setattr(Config,'AGENT_POSTGRES_DSN','') + with pytest.raises(ValueError):Config.validate()