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()