feat(services): assignment 6 skeleton - app service, field registry, in-memory repository

- fields_registry: 24 business keys in acd/human1/human2 groups with
  value_type, writable and SensitiveLevel metadata; strict bool/int
  parsing (no bool('0') trap); INTERNAL_FIELD_KEYS protection list
- repository: SessionRepository protocol + InMemorySessionRepository
  (idempotent delete; PostgreSQL swap-in is assignment 7)
- AccidentAppService: chat/set_info/get_info/delete_session share one
  business state - chat injects latest repository fields into the graph
  and writes back validated field diffs; set/get/delete make zero LLM
  calls and zero helper conversations (includeInputInfo accepted but
  unused); unknown get keys keep the '' JSON-empty-string contract,
  unknown set keys are rejected by whitelist
- ChatInput.injected_fields neutral contract extension; LangGraphBackend
  merges injected fields into graph input (set_info visible next chat)
- endpoints: thin layer - langgraph mode routes business state through
  the app service (fastgpt mode keeps the legacy path); delete cleans
  repository + graph checkpoint, idempotent
- tests: registry completeness (sfzmwh/sjwh/csbw included), app service
  combos with LLM-call spy, HTTP acceptance flow (set -> get -> chat ->
  groups -> delete -> no leak, idempotent re-delete)
- docs/assignment6-compat.md: behavior diff list vs FastGPT status quo
  and the audit backlog (arbitrary set_info key usage, stream formUpdate
  write-back, sensitive-field display boundary)

