From 24e2cfbe102c4239c99b95d9f2fb9320508b9810 Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Fri, 23 Oct 2026 15:54:51 +0800 Subject: [PATCH] fix(backends): make streamed replies and session state consistent - Validate stage prefixes before forwarding model tokens. Invalid prefixes may use the existing repair budget; once output starts, validation failure aborts the turn without appending a regenerated reply. Bound the token queue and close inner streams when consumption stops. - Replace delete-time lock eviction with reference-counted session locks covering holders and waiters. Serialize set/get with chat/stream/delete and release cancelled waiters without replacing a live lock. - Store person synonyms under workflow-native driver keys and expose consistent alias views to single/group reads and graph prompt injection. - Separate committed field patches from public formUpdate visibility. Commit callbacks keep repository fields synchronized even when updates are hidden or the consumer closes after graph completion. - Disable OpenAI SDK automatic retries. Own bounded retry attempts in the application, use 1/2/4 second backoff, retry streams only before text is emitted, enforce attempt timeouts, and close upstream/client resources. - Translate FastGPT SDK max_retries (total attempts) from the configured extra-retry count, and verify the real SDK loop uses four attempts. - Document the per-LLM budget and streaming failure semantics without claiming the whole multi-call HTTP turn has the same timeout. Validation: 267 offline tests passed, including actual SDK requests through MockTransport, stream repair/abort, timeout/close, lock races, aliases and formUpdate visibility; existing four dependency deprecation warnings. --- .env.example | 5 +- docs/assignment6-compat.md | 34 ++- src/agent/collect.py | 11 +- src/agent/llm.py | 255 ++++++++---------- src/api/endpoints.py | 4 +- src/backends/chat.py | 6 +- src/backends/langgraph.py | 120 +++++---- src/core/fastgpt_client.py | 18 +- src/services/app_service.py | 125 +++++---- src/services/fields_registry.py | 22 +- src/services/session_locks.py | 28 ++ test/agent/test_llm_stream_tokens.py | 2 +- test/backends/test_langgraph_streaming.py | 2 +- test/backends/test_sla_alignment.py | 48 +++- test/services/test_app_service.py | 85 +++--- test/services/test_field_commit_and_retry.py | 158 +++++++++++ .../test_stream_and_session_safety.py | 151 +++++++++++ test/services/test_stream_retry_cleanup.py | 94 +++++++ 18 files changed, 852 insertions(+), 316 deletions(-) create mode 100644 src/services/session_locks.py create mode 100644 test/services/test_field_commit_and_retry.py create mode 100644 test/services/test_stream_and_session_safety.py create mode 100644 test/services/test_stream_retry_cleanup.py diff --git a/.env.example b/.env.example index 3539ca8..96dc4e1 100644 --- a/.env.example +++ b/.env.example @@ -17,8 +17,9 @@ VOICE_CONFIG=config/voice.json # 每组 LLM_<名>_BASE_URL / _API_KEY / _MODEL 必填； # 可选 _TIMEOUT_SECONDS / _MAX_RETRIES / _TEMPERATURE / _EXTRA_BODY（JSON 字符串， # 如思考开关 KIMI 网关为 {"reasoning_effort":"none"}）。 -# 超时/重试缺省 60 秒 / 3 次，与 FastGPT 客户端 SLA 对齐 -# （契约测试 test/backends/test_sla_alignment.py）。 +# 超时/重试缺省：每次尝试 60 秒，最多额外重试 3 次，退避 1、2、4 秒。 +# SDK 自动重试关闭；流式发出首个文本 token 后不重试。 +# 单次 LLM 调用预算最多 247 秒（另加响应关闭清理）；不是整轮聊天总超时。 # AGENT_HISTORY_MESSAGES=50：注入模型的对话历史上限（最近 N 条， # 双向计数）；调大支持长对话；0 = 不截断（对齐 FastGPT 全量历史）。 # AGENT_HISTORY_MESSAGES=0 diff --git a/docs/assignment6-compat.md b/docs/assignment6-compat.md index bda81d7..65c2e7c 100644 --- a/docs/assignment6-compat.md +++ b/docs/assignment6-compat.md @@ -71,22 +71,42 @@ 这修正了旧实现"workflow 真实人员数据无法分组读到"的问题； - `driver{i}License` 非 state 键，对其 set_info 按未知键拒绝 （workflow 不消费它）； -- 同义键（driver1ID vs sfzmhm1）暂不合并：合并需调用方确认 - 实际使用侧后作为后续优化，已列入审计项。 +- 同义键统一存储：`sfzmhm{i}` 与 `driver{i}ID` 共用后者， + `sjhm{i}` 与 `driver{i}Phonenumber` 共用后者。单键和分组读取 + 继续提供两套键名；图输入同时注入一致的别名视图，避免旧 driver 值 + 遮盖 API 键的新值。同一批差量包含相互冲突的别名值时拒绝提交。 ## 待审计项（作业 6-7 要求的后续工作） - [ ] 审计现有调用方是否存在对 `set_info` 任意 key 写入的依赖 （如动态自定义键）。存在则在注册表补登记或提供逃生口。 - **对账新增**：workflow 20261022 state 人员键 - （`driver1ID`/`driver1License`/`driver1Phonenumber` ×2） - 与 API 分组键不同名——确认线上调用方实际使用哪套键名， - 必要时注册表补别名。 + workflow 人员键 `driver{i}ID` / `driver{i}Phonenumber` 及对应 + API 别名均已登记并统一存储；`driver{i}License` 只是车牌的 + 输出名称，外部写车牌继续使用 `hphm{i}`。 - [x] ~~流式 `chat` 的 `formUpdate` 差量写回 repository~~ - （完整版已实现：FormUpdate 事件捕获后经映射写回，见 `app_service.stream`）。 + （图提交时即同步业务字段；FormUpdate 是否返回不影响写回）。 - [ ] `get_info` 返回值中的敏感字段（证件号/手机号）回显边界： 注册表已记录 `SensitiveLevel`，访问控制随作业 9 落地。 +## 流式、并发与字段一致性约定 + +- 阶段前缀先缓冲并按当前节点的转换规则校验，非法前缀不推送； + 未输出时仍可使用一次修复预算。已开始输出后，全文格式或时间依据 + 校验失败会以错误终止，不在同一流中拼接修复答案，不提交本轮业务字段。 + 已经到达客户端的正文不可撤回；客户端只在收到 done 后将一轮视为成功。 +- 会话锁引用包括持有者与等待者，只有最后一个使用者退出才回收。 + chat、stream、set、get、delete 全部按同一会话串行；不同会话独立。 + 取消流时先关闭内部生成器、取消图任务，再释放会话锁。 +- needFormUpdate 只决定对外是否返回差量。图提交时通过内部回调同步 + repository，关闭差量显示或消费者在提交后断开不会丢失业务字段。 +- SDK 自动重试关闭，由应用层统一控制；默认首次尝试加 3 次重试， + 每次尝试总超时 60 秒，退避 1、2、4 秒。单次 LLM 调用理论预算 + 4×60+1+2+4=247 秒（另有响应关闭的清理时间）。FastGPT SDK 的 + max_retries 实际代表总尝试数，适配层传入 4 以表达额外重试 3 次； + 其 timeout 是 HTTP SDK 超时，不将它等同于整个多节点请求总期限。 + 流式首个文本 token 前可重试，发出文本后失败则终止，不重放。 + 该预算并非整个多次 LLM 调用的聊天回合总超时。 + ## 验证 - `test/services/test_fields_registry.py`：注册表完整性、类型/敏感级别、 diff --git a/src/agent/collect.py b/src/agent/collect.py index 64890b5..572f282 100644 --- a/src/agent/collect.py +++ b/src/agent/collect.py @@ -100,11 +100,20 @@ async def validated_reply( + f"\n当前阶段为 {current}。请重新检查该阶段允许的转换" "和当前 clientMode，不得跳步或切换到另一模式的出口。" ) - raw = await llm.complete(prompt, user_message, history=history) + streaming_complete = getattr(llm, "complete_validated", None) + if streaming_complete is None: + raw = await llm.complete(prompt, user_message, history=history) + else: + raw = await streaming_complete( + prompt, user_message, history=history, + stage_validator=lambda target: validator(current, target), + ) parsed = _try_parse(raw) if (parsed is not None and validator(current, parsed[0]) and (reply_validator is None or reply_validator(parsed[0], parsed[1]))): return raw, parsed[0] + if getattr(llm, "response_started", False): + raise CollectReplyError("流式回复已开始后校验失败，已终止本轮，未提交业务状态") raise CollectReplyError("模型回复格式、阶段转换或时间依据不合规（一次修复后仍失败）") diff --git a/src/agent/llm.py b/src/agent/llm.py index 376aff9..8b2c2a7 100644 --- a/src/agent/llm.py +++ b/src/agent/llm.py @@ -87,176 +87,153 @@ class FakeLLMService: class OpenAILLMService: - """真实 LLM provider（OpenAI 兼容接口，如 OpenAI / DeepSeek 等）。 + """单层重试：首次 + max_retries；每次尝试超时覆盖整个响应读取。 - 参考实现：async 客户端 + asyncio.wait_for 超时（取消会向外传播， - 客户端断连时图任务随之取消）；仅对可重试错误（429 / 5xx / 连接错） - 按 LLM_MAX_RETRIES 指数退避重试；错误统一转换为 - LLMTimeoutError / LLMAPIError，不向上层泄漏原始响应体。 + SDK 自动重试关闭。流式仅在首个文本 token 前重试，输出后失败终止。 + 默认 60 秒、3 次重试、1/2/4 秒退避：单次 LLM 调用预算上界 247 秒 + （不含响应关闭的清理时间），不代表多次 LLM 调用的整轮 HTTP 超时。 """ - def __init__( - self, - model: str, - api_key: str, - base_url: str = "", - timeout_seconds: float = 60.0, - max_retries: int = 3, - temperature: float | None = None, - extra_body: dict | None = None, - ): + RETRY_DELAY_SECONDS = 1.0 + + def __init__(self, model: str, api_key: str, base_url: str = "", + timeout_seconds: float = 60.0, max_retries: int = 3, + temperature: float | None = None, extra_body: dict | None = None): + if timeout_seconds <= 0 or max_retries < 0: + raise ValueError("timeout_seconds must be positive; max_retries must be nonnegative") self._model = model self._api_key = api_key self._base_url = base_url self._timeout_seconds = timeout_seconds self._max_retries = max_retries self._temperature = temperature - # 附加请求体（如思考开关），原样合并进 chat.completions.create 参数。 self._extra_body = dict(extra_body or {}) + if {"model", "messages", "stream"} & self._extra_body.keys(): + raise ValueError("extra_body cannot override model, messages or stream") self._client = None def _get_client(self): - # 延迟构造：fake provider 使用时无需安装/初始化真实客户端。 - # AsyncOpenAI 的创建不绑定事件循环，可在首次调用前安全缓存。 - # trust_env=False：不走系统代理，本地网关直连（代理曾导致 503/500）。 if self._client is None: import httpx from openai import AsyncOpenAI - from ..core.config import Config - kwargs = {"api_key": self._api_key, "timeout": self._timeout_seconds} + options = { + "api_key": self._api_key, "timeout": self._timeout_seconds, + "max_retries": 0, + "http_client": httpx.AsyncClient(trust_env=Config.LLM_HTTP_TRUST_ENV), + } if self._base_url: - kwargs["base_url"] = self._base_url - kwargs["http_client"] = httpx.AsyncClient(trust_env=Config.LLM_HTTP_TRUST_ENV) - self._client = AsyncOpenAI(**kwargs) + options["base_url"] = self._base_url + self._client = AsyncOpenAI(**options) return self._client - async def complete( - self, system_prompt: str, user_message: str, *, - history: list[dict[str, str]] | None = None, - ) -> str: - import asyncio + async def aclose(self): + if self._client is not None: + await self._client.close() + self._client = None - from openai import ( - APIConnectionError, - APITimeoutError, - AuthenticationError, - RateLimitError, + def _request_args(self, system_prompt, user_message, history, *, stream=False): + options = { + "model": self._model, + "messages": [ + {"role": "system", "content": system_prompt}, + *[dict(message) for message in (history or [])], + {"role": "user", "content": user_message}, + ], + } + if stream: + options["stream"] = True + if self._temperature is not None: + options["temperature"] = self._temperature + if self._extra_body: + options["extra_body"] = dict(self._extra_body) + return options + + @staticmethod + def _retryable(exc): + from openai import APIConnectionError, APIStatusError + + return ( + isinstance(exc, (TimeoutError, APIConnectionError)) + or isinstance(exc, APIStatusError) + and (exc.status_code in (408, 409, 429) or exc.status_code >= 500) ) + def _error(self, exc): + from openai import APITimeoutError, AuthenticationError, RateLimitError + + if isinstance(exc, AuthenticationError): + return LLMAuthError("LLM provider 认证失败") + if isinstance(exc, RateLimitError): + return LLMRateLimitError("LLM provider 限流（重试预算耗尽或已开始输出）") + if isinstance(exc, (TimeoutError, APITimeoutError)): + return LLMTimeoutError(f"LLM 调用超时（每次尝试 {self._timeout_seconds}s）") + if isinstance(exc, LLMError): + return exc + return LLMAPIError(f"LLM 调用失败：{type(exc).__name__}") + + async def _retry_delay(self, attempt): + import asyncio + await asyncio.sleep(self.RETRY_DELAY_SECONDS * (2 ** attempt)) + + async def complete(self, system_prompt: str, user_message: str, *, + history: list[dict[str, str]] | None = None) -> str: + import asyncio + client = self._get_client() - messages = [ - {"role": "system", "content": system_prompt}, - *[dict(message) for message in (history or [])], - {"role": "user", "content": user_message}, - ] - kwargs: dict = {"model": self._model, "messages": messages} - if self._temperature is not None: - kwargs["temperature"] = self._temperature - kwargs.update(self._extra_body) - retries = 0 - while True: + options = self._request_args(system_prompt, user_message, history) + for attempt in range(self._max_retries + 1): try: - response = await asyncio.wait_for( - client.chat.completions.create(**kwargs), - timeout=self._timeout_seconds, - ) - except (APITimeoutError, asyncio.TimeoutError) as exc: - # 超时：可重试，重试耗尽后转换为 LLMTimeoutError - if retries < self._max_retries: - retries += 1 - await asyncio.sleep(min(2 ** retries, 10)) - continue - raise LLMTimeoutError( - f"LLM 调用超时（{self._timeout_seconds}s，" - f"重试 {self._max_retries} 次后仍失败）" - ) from exc - except (RateLimitError, APIConnectionError) as exc: - # 限流 / 连接错误：可重试 - if retries < self._max_retries: - retries += 1 - await asyncio.sleep(min(2 ** retries, 10)) - continue - if isinstance(exc, RateLimitError): - # 限流耗尽：区分上报，API 层映射为 429（两后端错误码等价） - raise LLMRateLimitError( - f"LLM provider 限流（重试 {self._max_retries} 次后仍失败）" - ) from exc - raise LLMAPIError(f"LLM provider 不可用：{type(exc).__name__}") from exc - except AuthenticationError as exc: - # 401 鉴权失败：不可重试，区分上报，API 层映射为 401 - raise LLMAuthError("LLM provider 认证失败") from exc + async with asyncio.timeout(self._timeout_seconds): + response = await client.chat.completions.create(**options) + content = response.choices[0].message.content if response.choices else None + if not content: + raise LLMAPIError("LLM 返回了空内容") + return content + except LLMError: + raise except Exception as exc: - # 其他错误（如 4xx 参数错）不可重试 - raise LLMAPIError(f"LLM 调用失败：{type(exc).__name__}") from exc + if attempt < self._max_retries and self._retryable(exc): + await self._retry_delay(attempt) + continue + raise self._error(exc) from exc - content = response.choices[0].message.content if response.choices else None - if not content: - raise LLMAPIError("LLM 返回了空内容") - return content - - async def stream_tokens( - self, system_prompt: str, user_message: str, *, - history: list[dict[str, str]] | None = None, - ) -> AsyncIterator[str]: - """逐 token 产出对话文本（作业 8 真实增量的模型侧）。 - - 语义约束（与 complete 对齐的部分）： - - 同样的消息构造（system + role-preserving history + user）、 - temperature 与 extra_body 透传； - - 同样的错误分类：401 -> LLMAuthError、429 -> LLMRateLimitError、 - 超时 -> LLMTimeoutError、其余 -> LLMAPIError。 - - 有意差异： - - **不做重试**——首个 token 发出后重试没有意义（已流出的内容 - 无法收回）；首包前的连接/鉴权错误照样按上述分类抛出， - 调用方（透传包装器）决定如何呈现； - - 超时用 asyncio.timeout 包整个迭代（末 token 迟到同样算超时）。 - """ + async def stream_tokens(self, system_prompt: str, user_message: str, *, + history: list[dict[str, str]] | None = None) -> AsyncIterator[str]: import asyncio - - from openai import ( - APIConnectionError, - AuthenticationError, - RateLimitError, - ) + from contextlib import suppress client = self._get_client() - messages = [ - {"role": "system", "content": system_prompt}, - *[dict(message) for message in (history or [])], - {"role": "user", "content": user_message}, - ] - kwargs: dict = {"model": self._model, "messages": messages, "stream": True} - if self._temperature is not None: - kwargs["temperature"] = self._temperature - kwargs.update(self._extra_body) - try: - async with asyncio.timeout(self._timeout_seconds): - response = await client.chat.completions.create(**kwargs) - async for event in response: - choices = getattr(event, "choices", None) - if not choices: - continue - delta = getattr(choices[0], "delta", None) - token = getattr(delta, "content", None) - if token: - yield token - except AuthenticationError as exc: - raise LLMAuthError("LLM provider 认证失败") from exc - except RateLimitError as exc: - raise LLMRateLimitError("LLM provider 限流") from exc - except TimeoutError as exc: - raise LLMTimeoutError( - f"LLM 流式调用超时（{self._timeout_seconds}s）" - ) from exc - except APIConnectionError as exc: - raise LLMAPIError(f"LLM provider 连接失败:{type(exc).__name__}") from exc - except LLMError: - raise - except Exception as exc: - raise LLMAPIError(f"LLM 流式调用失败:{type(exc).__name__}") from exc + options = self._request_args(system_prompt, user_message, history, stream=True) + emitted = False + for attempt in range(self._max_retries + 1): + response = None + try: + async with asyncio.timeout(self._timeout_seconds): + response = await client.chat.completions.create(**options) + async for event in response: + choices = getattr(event, "choices", None) + if not choices: + continue + token = getattr(getattr(choices[0], "delta", None), "content", None) + if token: + emitted = True + yield token + if not emitted: + raise LLMAPIError("LLM 返回了空内容") + return + except LLMError: + raise + except Exception as exc: + if emitted or attempt == self._max_retries or not self._retryable(exc): + raise self._error(exc) from exc + finally: + close = getattr(response, "close", None) + if close is not None: + with suppress(Exception): + await asyncio.wait_for(close(), timeout=5) + await self._retry_delay(attempt) def _default_fake_responder(system_prompt: str, user_message: str) -> str: diff --git a/src/api/endpoints.py b/src/api/endpoints.py index 13907b4..2e4e129 100644 --- a/src/api/endpoints.py +++ b/src/api/endpoints.py @@ -556,7 +556,7 @@ async def set_info( json_data = request.model_dump() if Config.AGENT_BACKEND == "langgraph": - code, msg = get_app_service().set_info( + code, msg = await get_app_service().set_info( json_data['sessionId'], json_data['key'], json_data['value'] ) return ProcessResponse_set( @@ -668,7 +668,7 @@ async def get_info( return ProcessResponse_get( sessionId=json_data['sessionId'], timeStamp=json_data['timeStamp'], - value=get_app_service().get_info(json_data['sessionId'], json_data['key']), + value=await get_app_service().get_info(json_data['sessionId'], json_data['key']), code="200", msg="", ) diff --git a/src/backends/chat.py b/src/backends/chat.py index 4c43c06..3a3f916 100644 --- a/src/backends/chat.py +++ b/src/backends/chat.py @@ -1,6 +1,6 @@ """Backend-neutral chat contract.""" -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Callable from dataclasses import dataclass, field from typing import Any, Protocol @@ -17,6 +17,8 @@ class ChatInput: # 人员进 person_fields）。LangGraph 后端合并进图输入； # FastGPT 后端忽略该字段。 injected_fields: dict | None = None + # 图提交后、公开结果之前同步业务字段；不进入图状态或 HTTP schema。 + on_commit: Callable[[dict], None] | None = field(default=None, repr=False, compare=False) @dataclass(frozen=True) @@ -47,6 +49,8 @@ class ChatResult: content: str status_code: str | None = None form_update: Any = field(default_factory=dict) + # 内部提交差量，独立于 needFormUpdate 对外显隐。 + committed_patch: dict | None = field(default=None, repr=False, compare=False) class ChatBackend(Protocol): diff --git a/src/backends/langgraph.py b/src/backends/langgraph.py index 447e932..2d8c785 100644 --- a/src/backends/langgraph.py +++ b/src/backends/langgraph.py @@ -15,9 +15,10 @@ test/backends/test_langgraph_streaming.py """ import asyncio -from collections.abc import AsyncIterator, Callable +from contextlib import suppress +from collections.abc import AsyncIterator, Awaitable, Callable -from ..agent.collect import _EXTRACTION_MARKER +from ..agent.collect import STATE_PREFIX_RE, parse_state_reply from ..agent.graph import GraphDeps, build_graph, workflow_fake_llm from ..agent.llm import LLMAuthError, LLMRateLimitError, LLMService from .chat import ( @@ -36,42 +37,58 @@ _DONE = object() class _TokenPassThroughLLM: - """LLMService 包装器：对话调用逐 token 旁路给 sink，提取调用静默。 + """先验证阶段前缀；已经输出的回复不能重生成后拼接。""" - 图节点仍按批处理接口 ``complete()`` 使用 LLM（校验/解析依赖完整 - 文本，语义不变）；本包装器在收集 token 的同时推给 SSE 桥： - - 系统提示含【字段提取】标记的辅助调用走 inner.complete， - 不推流（提取 JSON 不应出现在客户端流里）； - - 其余调用若 inner 提供 stream_tokens 则逐 token 推送； - 否则退化为整段推送（旧行为，fake/测试 LLM 兼容）。 - """ - - def __init__(self, inner: LLMService, sink: Callable[[str], None]): + def __init__(self, inner: LLMService, sink: Callable[[str], Awaitable[None]]): self._inner = inner self._sink = sink + self.response_started = False - async def complete( - self, system_prompt: str, user_message: str, *, - history: list[dict[str, str]] | None = None, + async def complete(self, system_prompt, user_message, *, history=None): + # 提取等未经过 validated_reply 的辅助调用始终静默。 + return await self._inner.complete(system_prompt, user_message, history=history) + + async def complete_validated( + self, system_prompt, user_message, *, history=None, stage_validator, ) -> str: - if _EXTRACTION_MARKER in system_prompt: - return await self._inner.complete( - system_prompt, user_message, history=history - ) stream_tokens = getattr(self._inner, "stream_tokens", None) if stream_tokens is None: - text = await self._inner.complete( - system_prompt, user_message, history=history - ) - self._sink(text) - return text + raw = await self.complete(system_prompt, user_message, history=history) + try: + code, _ = parse_state_reply(raw) + except ValueError: + return raw + if stage_validator(code): + self.response_started = True + await self._sink(raw) + return raw + chunks: list[str] = [] - async for token in stream_tokens( - system_prompt, user_message, history=history - ): - chunks.append(token) - self._sink(token) - return "".join(chunks) + prefix = "" + accepted = False + source = stream_tokens(system_prompt, user_message, history=history) + try: + async for token in source: + chunks.append(token) + if not accepted: + prefix += token + match = STATE_PREFIX_RE.match(prefix) + if match is None: + if len(prefix) > 1024 or "" in prefix: + return "".join(chunks) + continue + if not stage_validator(match.group(1)): + return "".join(chunks) + accepted = True + self.response_started = True + await self._sink(prefix) + else: + await self._sink(token) + return "".join(chunks) + finally: + close = getattr(source, "aclose", None) + if close is not None: + await close() class LangGraphBackend(ChatBackend): @@ -99,10 +116,11 @@ class LangGraphBackend(ChatBackend): （作业 6 起实现）。 """ result = await self._complete_turn(chat_input) + patch = result.get("turn_patch") or {} return ChatResult( - content = result["response_text"], - status_code = None, - form_update = result.get("turn_patch") or {} + content=result["response_text"], + form_update=patch if chat_input.need_form_update else {}, + committed_patch=patch, ) async def _complete_turn( @@ -127,11 +145,15 @@ class LangGraphBackend(ChatBackend): if chat_input.injected_fields: graph_input.update(chat_input.injected_fields) try: - return await self._graph.ainvoke( + result = await self._graph.ainvoke( graph_input, config={"configurable": {"thread_id": chat_input.session_id}}, context=GraphDeps(llm=llm or self._llm), ) + # 即使消费者在图完成后断开，业务 repository 仍与 checkpoint 一致。 + if chat_input.on_commit is not None: + chat_input.on_commit(result.get("turn_patch") or {}) + return result except LLMAuthError as exc: raise ChatBackendAuthenticationError(str(exc)) from exc except LLMRateLimitError as exc: @@ -148,23 +170,25 @@ class LangGraphBackend(ChatBackend): - 生成器被关闭（客户端断开）时取消图任务，停止 LLM 调用； - 图任务异常在消费端重新抛出（错误映射沿用 _complete_turn）。 """ - queue: asyncio.Queue = asyncio.Queue() - token_count = {"n": 0} # sink 同步回调不可直接闭包递增，用盒 + queue: asyncio.Queue = asyncio.Queue(maxsize=32) + token_count = 0 - def sink(token: str) -> None: - # LLM 在图任务的事件循环线程内同步回调 - token_count["n"] += 1 - queue.put_nowait(token) + async def sink(token: str) -> None: + nonlocal token_count + token_count += 1 + await queue.put(token) async def run_turn() -> dict: try: result = await self._complete_turn( chat_input, llm=_TokenPassThroughLLM(self._llm, sink) ) - except BaseException as exc: - queue.put_nowait(exc) + except asyncio.CancelledError: + raise + except Exception as exc: + await queue.put(exc) return {} - queue.put_nowait(_DONE) + await queue.put(_DONE) return result task = asyncio.ensure_future(run_turn()) @@ -177,19 +201,17 @@ class LangGraphBackend(ChatBackend): raise item yield TextDelta(item) result = await task - if token_count["n"] == 0: + if token_count == 0: # 本回合无 LLM 调用（welcome 等固定应答轮）： # sink 无 token 可推，回退为完整文本切片（旧行为） text = result.get("response_text") or "" if text: yield TextDelta(text) form_update = result.get("turn_patch") or {} - if form_update: + if form_update and chat_input.need_form_update: yield FormUpdate(form_update) finally: if not task.done(): task.cancel() - try: - await task - except (asyncio.CancelledError, Exception): - pass + with suppress(asyncio.CancelledError): + await task diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py index 9e2f7c7..4fd9e92 100644 --- a/src/core/fastgpt_client.py +++ b/src/core/fastgpt_client.py @@ -16,11 +16,11 @@ _chat_backend: ChatBackend | None = None _langgraph_graph = None _app_service: AccidentAppService | None = None -# 后端超时/重试 SLA（两后端对齐；契约测试防止再漂移）。 -# FastGPT 侧：客户端每次请求总超时 + 重试次数； -# LangGraph 侧：LLM 每次调用总超时 + 重试次数（退避为指数式， -# 与 FastGPT 固定 1s 不同——总尝试次数与单次超时可比，见 -# test/backends/test_sla_alignment.py）。 +# 两后端缺省 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 @@ -30,6 +30,7 @@ async def lifespan(app: FastAPI): """Manage chat backend resource lifecycle.""" global _chat_backend, _fastgpt_client Config.validate() + llm = None if Config.AGENT_BACKEND == "langgraph": # 作业 2：LangGraph fake 模式不要求 FastGPT 密钥。 @@ -38,7 +39,8 @@ async def lifespan(app: FastAPI): global _langgraph_graph, _app_service _langgraph_graph = build_graph() - _chat_backend = LangGraphBackend(_langgraph_graph, llm=build_llm_service()) + llm = build_llm_service() + _chat_backend = LangGraphBackend(_langgraph_graph, llm=llm) # 作业 6：应用服务 + 内存 repository（作业 7 换 PostgreSQL）。 _app_service = AccidentAppService( _chat_backend, InMemorySessionRepository() @@ -48,7 +50,7 @@ async def lifespan(app: FastAPI): api_key=Config.FASTGPT_API_KEY, base_url=Config.FASTGPT_BASE_URL, timeout=FASTGPT_TIMEOUT_SECONDS, - max_retries=FASTGPT_MAX_RETRIES, + max_retries=FASTGPT_MAX_RETRIES + 1, retry_delay=1.0, enable_logging=Config.DEBUG, ) @@ -61,6 +63,8 @@ async def lifespan(app: FastAPI): _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 diff --git a/src/services/app_service.py b/src/services/app_service.py index 7ccfd33..478ad95 100644 --- a/src/services/app_service.py +++ b/src/services/app_service.py @@ -5,11 +5,9 @@ （对照旧 FastGPT 实现的“辅助对话 + 删记录”hack）； - chat 每轮从 repository 读最新业务字段注入图（set_info 写入后 下一轮 chat 必须可见），图产出字段差量后写回 repository； -- **同会话串行**（A2 并发隔离）：chat / stream / delete_session 按 - session_id 互斥执行——FastGPT 服务端对同一 chatId 串行处理， - 内存态无锁会交错（checkpoint 写竞争、簿记错乱）。不同会话 - 无竞争。set/get 保持同步无锁：repository 的 load/apply_update - 是 dict 原子操作，不存在混合中间态。 +- chat / stream / set_info / get_info / delete_session 都在同会话锁内。 + 锁的引用计数包含等待者，取消和异常都会归还引用；没有使用者时回收。 + 不同会话独立执行，同会话 set_info 不会被在途聊天的旧结果覆盖。 - 未知 key 兼容策略：get 对未知 key 返回 JSON 空串 '""'（现状兼容）； set 对未知 key 拒绝写入（白名单保护，差异见兼容清单）； - includeInputInfo 继续被 endpoint 接受，但本层不再使用 @@ -18,7 +16,8 @@ endpoint 只负责校验、协议（SSE / 响应模型）与依赖注入。 """ -import asyncio +from contextlib import aclosing +from dataclasses import replace from ..backends.chat import ChatBackend, ChatInput, ChatResult, FormUpdate from .fields_registry import ( @@ -27,10 +26,13 @@ from .fields_registry import ( BUSINESS_TO_FORM_KEY, FieldGroup, business_patch_from_form, + business_value, + canonical_key, parse_field_value, to_external_string, ) from .repository import SessionRepository +from .session_locks import SessionLocks # set_info 对未知/受保护 key 的拒绝码（与现有“入参不规范”文案对齐） SET_INFO_INVALID_KEY_MSG = "入参不规范：未知或受保护的字段 key" @@ -42,17 +44,7 @@ class AccidentAppService: def __init__(self, backend: ChatBackend, repository: SessionRepository): self._backend = backend self._repository = repository - # 会话级互斥锁：同 session 的 chat/stream/delete 串行执行。 - # delete 成功后清理锁项；清理与等待者的竞争语义见 - # test_concurrent_delete_then_chat_behaves_as_fresh_session。 - self._locks: dict[str, asyncio.Lock] = {} - - def _lock_for(self, session_id: str) -> asyncio.Lock: - lock = self._locks.get(session_id) - if lock is None: - lock = asyncio.Lock() - self._locks[session_id] = lock - return lock + self._locks = SessionLocks() # ---- chat：图每轮读取最新业务状态，产出差量写回 ---- @@ -67,9 +59,10 @@ class AccidentAppService: 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: + # 注入同义键的一致视图，覆盖 checkpoint 里的旧值。 + for key, spec in FIELD_REGISTRY.items(): + value = business_value(business, key) + if value is None: continue if spec.group is FieldGroup.ACD: form_key = BUSINESS_TO_FORM_KEY.get(key) @@ -87,49 +80,64 @@ class AccidentAppService: async def chat(self, chat_input: ChatInput) -> ChatResult: """非流式一轮对话：注入最新业务字段 -> 执行图 -> 写回字段差量。""" - async with self._lock_for(chat_input.session_id): + 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( - 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, - ) + 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 {}, ) - # 图产出的字段差量写回 repository， - # 使 set/get_info 与后续轮次看到同一事实来源 - if isinstance(result.form_update, dict): - self._writeback_patch(chat_input.session_id, result.form_update) - return result async def stream(self, chat_input: ChatInput): """流式代理：字段注入同 chat()；FormUpdate 事件捕获写回后透传。""" - async with self._lock_for(chat_input.session_id): - async for event in self._stream_locked(chat_input): - yield event + 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) - 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, - ) - async for event in self._backend.stream(result_input): - if isinstance(event, FormUpdate) and isinstance(event.data, dict): - self._writeback_patch(chat_input.session_id, event.data) - yield event + 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、零辅助对话 ---- - def set_info(self, session_id: str, key: str, value: str) -> tuple[str, str]: + 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: @@ -138,10 +146,14 @@ class AccidentAppService: parsed = parse_field_value(spec, value) except ValueError: return "500", "入参不规范：字段值无法解析" - self._repository.apply_update(session_id, {key: parsed}) + self._repository.apply_update(session_id, {canonical_key(key): parsed}) return "200", "" - def get_info(self, session_id: str, key: str) -> str: + async def get_info(self, session_id: str, key: str) -> str: + async with self._locks.hold(session_id): + return self._get_info(session_id, key) + + def _get_info(self, session_id: str, key: str) -> str: """按分组或单键读出，返回 JSON 字符串（兼容现有 /get_info 契约）。""" import json @@ -150,7 +162,7 @@ class AccidentAppService: return json.dumps( { group.value: { - k: to_external_string(business.get(k)) + k: to_external_string(business_value(business, k)) for k in GROUP_KEYS[group] } for group in FieldGroup @@ -163,13 +175,13 @@ class AccidentAppService: group = None if group is not None: return json.dumps( - {k: to_external_string(business.get(k)) for k in GROUP_KEYS[group]}, + {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.get(key)), ensure_ascii=False) + return json.dumps(to_external_string(business_value(business, key)), ensure_ascii=False) # ---- delete_session：业务状态 + 图 checkpoint 一起清理 ---- @@ -178,10 +190,11 @@ class AccidentAppService: drop_checkpoint：由 core 层注入的 checkpoint 清理函数 （图生命周期归 lifespan 管理，本层不持有图引用）。 - 与会话锁互斥：删除不与在途 chat 交错。删除后清理锁项。 + 与会话锁互斥：删除不与在途 chat/set/get 交错。 + 释放锁引用时保留等待者使用的锁，最后一个使用者退出后回收。 repository 删除幂等；checkpoint 缺失时静默（幂等语义一致）。 """ - async with self._lock_for(session_id): + async with self._locks.hold(session_id): self._repository.delete(session_id) if drop_checkpoint is not None: try: @@ -189,8 +202,6 @@ class AccidentAppService: except Exception: # checkpoint 已不存在视为删除成功的一部分（幂等） pass - # 清理锁项：引用随等待者自然释放，新会话重建新锁 - self._locks.pop(session_id, None) __all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"] diff --git a/src/services/fields_registry.py b/src/services/fields_registry.py index 121929c..b3209fa 100644 --- a/src/services/fields_registry.py +++ b/src/services/fields_registry.py @@ -125,6 +125,22 @@ BUSINESS_TO_FORM_KEY: dict[str, str] = { } +# 同义人员键只存 workflow 原生键；API 键保留为读写别名。 +FIELD_ALIASES = { + f"{external}{index}": f"driver{index}{native}" + for index in (1, 2) + for external, native in (("sfzmhm", "ID"), ("sjhm", "Phonenumber")) +} + + +def canonical_key(key: str) -> str: + return FIELD_ALIASES.get(key, key) + + +def business_value(fields: dict, key: str): + return fields.get(canonical_key(key)) + + def business_patch_from_form(form_patch: dict) -> dict: """图产出的表单差量 -> repository 业务键差量（白名单内）。 @@ -137,8 +153,10 @@ def business_patch_from_form(form_patch: dict) -> dict: spec = FIELD_REGISTRY.get(business_key) if spec is None or not spec.writable: continue - # 类型以注册表为准：图表单值已是内部类型，直接采用 - patch[business_key] = value + canonical = canonical_key(business_key) + if canonical in patch and patch[canonical] != value: + raise ValueError("同一字段的别名包含冲突值") + patch[canonical] = value return patch diff --git a/src/services/session_locks.py b/src/services/session_locks.py new file mode 100644 index 0000000..5920e42 --- /dev/null +++ b/src/services/session_locks.py @@ -0,0 +1,28 @@ +"""单事件循环内的会话互斥：引用计数包含持有者和等待者。""" + +import asyncio +from contextlib import asynccontextmanager +from dataclasses import dataclass, field + + +@dataclass +class _Entry: + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + users: int = 0 + + +class SessionLocks: + def __init__(self): + self._entries: dict[str, _Entry] = {} + + @asynccontextmanager + async def hold(self, session_id: str): + entry = self._entries.setdefault(session_id, _Entry()) + entry.users += 1 + try: + async with entry.lock: + yield + finally: + entry.users -= 1 + if entry.users == 0: + del self._entries[session_id] diff --git a/test/agent/test_llm_stream_tokens.py b/test/agent/test_llm_stream_tokens.py index d672d95..a94c3fc 100644 --- a/test/agent/test_llm_stream_tokens.py +++ b/test/agent/test_llm_stream_tokens.py @@ -64,7 +64,7 @@ async def test_stream_tokens_yields_incrementally_and_forwards_kwargs(): assert kwargs["model"] == "test-model" assert kwargs["stream"] is True assert kwargs["temperature"] == 0 - assert kwargs["reasoning_effort"] == "none" # extra_body 透传 + assert kwargs["extra_body"]["reasoning_effort"] == "none" # extra_body 透传 assert kwargs["messages"][0] == {"role": "system", "content": "系统"} assert kwargs["messages"][-1] == {"role": "user", "content": "输入"} diff --git a/test/backends/test_langgraph_streaming.py b/test/backends/test_langgraph_streaming.py index e03dbd1..ae8131e 100644 --- a/test/backends/test_langgraph_streaming.py +++ b/test/backends/test_langgraph_streaming.py @@ -58,7 +58,7 @@ class StreamingFakeLLM: def make_input(session="stream-s1", text="继续办理"): - return ChatInput(session_id=session, text=text, client_mode="direct") + return ChatInput(session_id=session, text=text, client_mode="direct", need_form_update=True) async def _second_turn_events(backend, session="stream-s1"): diff --git a/test/backends/test_sla_alignment.py b/test/backends/test_sla_alignment.py index 686b4f8..9e96ca6 100644 --- a/test/backends/test_sla_alignment.py +++ b/test/backends/test_sla_alignment.py @@ -1,11 +1,8 @@ -"""后端超时/重试 SLA 对齐契约测试。 +"""超时/重试配置与 SDK 适配契约。 -背景（可替代性清单）：FastGPT 客户端为 60s/3 次，LangGraph LLM 侧 -曾是 30s/2 次，客户端感知到的超时行为不等价。本测试把两端常量 -钉在一起，防止再次漂移。 - -注：退避策略有意不同——FastGPT 固定 1s，LLM 指数退避（限流场景 -更友好）；对齐的是"单次调用总超时"与"最大重试次数"两个 SLA 参数。 +LLM max_retries 表示额外重试次数；FastGPT SDK 同名参数实际表示 +总尝试次数，需要 +1 转换。退避统一为 1、2、4 秒。 +超时分别约束一次 LLM 尝试与 FastGPT HTTP SDK，不代表整轮总期限。 """ import inspect @@ -40,3 +37,40 @@ def test_named_profile_defaults_match_fastgpt_client(monkeypatch): profile = Config.llm_profile("probe") assert profile["timeout_seconds"] == FASTGPT_TIMEOUT_SECONDS assert profile["max_retries"] == FASTGPT_MAX_RETRIES + + +import asyncio +from unittest.mock import AsyncMock +import httpx +import pytest +from fastapi import FastAPI +from fastgpt_client.exceptions import APIError +from src.core import fastgpt_client as wiring +from src.core.config import Config + + +@pytest.mark.asyncio +async def test_fastgpt_sdk_uses_same_attempt_count_and_backoff(monkeypatch): + monkeypatch.setattr(Config, "validate", lambda: None) + monkeypatch.setattr(Config, "AGENT_BACKEND", "fastgpt") + monkeypatch.setattr(Config, "FASTGPT_BASE_URL", "http://audit") + monkeypatch.setattr(Config, "FASTGPT_API_KEY", "audit") + sleep = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep) + async with wiring.lifespan(FastAPI()): + client = wiring.get_fastgpt_client() + request = AsyncMock(return_value=httpx.Response(503)) + with pytest.raises(APIError): + await client._retry_request_async(request, "audit") + assert request.await_count == 4 + assert [call.args[0] for call in sleep.await_args_list] == [1, 2, 4] + + +@pytest.mark.asyncio +async def test_llm_backoff_uses_same_delays(monkeypatch): + sleep = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep) + llm = OpenAILLMService(model="audit", api_key="audit") + for attempt in range(3): + await llm._retry_delay(attempt) + assert [call.args[0] for call in sleep.await_args_list] == [1, 2, 4] diff --git a/test/services/test_app_service.py b/test/services/test_app_service.py index ae17de5..a58cbab 100644 --- a/test/services/test_app_service.py +++ b/test/services/test_app_service.py @@ -52,55 +52,60 @@ def make_service(form_update=None): # ---- set/get：白名单、分组、外部表示 ---- -def test_set_then_get_single_key_roundtrip(): +@pytest.mark.asyncio +async def test_set_then_get_single_key_roundtrip(): service, backend = make_service() - code, msg = service.set_info("s1", "hphm1", "苏B12345") + code, msg = await service.set_info("s1", "hphm1", "苏B12345") assert (code, msg) == ("200", "") - assert json.loads(service.get_info("s1", "hphm1")) == "苏B12345" + assert json.loads(await service.get_info("s1", "hphm1")) == "苏B12345" # set/get 零模型调用 assert backend.llm_calls == 0 -def test_get_groups_match_existing_contract(): +@pytest.mark.asyncio +async 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", "张三") + await service.set_info("s1", "ywrysw", "0") # False + await service.set_info("s1", "jdcsl", "0") # int 0 + await service.set_info("s1", "xm2", "张三") - acd = json.loads(service.get_info("s1", "acdinfo")) + acd = json.loads(await 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")) + human2 = json.loads(await service.get_info("s1", "acdhuman2")) assert human2["xm2"] == "张三" assert human2["sfzmwh2"] == "" # 全字段齐全，未填为空 - all_groups = json.loads(service.get_info("s1", "all")) + all_groups = json.loads(await service.get_info("s1", "all")) assert set(all_groups) == {"acdinfo", "acdhuman1", "acdhuman2"} -def test_get_unknown_key_returns_json_empty_string(): +@pytest.mark.asyncio +async 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") == '""' + await service.set_info("s1", "hphm1", "苏B12345") + assert await service.get_info("s1", "no_such_key") == '""' + assert await service.get_info("ghost-session", "hphm1") == '""' -def test_set_unknown_or_internal_key_rejected(): +@pytest.mark.asyncio +async def test_set_unknown_or_internal_key_rejected(): service, backend = make_service() - code, msg = service.set_info("s1", "no_such_key", "x") + code, msg = await 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") + code, _ = await service.set_info("s1", "stage_code", "1002") assert code == "500" # 内部字段保护 - assert service.get_info("s1", "hphm1") == '""' # 未被污染 + assert await service.get_info("s1", "hphm1") == '""' # 未被污染 assert backend.llm_calls == 0 -def test_set_rejects_unparsable_value(): +@pytest.mark.asyncio +async def test_set_rejects_unparsable_value(): service, _ = make_service() - code, msg = service.set_info("s1", "ywrysw", "maybe") + code, msg = await service.set_info("s1", "ywrysw", "maybe") assert code == "500" and "无法解析" in msg @@ -110,8 +115,8 @@ def test_set_rejects_unparsable_value(): 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", "李四") + await service.set_info("s1", "hphm1", "苏B12345") + await service.set_info("s1", "xm2", "李四") result = await service.chat( _input("s1", "我是刚才报警的驾驶员") @@ -124,8 +129,8 @@ async def test_chat_sees_set_info_values_and_writes_back_patch(): 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" + assert json.loads(await service.get_info("s1", "acdinfo"))["ywrysw"] == "0" + assert json.loads(await service.get_info("s1", "hphm1")) == "苏B12345" @pytest.mark.asyncio @@ -135,21 +140,21 @@ async def test_chat_form_patch_mapped_to_business_keys(): "has_injury": False, "vehicle_count": 2, "accident_cause": "追尾", }) await service.chat(_input("s1", "两辆汽车追尾，没有人受伤")) - acd = json.loads(service.get_info("s1", "acdinfo")) + acd = json.loads(await service.get_info("s1", "acdinfo")) assert acd["ywrysw"] == "0" assert acd["jdcsl"] == "2" assert acd["sgyy"] == "追尾" # 表单键本身不落 repository（未知业务键 -> JSON 空串） - assert json.loads(service.get_info("s1", "has_injury")) == "" + assert json.loads(await service.get_info("s1", "has_injury")) == "" @pytest.mark.asyncio async def test_set_info_acd_value_mapped_into_form_injection(): """set_info 写业务键 -> chat 注入为图表单键（反向映射）。""" service, backend = make_service() - service.set_info("s1", "ywrysw", "0") - service.set_info("s1", "jdcsl", "2") - service.set_info("s1", "bjrjs", "1") # 无表单来源，不注入 + await service.set_info("s1", "ywrysw", "0") + await service.set_info("s1", "jdcsl", "2") + await service.set_info("s1", "bjrjs", "1") # 无表单来源，不注入 await service.chat(_input("s1", "继续办理")) injected = backend.calls[0].injected_fields assert injected["accident_fields"] == {"has_injury": False, "vehicle_count": 2} @@ -166,7 +171,7 @@ async def test_chat_without_business_state_passes_no_injection(): async def test_stream_injects_fields_writes_back_form_update(): """流式：注入同 chat；FormUpdate 事件差量捕获写回后透传。""" service, backend = make_service(form_update={"has_injury": False}) - service.set_info("s1", "xm1", "王五") + await service.set_info("s1", "xm1", "王五") events = [ event async for event in service.stream(_input("s1", "继续办理")) ] @@ -174,7 +179,7 @@ async def test_stream_injects_fields_writes_back_form_update(): assert kinds == ["TextDelta", "FormUpdate", "TextDelta"] assert backend.stream_calls[0].injected_fields["person_fields"]["xm1"] == "王五" # FormUpdate 差量已写回 repository（表单键 -> 业务键） - assert json.loads(service.get_info("s1", "acdinfo"))["ywrysw"] == "0" + assert json.loads(await service.get_info("s1", "acdinfo"))["ywrysw"] == "0" # ---- delete：清理 + 幂等 ---- @@ -187,13 +192,13 @@ async def test_delete_clears_business_state_and_is_idempotent(): drops.append(session_id) service, backend = make_service() - service.set_info("s1", "hphm1", "苏B12345") + await 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") == '""' + assert await service.get_info("s1", "hphm1") == '""' await service.chat(_input("s1", "还在吗")) assert backend.calls[-1].injected_fields is None @@ -210,16 +215,16 @@ async def test_delete_tolerates_missing_checkpoint(): raise RuntimeError("no such thread") service, _ = make_service() - service.set_info("s1", "xm1", "王五") + await service.set_info("s1", "xm1", "王五") # checkpoint 不存在视为删除成功的一部分，不向上抛 await service.delete_session("s1", drop_checkpoint=missing) - assert service.get_info("s1", "xm1") == '""' + assert await 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") + return ChatInput(session_id=session_id, text=text, client_mode="direct", need_form_update=True) # ---- 并发隔离（A2）：同会话串行、不同会话并行、delete 与 chat 互斥 ---- @@ -311,13 +316,13 @@ async def test_concurrent_delete_serialization(): @pytest.mark.asyncio async def test_concurrent_chat_and_set_info_no_corruption(): - """set_info（同步无锁）与并发 chat 交错：repository 键值保持完整。""" + """set_info 与 chat 按同一会话串行，最终保留写入的完整车牌。""" backend = SerialProbeBackend() service = AccidentAppService(backend, InMemorySessionRepository()) results = await asyncio.gather( service.chat(_input("s1", "a")), - asyncio.to_thread(service.set_info, "s1", "hphm1", "苏B12345"), + service.set_info("s1", "hphm1", "苏B12345"), ) assert results[1] == ("200", "") - # 无论时序如何，最终两个写者之一的车牌值完整存在（无撕裂） - assert json.loads(service.get_info("s1", "hphm1")) in ("苏B12345", "") + # set 写入在 chat 结束后执行，不能丢失。 + assert json.loads(await service.get_info("s1", "hphm1")) == "苏B12345" diff --git a/test/services/test_field_commit_and_retry.py b/test/services/test_field_commit_and_retry.py new file mode 100644 index 0000000..9d9fdff --- /dev/null +++ b/test/services/test_field_commit_and_retry.py @@ -0,0 +1,158 @@ +import asyncio +import json +from contextlib import aclosing +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import httpx +import pytest +from openai import AsyncOpenAI + +from src.agent.llm import LLMAPIError, LLMRateLimitError, OpenAILLMService +from src.agent.prompts import SINGLE_CONFIRM, render_prompt +from src.backends.chat import ChatInput, FormUpdate, TextDelta +from src.backends.langgraph import LangGraphBackend +from src.services.app_service import AccidentAppService +from src.services.repository import InMemorySessionRepository + + +class PatchGraph: + async def ainvoke(self, *args, **kwargs): + return {'response_text': '1002ok', 'turn_patch': {'has_injury': False}} + + +@pytest.mark.asyncio +async def test_person_synonyms_share_storage_and_prompt_values(): + repo = InMemorySessionRepository() + service = AccidentAppService(LangGraphBackend(graph=PatchGraph()), repo) + for index in (1, 2): + for api, native in ((f'sfzmhm{index}', f'driver{index}ID'), (f'sjhm{index}', f'driver{index}Phonenumber')): + await service.set_info('s', native, 'old') + 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'] + prompt = render_prompt(SINGLE_CONFIRM, {'person_fields': fields}) + assert 'old' not in prompt + assert 'new' in prompt + await service.set_info('s', 'driver1ID', 'latest') + group = json.loads(await service.get_info('s', 'acdhuman1')) + assert group['driver1ID'] == group['sfzmhm1'] == 'latest' + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +@pytest.mark.parametrize('visible', [False, True]) +async def test_form_visibility_does_not_control_internal_commit(stream, visible): + backend = LangGraphBackend(graph=PatchGraph()) + service = AccidentAppService(backend, InMemorySessionRepository()) + request = ChatInput('s', 'hello', need_form_update=visible) + if stream: + events = [e async for e in service.stream(request)] + assert any(isinstance(e, FormUpdate) for e in events) is visible + else: + result = await service.chat(request) + assert bool(result.form_update) is visible + assert json.loads(await service.get_info('s', 'ywrysw')) == '0' + + +@pytest.mark.asyncio +async def test_commit_is_not_lost_when_consumer_closes_after_graph_completes(): + backend = LangGraphBackend(graph=PatchGraph()) + repo = InMemorySessionRepository() + service = AccidentAppService(backend, repo) + async with aclosing(service.stream(ChatInput('s', 'hello'))) as events: + assert isinstance(await anext(events), TextDelta) + assert json.loads(await service.get_info('s', 'ywrysw')) == '0' + assert not service._locks._entries + + +@pytest.mark.asyncio +async def test_factory_disables_sdk_retries(): + llm = OpenAILLMService(model='m', api_key='audit') + try: + assert llm._get_client().max_retries == 0 + finally: + await llm.aclose() + + +def completion(stream): + if not stream: + return httpx.Response(200, json={ + 'id': 'r', 'object': 'chat.completion', 'created': 0, 'model': 'm', + 'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': 'ok'}, 'finish_reason': 'stop'}], + }) + data = {'id': 'r', 'object': 'chat.completion.chunk', 'created': 0, 'model': 'm', + 'choices': [{'index': 0, 'delta': {'content': 'ok'}, 'finish_reason': None}]} + return httpx.Response(200, headers={'content-type': 'text/event-stream'}, + content='data: '+json.dumps(data)+'\n\ndata: [DONE]\n\n') + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +@pytest.mark.parametrize('status', [429, 503]) +async def test_actual_http_attempt_count_is_not_multiplied_by_sdk(stream, status): + calls = [] + def handler(request): + calls.append(json.loads(request.content)) + if len(calls) <= 3: + return httpx.Response(status, json={'error': {'message': 'retry'}}) + return completion(stream) + llm = OpenAILLMService(model='m', api_key='audit', max_retries=3, + extra_body={'reasoning_effort': 'none'}) + llm._client = AsyncOpenAI(api_key='audit', base_url='http://audit/v1', max_retries=0, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler))) + llm._retry_delay = AsyncMock() + try: + result = ''.join([t async for t in llm.stream_tokens('system', 'user')]) if stream else await llm.complete('system', 'user') + assert result == 'ok' + assert len(calls) == 4 + assert llm._retry_delay.await_count == 3 + assert all(c['reasoning_effort'] == 'none' for c in calls) + finally: + await llm.aclose() + + +@pytest.mark.asyncio +async def test_stream_failure_after_first_token_closes_response_without_retry(): + class Response: + def __init__(self): self.closed = False + def __aiter__(self): return self.generate() + async def generate(self): + yield SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace(content='partial'))]) + raise OSError('broken socket') + async def close(self): self.closed = True + response = Response() + create = AsyncMock(return_value=response) + llm = OpenAILLMService(model='m', api_key='audit') + llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create))) + tokens = [] + with pytest.raises(LLMAPIError): + async for token in llm.stream_tokens('system', 'user'): + tokens.append(token) + assert tokens == ['partial'] + assert create.await_count == 1 + assert response.closed + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +async def test_retry_budget_exhaustion_is_exact(stream): + calls = [] + def handler(request): + calls.append(request) + return httpx.Response(429, json={'error': {'message': 'retry'}}) + llm = OpenAILLMService(model='m', api_key='audit', max_retries=2) + llm._retry_delay = AsyncMock() + llm._client = AsyncOpenAI(api_key='audit', base_url='http://audit/v1', max_retries=0, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler))) + try: + with pytest.raises(LLMRateLimitError): + if stream: + async for _ in llm.stream_tokens('system', 'user'): pass + else: + await llm.complete('system', 'user') + assert len(calls) == 3 + finally: + await llm.aclose() diff --git a/test/services/test_stream_and_session_safety.py b/test/services/test_stream_and_session_safety.py new file mode 100644 index 0000000..871f578 --- /dev/null +++ b/test/services/test_stream_and_session_safety.py @@ -0,0 +1,151 @@ +import asyncio + +import pytest + +from src.agent.collect import CollectReplyError, validated_reply +from src.agent.llm import FakeLLMService +from src.backends.chat import ChatInput, ChatResult, TextDelta +from src.backends.langgraph import LangGraphBackend +from src.services.app_service import AccidentAppService +from src.services.repository import InMemorySessionRepository + + +class ValidatingGraph: + def __init__(self): + self.committed = None + + async def ainvoke(self, data, *, config, context): + raw, stage = await validated_reply(context.llm, 'system', 'user', '1002') + self.committed = stage + return {'response_text': raw, 'turn_patch': {}} + + +class ScriptedStream: + def __init__(self, replies): + self.replies = iter(replies) + self.calls = 0 + + async def complete(self, *args, **kwargs): + return '{}' + + async def stream_tokens(self, *args, **kwargs): + self.calls += 1 + for chunk in next(self.replies): + yield chunk + + +@pytest.mark.asyncio +async def test_invalid_prefix_is_not_sent_and_repair_is_the_only_reply(): + graph = ValidatingGraph() + llm = ScriptedStream([ + ['9999invalid'], + ['1002', 'repaired'], + ]) + backend = LangGraphBackend(graph=graph, llm=llm) + events = [e async for e in backend.stream(ChatInput('s', 'hello'))] + assert ''.join(e.text for e in events if isinstance(e, TextDelta)) == '1002repaired' + assert graph.committed == '1002' + assert llm.calls == 2 + + +@pytest.mark.asyncio +async def test_failure_after_output_aborts_without_appending_a_repair(): + graph = ValidatingGraph() + llm = ScriptedStream([ + ['1002', 'partial', '9999'], + ['1002must not appear'], + ]) + backend = LangGraphBackend(graph=graph, llm=llm) + events = [] + with pytest.raises(CollectReplyError): + async for event in backend.stream(ChatInput('s', 'hello')): + events.append(event) + assert llm.calls == 1 + assert graph.committed is None + assert 'must not appear' not in ''.join(e.text for e in events) + + +@pytest.mark.asyncio +async def test_repair_exhaustion_emits_no_invalid_text(): + llm = ScriptedStream([['9999x'], ['malformed']]) + backend = LangGraphBackend(graph=ValidatingGraph(), llm=llm) + events = [] + with pytest.raises(CollectReplyError): + async for event in backend.stream(ChatInput('s', 'hello')): + events.append(event) + assert events == [] + + +class BlockingBackend: + def __init__(self): + self.entered = asyncio.Queue() + self.release = asyncio.Event() + self.active = 0 + self.peak = 0 + + async def complete(self, request): + self.active += 1 + self.peak = max(self.peak, self.active) + await self.entered.put(request.text) + try: + await self.release.wait() + return ChatResult('1002ok', form_update={'vehicle_count': 1}) + finally: + self.active -= 1 + + +@pytest.mark.asyncio +async def test_delete_waiter_and_new_chat_keep_one_lock(): + backend = BlockingBackend() + service = AccidentAppService(backend, InMemorySessionRepository()) + dropping, drop_release = asyncio.Event(), asyncio.Event() + + async def drop(_): + dropping.set() + await drop_release.wait() + + deletion = asyncio.create_task(service.delete_session('s', drop)) + await dropping.wait() + first = asyncio.create_task(service.chat(ChatInput('s', 'a'))) + await asyncio.sleep(0) + drop_release.set() + await deletion + assert await asyncio.wait_for(backend.entered.get(), 1) == 'a' + second = asyncio.create_task(service.chat(ChatInput('s', 'b'))) + await asyncio.sleep(0) + assert backend.entered.empty() + backend.release.set() + await asyncio.wait_for(asyncio.gather(first, second), 2) + assert backend.peak == 1 + assert not service._locks._entries + + +@pytest.mark.asyncio +async def test_set_waits_for_chat_and_cannot_be_overwritten_by_old_result(): + backend = BlockingBackend() + service = AccidentAppService(backend, InMemorySessionRepository()) + chat = asyncio.create_task(service.chat(ChatInput('s', 'a'))) + await backend.entered.get() + setting = asyncio.create_task(service.set_info('s', 'jdcsl', '2')) + await asyncio.sleep(0) + assert not setting.done() + backend.release.set() + await asyncio.gather(chat, setting) + assert await service.get_info('s', 'jdcsl') == '"2"' + + +@pytest.mark.asyncio +async def test_cancelled_waiter_does_not_leak_or_replace_lock(): + backend = BlockingBackend() + service = AccidentAppService(backend, InMemorySessionRepository()) + chat = asyncio.create_task(service.chat(ChatInput('s', 'a'))) + await backend.entered.get() + waiting = asyncio.create_task(service.get_info('s', 'jdcsl')) + await asyncio.sleep(0) + waiting.cancel() + with pytest.raises(asyncio.CancelledError): + await waiting + assert len(service._locks._entries) == 1 + backend.release.set() + await chat + assert not service._locks._entries diff --git a/test/services/test_stream_retry_cleanup.py b/test/services/test_stream_retry_cleanup.py new file mode 100644 index 0000000..1f4cd22 --- /dev/null +++ b/test/services/test_stream_retry_cleanup.py @@ -0,0 +1,94 @@ +import asyncio +from contextlib import aclosing +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from src.agent.collect import validated_reply +from src.agent.llm import LLMTimeoutError, OpenAILLMService +from src.backends.chat import ChatInput, TextDelta +from src.backends.langgraph import LangGraphBackend +from src.services.app_service import AccidentAppService +from src.services.repository import InMemorySessionRepository + + +class Graph: + async def ainvoke(self, data, *, config, context): + raw, _ = await validated_reply(context.llm, 'system', 'user', '1002') + return {'response_text': raw, 'turn_patch': {}} + + +@pytest.mark.asyncio +async def test_valid_prefix_reaches_client_before_model_finishes(): + release = asyncio.Event() + finished = asyncio.Event() + class LLM: + async def stream_tokens(self, *args, **kwargs): + yield '1002' + await release.wait() + yield 'tail' + finished.set() + backend = LangGraphBackend(graph=Graph(), llm=LLM()) + async with aclosing(backend.stream(ChatInput('s', 'hello'))) as events: + first = await asyncio.wait_for(anext(events), 1) + assert first == TextDelta('1002') + assert not finished.is_set() + release.set() + remaining = [e async for e in events] + assert ''.join(e.text for e in remaining) == 'tail' + assert finished.is_set() + + +@pytest.mark.asyncio +async def test_closing_application_stream_cancels_model_before_unlocking(): + cancelled = asyncio.Event() + class LLM: + async def stream_tokens(self, *args, **kwargs): + yield '1002' + try: + await asyncio.Future() + finally: + cancelled.set() + service = AccidentAppService(LangGraphBackend(graph=Graph(), llm=LLM()), InMemorySessionRepository()) + async with aclosing(service.stream(ChatInput('s', 'hello'))) as events: + await asyncio.wait_for(anext(events), 1) + assert cancelled.is_set() + assert not service._locks._entries + assert await service.set_info('s', 'hphm1', 'test') == ('200', '') + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +async def test_timeout_covers_whole_attempt_and_retries_are_bounded(stream): + responses = [] + class HangingResponse: + closed = False + def __aiter__(self): return self.generate() + async def generate(self): + await asyncio.Future() + yield None + async def close(self): self.closed = True + async def create(**kwargs): + if stream: + response = HangingResponse() + responses.append(response) + return response + await asyncio.Future() + llm = OpenAILLMService(model='m', api_key='audit', timeout_seconds=0.01, max_retries=2) + call = AsyncMock(side_effect=create) + llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=call))) + llm._retry_delay = AsyncMock() + # 先加载 SDK 错误类；外层看门狗不测模块惰性导入耗时。 + assert llm._retryable(TimeoutError()) + assert isinstance(llm._error(TimeoutError()), LLMTimeoutError) + async with asyncio.timeout(10): + with pytest.raises(LLMTimeoutError): + if stream: + async for _ in llm.stream_tokens('system', 'user'): pass + else: + await llm.complete('system', 'user') + assert call.await_count == 3 + assert llm._retry_delay.await_count == 2 + assert all(response.closed for response in responses)