Verified: 200 unit tests (172 existing + 28 new)
This commit is contained in:
Eric Wang
2026-10-22 20:50:10 +08:00
parent 6c1fbd0295
commit 8717d9066c
12 changed files with 882 additions and 16 deletions
+44
View File
@@ -0,0 +1,44 @@
# 作业 6 兼容差异清单:LangGraph 业务状态 vs FastGPT 现状
> 范围:`/set_info`、`/get_info`、`/delete_session`、`/chat` 四个接口在
> `AGENT_BACKEND=langgraph` 下接入应用服务 + repository 后的行为差异。
> FastGPT 模式(`AGENT_BACKEND=fastgpt`)保留旧实现,不受影响。
## 接口响应契约(保持不变)
| 项 | 现状(FastGPT) | LangGraph(作业 6) |
|---|---|---|
| 请求/响应模型 | `ProcessRequest/Response_*` | 相同(endpoint 未改模型) |
| `get_info` 分组名 | `all` / `acdinfo` / `acdhuman1` / `acdhuman2` | 相同 |
| `get_info` 布尔外部表示 | `True→"1"`,`False→"0"`,未知→`""` | 相同(`to_external_string`) |
| `get_info` 未知 key | 返回 JSON 空串 `'""'` | **保持** `'""'`(现状兼容) |
| `includeInputInfo` 字段 | 触发 FastGPT 辅助对话 | 继续接受但**忽略**——新后端零辅助对话 |
## 有意的行为差异(新后端收紧)
| 项 | FastGPT 现状 | LangGraph | 理由 |
|---|---|---|---|
| `set_info` 未知 key | 任意 key 写入 state 变量 | **拒绝**(500 + "入参不规范:未知或受保护的字段 key") | 白名单保护;任意写入会污染图输入。上线前需审计旧调用方是否依赖任意 key 写入 |
| `set_info` 图内部键(`stage_code` 等) | 可写(state 是裸 dict) | **拒绝**(`INTERNAL_FIELD_KEYS` 保护) | 内部状态只允许图自己推进 |
| `set_info` 字段值 | 任意字符串原样存 | 按注册表类型解析(bool/int 严格),无法解析拒绝 | 保留 空/False/0/未知 区分,杜绝 `bool("0")` 陷阱 |
| LLM 调用 | set/get 走辅助对话 + 事后删记录 | **零 LLM 调用**(测试 spy 保证) | 成本与延迟;删除 race 消失 |
| 重复 `delete_session` | FastGPT SDK 行为未固定 | **幂等成功**(业务状态 + checkpoint 均静默容忍缺失) | 行为固定为测试 |
## 待审计项(作业 6-7 要求的后续工作)
- [ ] 审计现有调用方是否存在对 `set_info` 任意 key 写入的依赖
(如动态自定义键)。存在则在注册表补登记或提供逃生口。
- [ ] 流式 `chat` 的 `formUpdate` 差量写回 repository:
骨架阶段流式原样透传,差量写回仅在非流式 `chat()` 生效;
完整版随 FormUpdate 事件捕获补齐(见 `app_service.stream` 注释)。
- [ ] `get_info` 返回值中的敏感字段(证件号/手机号)回显边界:
注册表已记录 `SensitiveLevel`,访问控制随作业 9 落地。
## 验证
- `test/services/test_fields_registry.py`:注册表完整性、类型/敏感级别、
外部表示、`bool("0")` 防线。
- `test/services/test_app_service.py`:四接口组合(set→get→chat 注入→
差量写回→分组读→删除→无泄漏),set/get/delete 零模型调用 spy,
重复删除幂等。
- `test/api/test_assignment6_endpoints.py`:HTTP 层验收演示链路。
+68 -10
View File
@@ -11,7 +11,12 @@ from ..backends.chat import (
FormUpdate,
TextDelta,
)
from ..core.fastgpt_client import get_chat_backend, get_fastgpt_client
from ..core.fastgpt_client import (
delete_chat_session,
get_app_service,
get_chat_backend,
get_fastgpt_client,
)
from ..core.config import Config
from ..utils.text_chunker import SentenceTextChunker, SentenceTextChunkerConfig
from loguru import logger
@@ -131,7 +136,11 @@ async def chat(
stream: bool = False,
backend: ChatBackend = Depends(get_chat_backend)
):
"""Handle chat completion request."""
"""Handle chat completion request.
作业 6:langgraph 模式由应用服务承载业务状态同步
(repository 字段注入 + 差量写回),endpoint 仍负责协议解析。
"""
json_data = request.model_dump()
client_mode = json_data['clientMode']
need_form_update = json_data.get('needFormUpdate', False)
@@ -142,6 +151,14 @@ async def chat(
need_form_update=need_form_update,
client_mode=client_mode,
)
# 应用服务:langgraph 模式下包装字段注入/写回;fastgpt 模式透传。
# 单测直调函数(未走 lifespan)时回退 backend 直调。
app_service = None
if Config.AGENT_BACKEND == "langgraph":
try:
app_service = get_app_service()
except RuntimeError:
app_service = None
request_started_at = time.perf_counter()
logger.info(
"Chat request received "
@@ -250,7 +267,11 @@ async def chat(
state_filter_buffer = ""
return cleaned
async for event in backend.stream(chat_input):
async for event in (
app_service.stream(chat_input)
if app_service is not None
else backend.stream(chat_input)
):
try:
if not first_event_logged:
first_event_logged = True
@@ -370,7 +391,10 @@ async def chat(
return StreamingResponse(event_generator(), media_type="text/event-stream")
try:
result = await backend.complete(chat_input)
if app_service is not None:
result = await app_service.chat(chat_input)
else:
result = await backend.complete(chat_input)
logger.info(
"Chat backend non-stream response received "
f"sessionId={json_data['sessionId']} "
@@ -485,10 +509,28 @@ async def chat(
@router.post("/set_info", response_model=ProcessResponse_set)
async def set_info(
request: ProcessRequest_set,
client: AsyncChatClient = Depends(get_fastgpt_client),
client=None, # fastgpt 模式使用;注解省略(FastAPI 会解析参数注解)
):
"""Set information in chat state."""
"""Set information in chat state.
作业 6:langgraph 模式走应用服务 + repository——零 LLM 调用、
零辅助对话;includeInputInfo 继续接受但不再触发辅助对话。
fastgpt 模式保留旧实现。
"""
json_data = request.model_dump()
if Config.AGENT_BACKEND == "langgraph":
code, msg = get_app_service().set_info(
json_data['sessionId'], json_data['key'], json_data['value']
)
return ProcessResponse_set(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
code=code,
msg=msg or None,
)
client = client or get_fastgpt_client()
set_info_payload = {'key': json_data['key'], 'value': json_data['value']}
set_info_input = format_set_info_input(
set_info_payload,
@@ -577,10 +619,25 @@ async def set_info(
@router.post("/get_info", response_model=ProcessResponse_get)
async def get_info(
request: ProcessRequest_get,
client: AsyncChatClient = Depends(get_fastgpt_client),
client=None, # fastgpt 模式使用;注解省略(FastAPI 会解析参数注解)
):
"""Get information from chat state."""
"""Get information from chat state.
作业 6:langgraph 模式走应用服务 + repository——零 LLM 调用;
分组与单键读法兼容现有契约,未知 key 返回 JSON 空串。
"""
json_data = request.model_dump()
if Config.AGENT_BACKEND == "langgraph":
return ProcessResponse_get(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
value=get_app_service().get_info(json_data['sessionId'], json_data['key']),
code="200",
msg="",
)
client = client or get_fastgpt_client()
get_info_payload = {'key': json_data['key']}
get_info_input = format_set_info_input(
get_info_payload,
@@ -700,9 +757,10 @@ async def delete_session(
)
if Config.AGENT_BACKEND == "langgraph":
from ..core.fastgpt_client import delete_chat_session
try:
await delete_chat_session(chat_id)
await get_app_service().delete_session(
chat_id, drop_checkpoint=delete_chat_session
)
return ProcessResponse_delete_session(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
+4
View File
@@ -13,6 +13,10 @@ class ChatInput:
text: str
need_form_update: bool = False
client_mode: str = ""
# 作业 6:应用服务注入的最新业务字段(acd 进 accident_fields,
# 人员进 person_fields)。LangGraph 后端合并进图输入;
# FastGPT 后端忽略该字段。
injected_fields: dict | None = None
@dataclass(frozen=True)
+10 -5
View File
@@ -49,12 +49,17 @@ class LangGraphBackend(ChatBackend):
- 本阶段不处理 ``need_form_update`` 的真实表单差量
(作业 6 起实现)。
"""
graph_input: dict = {
"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)
result = await self._graph.ainvoke(
{
"session_id": chat_input.session_id,
"input_text": chat_input.text,
"client_mode": chat_input.client_mode,
},
graph_input,
config={"configurable": {"thread_id": chat_input.session_id}},
context=GraphDeps(llm=self._llm),
)
+16 -1
View File
@@ -6,12 +6,15 @@ 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 .config import Config
_fastgpt_client: AsyncChatClient | None = None
_chat_backend: ChatBackend | None = None
_langgraph_graph = None
_app_service: AccidentAppService | None = None
@asynccontextmanager
@@ -25,9 +28,13 @@ async def lifespan(app: FastAPI):
# TODO(作业 2): 若语音仍启用,需在此单独初始化并校验 FastGPT 配置。
from ..agent.llm import build_llm_service
global _langgraph_graph
global _langgraph_graph, _app_service
_langgraph_graph = build_graph()
_chat_backend = LangGraphBackend(_langgraph_graph, llm=build_llm_service())
# 作业 6:应用服务 + 内存 repository(作业 7 换 PostgreSQL)。
_app_service = AccidentAppService(
_chat_backend, InMemorySessionRepository()
)
else:
_fastgpt_client = AsyncChatClient(
api_key=Config.FASTGPT_API_KEY,
@@ -45,6 +52,7 @@ async def lifespan(app: FastAPI):
finally:
_chat_backend = None
_langgraph_graph = None
_app_service = None
if _fastgpt_client:
await _fastgpt_client.__aexit__(None, None, None)
_fastgpt_client = None
@@ -68,6 +76,13 @@ def get_fastgpt_client() -> AsyncChatClient:
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")
return _app_service
def get_chat_backend() -> ChatBackend:
"""Get the backend-neutral chat service."""
if _chat_backend is None:
+5
View File
@@ -0,0 +1,5 @@
"""应用服务层(作业 6):应用服务、字段注册表、会话 repository。
endpoint 只处理校验、协议(SSE/响应模型)和依赖注入;
业务状态读写、字段白名单与分组语义集中在本层。
"""
+152
View File
@@ -0,0 +1,152 @@
"""事故处理应用服务(作业 6-1):四业务接口共享同一份业务状态。
职责边界(对应计划作业 6):
- set_info / get_info / delete_session **不调用 LLM、不创建辅助对话**
(对照旧 FastGPT 实现的“辅助对话 + 删记录”hack);
- chat 每轮从 repository 读最新业务字段注入图(set_info 写入后
下一轮 chat 必须可见),图产出字段差量后写回 repository;
- 未知 key 兼容策略:get 对未知 key 返回 JSON 空串 '""'(现状兼容);
set 对未知 key 拒绝写入(白名单保护,差异见兼容清单);
- includeInputInfo 继续被 endpoint 接受,但本层不再使用
——新后端无需触发辅助对话。
endpoint 只负责校验、协议(SSE / 响应模型)与依赖注入。
"""
from ..backends.chat import ChatBackend, ChatInput, ChatResult
from .fields_registry import (
FIELD_REGISTRY,
GROUP_KEYS,
FieldGroup,
parse_field_value,
to_external_string,
)
from .repository import SessionRepository
# set_info 对未知/受保护 key 的拒绝码(与现有“入参不规范”文案对齐)
SET_INFO_INVALID_KEY_MSG = "入参不规范:未知或受保护的字段 key"
class AccidentAppService:
"""四个业务接口的应用服务(langgraph 后端)。"""
def __init__(self, backend: ChatBackend, repository: SessionRepository):
self._backend = backend
self._repository = repository
# ---- chat:图每轮读取最新业务状态,产出差量写回 ----
def _injected_fields(self, session_id: str) -> dict:
"""repository 业务字段 -> 图输入的分组注入(acd 进 accident_fields,
两组人员进 person_fields,与 AgentState 的键结构对齐)。"""
business = self._repository.load(session_id)
if not business:
return {}
injected: dict = {"accident_fields": {}, "person_fields": {}}
for key, value in business.items():
spec = FIELD_REGISTRY.get(key)
if spec is None:
continue
if spec.group is FieldGroup.ACD:
injected["accident_fields"][key] = value
else:
injected["person_fields"][key] = value
return injected
async def chat(self, chat_input: ChatInput) -> ChatResult:
"""非流式一轮对话:注入最新业务字段 -> 执行图 -> 写回字段差量。"""
injected = self._injected_fields(chat_input.session_id)
result = await self._backend.complete(
ChatInput(
session_id=chat_input.session_id,
text=chat_input.text,
need_form_update=chat_input.need_form_update,
client_mode=chat_input.client_mode,
injected_fields=injected or None,
)
)
# 图产出的字段差量(turn_patch / 完整字段)写回 repository,
# 使 set/get_info 与后续轮次看到同一事实来源
patch = result.form_update if isinstance(result.form_update, dict) else {}
if patch:
self._repository.apply_update(chat_input.session_id, patch)
return result
async def stream(self, chat_input: ChatInput):
"""流式代理:字段注入与写回语义同 chat(),事件流原样透传。"""
injected = self._injected_fields(chat_input.session_id)
result_input = ChatInput(
session_id=chat_input.session_id,
text=chat_input.text,
need_form_update=chat_input.need_form_update,
client_mode=chat_input.client_mode,
injected_fields=injected or None,
)
# 骨架阶段原样透传事件流;字段差量写回在作业 6 完整版随
# FormUpdate 事件捕获补齐(对照计划作业 8 的真实增量)。
async for event in self._backend.stream(result_input):
yield event
# ---- set_info / get_info:零 LLM、零辅助对话 ----
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
try:
parsed = parse_field_value(spec, value)
except ValueError:
return "500", "入参不规范:字段值无法解析"
self._repository.apply_update(session_id, {key: parsed})
return "200", ""
def get_info(self, session_id: str, key: str) -> str:
"""按分组或单键读出,返回 JSON 字符串(兼容现有 /get_info 契约)。"""
import json
business = self._repository.load(session_id)
if key == "all":
return json.dumps(
{
group.value: {
k: to_external_string(business.get(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.get(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.get(key)), ensure_ascii=False)
# ---- delete_session:业务状态 + 图 checkpoint 一起清理 ----
async def delete_session(self, session_id: str, drop_checkpoint) -> None:
"""删除会话全部数据。
drop_checkpoint:由 core 层注入的 checkpoint 清理函数
(图生命周期归 lifespan 管理,本层不持有图引用)。
repository 删除幂等;checkpoint 缺失时静默(幂等语义一致)。
"""
self._repository.delete(session_id)
if drop_checkpoint is not None:
try:
await drop_checkpoint(session_id)
except Exception:
# checkpoint 已不存在视为删除成功的一部分(幂等)
pass
__all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"]
+144
View File
@@ -0,0 +1,144 @@
"""字段注册表(作业 6-3/6-4):外部业务键 -> 类型/分组/可写性/敏感级别。
完整覆盖当前 get_info 代码中的三个分组:
- 事故字段(acdinfo):ywrysw/ywfjdc/ywmtc/bjrjs/sgfssj/sfsgxc/jdcsl/sgyy
- 本方人员(acdhuman1):xm1/hpzl1/hphm1/sfzmhm1/sfzmwh1/sjhm1/sjwh1/csbw1
- 对方人员(acdhuman2):同上一组,序号 2
设计约束:
- 外部传输一律字符串;内部按 value_type 存储强类型值,
保留区分 空值 / False / 0 / 未知 的能力(不做 bool("0") 转换);
- 敏感级别用于作业 9 的访问边界与日志脱敏,注册表先行记录;
- get 的分组名(acdinfo/acdhuman1/acdhuman2)与外部接口契约保持一致。
"""
from dataclasses import dataclass
from enum import StrEnum
from typing import Any
class FieldGroup(StrEnum):
"""get_info 对外分组名(兼容现有 /get_info 契约)。"""
ACD = "acdinfo"
HUMAN1 = "acdhuman1"
HUMAN2 = "acdhuman2"
class SensitiveLevel(StrEnum):
"""敏感级别(作业 9 用于访问边界/脱敏)。"""
PUBLIC = "public" # 无个人信息
PERSONAL = "personal" # 姓名等一般个人信息
SENSITIVE = "sensitive" # 证件号/手机号,日志与回显需保护
@dataclass(frozen=True)
class FieldSpec:
"""一个外部业务字段的注册信息。"""
key: str
group: FieldGroup
value_type: type # str / bool / int
writable: bool
sensitive: SensitiveLevel
description: str = ""
def _acd_fields() -> list[FieldSpec]:
return [
FieldSpec("ywrysw", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否有人受伤"),
FieldSpec("ywfjdc", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否撞非机动车"),
FieldSpec("ywmtc", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否撞摩托车"),
FieldSpec("bjrjs", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "报警人是否驾驶员"),
FieldSpec("sgfssj", FieldGroup.ACD, str, True, SensitiveLevel.PUBLIC, "事故发生时间"),
FieldSpec("sfsgxc", FieldGroup.ACD, bool, True, SensitiveLevel.PUBLIC, "是否在现场"),
FieldSpec("jdcsl", FieldGroup.ACD, int, True, SensitiveLevel.PUBLIC, "机动车数量"),
FieldSpec("sgyy", FieldGroup.ACD, str, True, SensitiveLevel.PUBLIC, "事故原因"),
]
def _human_fields(index: int) -> list[FieldSpec]:
group = FieldGroup.HUMAN1 if index == 1 else FieldGroup.HUMAN2
side = "本方" if index == 1 else "对方"
return [
FieldSpec(f"xm{index}", group, str, True, SensitiveLevel.PERSONAL, f"{side}姓名"),
FieldSpec(f"hpzl{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}号牌种类"),
FieldSpec(f"hphm{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}号牌号码"),
FieldSpec(f"sfzmhm{index}", group, str, True, SensitiveLevel.SENSITIVE, f"{side}身份证明号码"),
FieldSpec(f"sfzmwh{index}", group, str, True, SensitiveLevel.SENSITIVE, f"{side}身份证明文号"),
FieldSpec(f"sjhm{index}", group, str, True, SensitiveLevel.SENSITIVE, f"{side}手机号码"),
FieldSpec(f"sjwh{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}手机文号"),
FieldSpec(f"csbw{index}", group, str, True, SensitiveLevel.PUBLIC, f"{side}常熟部位/身份标识"),
]
# 注册表:key -> FieldSpec(构建一次,只读共享)
FIELD_REGISTRY: dict[str, FieldSpec] = {
spec.key: spec for spec in [*_acd_fields(), *_human_fields(1), *_human_fields(2)]
}
# 分组 -> 组内有序字段键(保持当前 get_info 的键顺序,输出稳定)
GROUP_KEYS: dict[FieldGroup, tuple[str, ...]] = {
group: tuple(spec.key for spec in FIELD_REGISTRY.values() if spec.group == group)
for group in FieldGroup
}
# 内部受保护字段:注册表之外、图内部使用的键,set_info 不得写入
# (兼容策略见 docs/assignment6-compat.md)
INTERNAL_FIELD_KEYS = frozenset(
{
"stage_code", "raw_status_code", "turn_count", "client_event",
"clarify_count", "no_reply_count", "state_version",
}
)
def parse_field_value(spec: FieldSpec, raw: str) -> Any:
"""外部字符串 -> 内部强类型值;无法解析时抛 ValueError。
bool 只接受明确的 1/0/true/false/是/否;拒绝空串与模糊值,
避免 bool("0") == True 这类陷阱。int 必须是纯数字字面量。
"""
if raw is None:
raise ValueError(f"{spec.key}: 值为空")
text = raw.strip()
if spec.value_type is bool:
lowered = text.lower()
if lowered in ("1", "true", "是"):
return True
if lowered in ("0", "false", "否"):
return False
raise ValueError(f"{spec.key}: 无法解析为布尔值: {raw!r}")
if spec.value_type is int:
if text.isdigit():
return int(text)
raise ValueError(f"{spec.key}: 无法解析为整数: {raw!r}")
return raw # str 原样存储(空串视为有效的外部空值)
def to_external_string(value: Any) -> str:
"""内部值 -> 外部字符串表示(兼容现有 get_info 的 bool_to_str)。
True -> "1",False -> "0",None/未知 -> "",其余 str() 原样。
内部仍保留 False 与 0 的区分,仅在对外序列化时收敛。
"""
if value is True:
return "1"
if value is False:
return "0"
if value is None:
return ""
return str(value)
__all__ = [
"FIELD_REGISTRY",
"GROUP_KEYS",
"INTERNAL_FIELD_KEYS",
"FieldGroup",
"FieldSpec",
"SensitiveLevel",
"parse_field_value",
"to_external_string",
]
+52
View File
@@ -0,0 +1,52 @@
"""会话业务状态 repository(作业 6-1/6-2)。
抽象出 repository 接口,使应用服务不依赖具体存储;
作业 6 用内存实现跑通四接口,作业 7 再替换为 PostgreSQL 实现
(对话 checkpoint 已由 LangGraph checkpointer 承担,本层只管
业务字段——单一事实来源,图每轮读取最新值)。
语义约定:
- 业务状态按外部业务键(hphm1、ywrysw...)扁平存储,值为内部强类型;
- 不存在的会话视为空状态(读取返回 {}),与首轮 checkpoint 无字段一致;
- delete 幂等:重复删除固定成功,不抛异常(行为进测试)。
"""
from typing import Protocol
class SessionRepository(Protocol):
"""业务字段存储接口(作业 7 由 PostgreSQL 实现替换)。"""
def load(self, session_id: str) -> dict:
"""读取会话业务字段;无记录时返回空 dict。"""
...
def apply_update(self, session_id: str, patch: dict) -> dict:
"""合并差量并返回合并后的完整业务字段。"""
...
def delete(self, session_id: str) -> None:
"""删除会话业务状态;幂等,重复删除不抛异常。"""
...
class InMemorySessionRepository:
"""内存实现:字典存储,仅用于单进程开发与测试。"""
def __init__(self) -> None:
self._store: dict[str, dict] = {}
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:
state = self._store.setdefault(session_id, {})
state.update(patch)
return dict(state)
def delete(self, session_id: str) -> None:
# 幂等:pop 的缺省值吞掉不存在的情况,重复删除行为固定为成功
self._store.pop(session_id, None)
__all__ = ["SessionRepository", "InMemorySessionRepository"]
+106
View File
@@ -0,0 +1,106 @@
"""作业 6 验收演示的 HTTP 层组合测试。
链路(对应计划作业 6):创建会话 -> /set_info 写车牌 -> /get_info 读回
-> /chat 继续核实 -> 分组读值 -> /delete_session -> 再读与再聊不泄漏。
fake LLM(LLM_PROVIDER=fake)保证本测试离线可跑、零真实模型调用。
"""
import json
import pytest
from fastapi.testclient import TestClient
from src.agent.llm import workflow_fake_llm
from src.core.config import Config
from src.main import app
@pytest.fixture()
def client(monkeypatch):
"""langgraph 后端 + fake LLM 的测试客户端(lifespan 完整初始化)。"""
monkeypatch.setattr(Config, "AGENT_BACKEND", "langgraph")
monkeypatch.setattr(Config, "LLM_PROVIDER", "fake")
import src.agent.llm as llm_module
monkeypatch.setattr(
llm_module, "build_llm_service", lambda *a, **k: workflow_fake_llm()
)
with TestClient(app) as test_client:
yield test_client
TS = "20261022120000"
def test_assignment6_acceptance_flow(client):
session = "a6-acceptance-001"
# set_info 写入车牌(零 LLM:fake 未注入任何 responder,调用即报错)
response = client.post("/set_info", json={
"sessionId": session, "timeStamp": TS,
"key": "hphm1", "value": "苏B12345",
})
assert response.status_code == 200
assert response.json()["code"] == "200"
# get_info 单键读回
response = client.post("/get_info", json={
"sessionId": session, "timeStamp": TS, "key": "hphm1",
})
assert response.json()["value"] == '"苏B12345"'
# chat 使用业务状态继续(非流式,协议层解析 state 前缀)
response = client.post("/chat", json={
"sessionId": session, "timeStamp": TS,
"text": "我是刚才报警的驾驶员", "clientMode": "direct",
})
body = response.json()
assert body["code"] == "200"
assert body["nextStageCode"] in ("1001", "1002")
# 分组读值:三个分组齐全,未填字段为空串
response = client.post("/get_info", json={
"sessionId": session, "timeStamp": TS, "key": "all",
})
groups = json.loads(response.json()["value"])
assert set(groups) == {"acdinfo", "acdhuman1", "acdhuman2"}
assert groups["acdhuman1"]["hphm1"] == "苏B12345"
assert groups["acdhuman1"]["xm1"] == ""
# 未知 key 兼容:JSON 空串
response = client.post("/get_info", json={
"sessionId": session, "timeStamp": TS, "key": "no_such_key",
})
assert response.json()["value"] == '""'
# set_info 未知 key:白名单拒绝
response = client.post("/set_info", json={
"sessionId": session, "timeStamp": TS,
"key": "no_such_key", "value": "x",
})
assert response.json()["code"] == "500"
# 删除 -> 再读为空
response = client.request("DELETE", "/delete_session", json={
"sessionId": session, "timeStamp": TS,
})
assert response.json()["code"] == "200"
response = client.post("/get_info", json={
"sessionId": session, "timeStamp": TS, "key": "hphm1",
})
assert response.json()["value"] == '""'
# 重复删除幂等成功
response = client.request("DELETE", "/delete_session", json={
"sessionId": session, "timeStamp": TS,
})
assert response.json()["code"] == "200"
# 删除后再聊天:不注入旧业务数据,服务正常应答
response = client.post("/chat", json={
"sessionId": session, "timeStamp": TS,
"text": "新对话", "clientMode": "direct",
})
assert response.json()["code"] == "200"
+192
View File
@@ -0,0 +1,192 @@
"""作业 6:应用服务四接口组合测试。
验收演示对应(计划作业 6):
创建会话 -> set_info 写车牌 -> get_info 读回 -> chat 使用车牌继续核实
-> 分组读值 -> 删除 -> 再读与再次聊天均不泄漏旧数据。
spy 断言:set_info / get_info / delete_session 全程零模型调用。
"""
import json
import pytest
from src.backends.chat import ChatResult, TextDelta, FormUpdate
from src.services.app_service import AccidentAppService, SET_INFO_INVALID_KEY_MSG
from src.services.repository import InMemorySessionRepository
class SpyBackend:
"""记录调用的 ChatBackend 替身;form_update 模拟图产出的字段差量。"""
def __init__(self, form_update=None):
self.calls = [] # complete 收到的 ChatInput
self.stream_calls = []
self._form_update = form_update or {}
# 用调用次数当“模型调用”spy:真实 backend 每次 complete 至少一次 LLM
self.llm_calls = 0
async def complete(self, chat_input):
self.calls.append(chat_input)
self.llm_calls += 1
return ChatResult(
content="<state>1002</state>收到",
status_code=None,
form_update=dict(self._form_update),
)
async def stream(self, chat_input):
self.stream_calls.append(chat_input)
self.llm_calls += 1
yield TextDelta("<state>1002</state>")
if self._form_update:
yield FormUpdate(dict(self._form_update))
yield TextDelta("收到")
def make_service(form_update=None):
backend = SpyBackend(form_update=form_update)
service = AccidentAppService(backend, InMemorySessionRepository())
return service, backend
# ---- set/get:白名单、分组、外部表示 ----
def test_set_then_get_single_key_roundtrip():
service, backend = make_service()
code, msg = service.set_info("s1", "hphm1", "苏B12345")
assert (code, msg) == ("200", "")
assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345"
# set/get 零模型调用
assert backend.llm_calls == 0
def test_get_groups_match_existing_contract():
service, _ = make_service()
service.set_info("s1", "ywrysw", "0") # False
service.set_info("s1", "jdcsl", "0") # int 0
service.set_info("s1", "xm2", "张三")
acd = json.loads(service.get_info("s1", "acdinfo"))
assert acd["ywrysw"] == "0" # False -> "0"
assert acd["jdcsl"] == "0" # int 0 -> "0"(内部类型不同)
assert acd["sgyy"] == "" # 未知 -> 空串
human2 = json.loads(service.get_info("s1", "acdhuman2"))
assert human2["xm2"] == "张三"
assert human2["sfzmwh2"] == "" # 全字段齐全,未填为空
all_groups = json.loads(service.get_info("s1", "all"))
assert set(all_groups) == {"acdinfo", "acdhuman1", "acdhuman2"}
def test_get_unknown_key_returns_json_empty_string():
"""兼容现状:未知 key 返回 JSON 空串 '""'。"""
service, _ = make_service()
service.set_info("s1", "hphm1", "苏B12345")
assert service.get_info("s1", "no_such_key") == '""'
assert service.get_info("ghost-session", "hphm1") == '""'
def test_set_unknown_or_internal_key_rejected():
service, backend = make_service()
code, msg = service.set_info("s1", "no_such_key", "x")
assert code == "500" and msg == SET_INFO_INVALID_KEY_MSG
code, _ = service.set_info("s1", "stage_code", "1002")
assert code == "500" # 内部字段保护
assert service.get_info("s1", "hphm1") == '""' # 未被污染
assert backend.llm_calls == 0
def test_set_rejects_unparsable_value():
service, _ = make_service()
code, msg = service.set_info("s1", "ywrysw", "maybe")
assert code == "500" and "无法解析" in msg
# ---- chat:注入 + 写回 ----
@pytest.mark.asyncio
async def test_chat_sees_set_info_values_and_writes_back_patch():
"""set_info 写入 -> chat 输入注入 -> 图差量写回 -> get_info 读到。"""
service, backend = make_service(form_update={"ywrysw": False})
service.set_info("s1", "hphm1", "苏B12345")
service.set_info("s1", "xm2", "李四")
result = await service.chat(
_input("s1", "我是刚才报警的驾驶员")
)
assert result.content == "<state>1002</state>收到"
# 图输入必须携带 repository 最新业务字段(车牌供核实使用)
injected = backend.calls[0].injected_fields
assert injected["person_fields"]["hphm1"] == "苏B12345"
assert injected["person_fields"]["xm2"] == "李四"
# 图产出的差量写回 repository,get_info 立即可读
assert json.loads(service.get_info("s1", "acdinfo"))["ywrysw"] == "0"
assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345"
@pytest.mark.asyncio
async def test_chat_without_business_state_passes_no_injection():
service, backend = make_service()
await service.chat(_input("fresh-session", "你好"))
assert backend.calls[0].injected_fields is None
@pytest.mark.asyncio
async def test_stream_injects_fields_and_passthrough_events():
service, backend = make_service(form_update={"ywrysw": False})
service.set_info("s1", "xm1", "王五")
events = [
event async for event in service.stream(_input("s1", "继续办理"))
]
kinds = [type(event).__name__ for event in events]
assert kinds == ["TextDelta", "FormUpdate", "TextDelta"]
assert backend.stream_calls[0].injected_fields["person_fields"]["xm1"] == "王五"
# ---- delete:清理 + 幂等 ----
@pytest.mark.asyncio
async def test_delete_clears_business_state_and_is_idempotent():
drops = []
async def drop_checkpoint(session_id):
drops.append(session_id)
service, backend = make_service()
service.set_info("s1", "hphm1", "苏B12345")
await service.chat(_input("s1", "你好"))
await service.delete_session("s1", drop_checkpoint=drop_checkpoint)
assert drops == ["s1"]
# 再读为空、再聊天不注入旧数据
assert service.get_info("s1", "hphm1") == '""'
await service.chat(_input("s1", "还在吗"))
assert backend.calls[-1].injected_fields is None
# 重复删除固定行为:成功返回;checkpoint 清理会再次触发,
# 但缺失时静默容忍(幂等语义由 drop_checkpoint 内部保证)
await service.delete_session("s1", drop_checkpoint=drop_checkpoint)
assert drops == ["s1", "s1"]
assert backend.llm_calls == 2 # 两次 chat 各一次;set/get/delete 零调用
@pytest.mark.asyncio
async def test_delete_tolerates_missing_checkpoint():
async def missing(_session_id):
raise RuntimeError("no such thread")
service, _ = make_service()
service.set_info("s1", "xm1", "王五")
# checkpoint 不存在视为删除成功的一部分,不向上抛
await service.delete_session("s1", drop_checkpoint=missing)
assert service.get_info("s1", "xm1") == '""'
def _input(session_id, text):
from src.backends.chat import ChatInput
return ChatInput(session_id=session_id, text=text, client_mode="direct")
+89
View File
@@ -0,0 +1,89 @@
"""作业 6:字段注册表测试(分组完整性、类型/敏感级别、外部表示)。"""
import pytest
from src.services.fields_registry import (
FIELD_REGISTRY,
GROUP_KEYS,
INTERNAL_FIELD_KEYS,
FieldGroup,
SensitiveLevel,
parse_field_value,
to_external_string,
)
def test_registry_covers_all_three_groups():
"""注册表完整覆盖 acd/human1/human2 三个分组(作业 6-3)。"""
assert set(GROUP_KEYS) == set(FieldGroup)
assert len(FIELD_REGISTRY) == 24
assert len(GROUP_KEYS[FieldGroup.ACD]) == 8
assert len(GROUP_KEYS[FieldGroup.HUMAN1]) == 8
assert len(GROUP_KEYS[FieldGroup.HUMAN2]) == 8
def test_person_fields_include_full_set():
"""人员字段必须包含 sfzmwh/sjwh/csbw,不能只迁姓名和号码。"""
for index in (1, 2):
group_keys = set(GROUP_KEYS[FieldGroup.HUMAN1 if index == 1 else FieldGroup.HUMAN2])
for suffix in ("xm", "hpzl", "hphm", "sfzmhm", "sfzmwh", "sjhm", "sjwh", "csbw"):
assert f"{suffix}{index}" in group_keys
def test_registry_metadata_recorded():
"""每个字段都记录类型、可写性、敏感级别(作业 6-4)。"""
for spec in FIELD_REGISTRY.values():
assert spec.value_type in (str, bool, int)
assert spec.writable is True
assert isinstance(spec.sensitive, SensitiveLevel)
assert spec.description
def test_sensitive_levels_assigned():
"""证件号/手机号 sensitive,姓名 personal,其余 public。"""
assert FIELD_REGISTRY["sfzmhm1"].sensitive is SensitiveLevel.SENSITIVE
assert FIELD_REGISTRY["sjhm2"].sensitive is SensitiveLevel.SENSITIVE
assert FIELD_REGISTRY["xm1"].sensitive is SensitiveLevel.PERSONAL
assert FIELD_REGISTRY["hphm1"].sensitive is SensitiveLevel.PUBLIC
@pytest.mark.parametrize("raw,expected", [
("1", True), ("true", True), ("是", True),
("0", False), ("false", False), ("否", False),
])
def test_parse_bool_explicit(raw, expected):
assert parse_field_value(FIELD_REGISTRY["ywrysw"], raw) is expected
@pytest.mark.parametrize("bad", ["", "maybe", "2", "有吧"])
def test_parse_bool_rejects_ambiguous(bad):
"""模糊值必须拒绝——保留 空/False/0/未知 的区分能力。"""
with pytest.raises(ValueError):
parse_field_value(FIELD_REGISTRY["ywrysw"], bad)
def test_parse_int_and_str():
assert parse_field_value(FIELD_REGISTRY["jdcsl"], "2") == 2
assert parse_field_value(FIELD_REGISTRY["jdcsl"], "0") == 0
with pytest.raises(ValueError):
parse_field_value(FIELD_REGISTRY["jdcsl"], "两辆")
assert parse_field_value(FIELD_REGISTRY["hphm1"], "苏B12345") == "苏B12345"
def test_external_string_representation():
"""对外序列化:True->1,False->0,None->空串;内部仍区分 False 与 0。"""
assert to_external_string(True) == "1"
assert to_external_string(False) == "0"
assert to_external_string(0) == "0"
assert to_external_string(None) == ""
assert to_external_string("苏B12345") == "苏B12345"
# 内部存储不混淆:False 是 bool、0 是 int
assert parse_field_value(FIELD_REGISTRY["ywrysw"], "0") is False
assert parse_field_value(FIELD_REGISTRY["jdcsl"], "0") == 0
assert isinstance(parse_field_value(FIELD_REGISTRY["jdcsl"], "0"), int)
def test_internal_field_protection_listed():
"""图内部键登记在保护清单中,set_info 白名单不会收录它们。"""
for key in INTERNAL_FIELD_KEYS:
assert key not in FIELD_REGISTRY