From 5c719ed2ea5e059aedb7281312e85151e6b24675 Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Sun, 26 Jul 2026 07:25:24 +0800 Subject: [PATCH] Refactor chat backend integration and introduce new backend-neutral architecture - Replaced direct FastGPT client usage with a backend-neutral chat interface, allowing for improved flexibility and maintainability. - Introduced a new `ChatBackend` protocol and implemented `FastGPTBackend` to handle chat operations. - Updated the chat endpoint to utilize the new backend structure, enhancing the handling of chat requests and responses. - Added comprehensive tests to ensure compatibility and functionality of the new backend integration. --- src/api/endpoints.py | 163 +++++--------------- src/backends/__init__.py | 27 ++++ src/backends/chat.py | 72 +++++++++ src/backends/fastgpt.py | 148 ++++++++++++++++++ src/core/fastgpt_client.py | 29 +++- test/api/test_chat_backend_boundary.py | 70 +++++++++ test/api/test_public_schema_contract.py | 195 ++++++++++++++++++++++++ test/backends/test_fastgpt_backend.py | 123 +++++++++++++++ 8 files changed, 697 insertions(+), 130 deletions(-) create mode 100644 src/backends/__init__.py create mode 100644 src/backends/chat.py create mode 100644 src/backends/fastgpt.py create mode 100644 test/api/test_chat_backend_boundary.py create mode 100644 test/api/test_public_schema_contract.py create mode 100644 test/backends/test_fastgpt_backend.py diff --git a/src/api/endpoints.py b/src/api/endpoints.py index 1268f18..fe82a4d 100644 --- a/src/api/endpoints.py +++ b/src/api/endpoints.py @@ -1,11 +1,17 @@ from fastapi import APIRouter, HTTPException, Depends from fastapi.responses import StreamingResponse from ..schemas.models import ProcessRequest_chat, ProcessResponse_chat, ProcessRequest_get, ProcessResponse_get, ProcessRequest_set, ProcessResponse_set, ProcessResponse_delete_session, ProcessRequest_delete_session -from fastgpt_client import AsyncChatClient, aiter_stream_events -from fastgpt_client.exceptions import ( - APIError, AuthenticationError, RateLimitError, ValidationError +from fastgpt_client import AsyncChatClient +from ..backends.chat import ( + ChatBackend, + ChatBackendAPIError, + ChatBackendAuthenticationError, + ChatBackendRateLimitError, + ChatInput, + FormUpdate, + TextDelta, ) -from ..core.fastgpt_client import get_fastgpt_client +from ..core.fastgpt_client import get_chat_backend, get_fastgpt_client from ..core.config import Config from ..utils.text_chunker import SentenceTextChunker, SentenceTextChunkerConfig from loguru import logger @@ -14,7 +20,6 @@ import re import time router = APIRouter() -FORM_EXTRACT_MODULE_NAME = "文本内容提取事故信息" STATE_TAG_PATTERN = re.compile(r"\s*(\d+)\s*", flags=re.DOTALL) STATUS_CODE_MAP = { '0000': '结束通话', @@ -60,45 +65,6 @@ def extract_first_state_and_clean_content(text: str) -> tuple[str | None, str]: return match.group(1), STATE_TAG_PATTERN.sub("", text) -def parse_json_value(value): - """Parse JSON string values when possible.""" - parsed = value - for _ in range(3): - if not isinstance(parsed, str): - return parsed - parsed = parsed.strip() - if not parsed: - return {} - try: - parsed = json.loads(parsed) - except json.JSONDecodeError: - return parsed - return parsed - - -def extract_form_update_from_flow_nodes(nodes): - """Extract form update data from the configured FastGPT content-extract node.""" - if not isinstance(nodes, list): - return {} - - for node in nodes: - if not isinstance(node, dict): - continue - if node.get("moduleName") != FORM_EXTRACT_MODULE_NAME: - continue - - extract_result = node.get("extractResult", {}) - if not isinstance(extract_result, dict): - return {} - - form_update = extract_result.get("formUpdate", "") - if not form_update: - return {} - return parse_json_value(form_update) - - return {} - - def format_set_info_input(payload: dict, include_input_info: bool) -> str: """Build optional setInfo input for FastGPT helper calls.""" if not include_input_info: @@ -156,13 +122,17 @@ def create_sse_event(event: str, data: dict) -> str: async def chat( request: ProcessRequest_chat, stream: bool = False, - client: AsyncChatClient = Depends(get_fastgpt_client) + backend: ChatBackend = Depends(get_chat_backend) ): """Handle chat completion request.""" json_data = request.model_dump() need_form_update = json_data.get('needFormUpdate', False) use_text_chunk = json_data.get('useTextChunk', False) - chat_variables = {'needFormUpdate': need_form_update} + chat_input = ChatInput( + session_id=json_data['sessionId'], + text=json_data['text'], + need_form_update=need_form_update, + ) request_started_at = time.perf_counter() logger.info( "Chat request received " @@ -193,16 +163,8 @@ async def chat( else None ) try: - # Use SDK's create_chat_completion with stream=True - response = await client.create_chat_completion( - messages=[{"role": "user", "content": json_data['text']}], - chatId=json_data['sessionId'], - stream=True, - detail=True, - variables=chat_variables - ) logger.info( - "FastGPT stream response opened " + "Chat backend stream opened " f"sessionId={json_data['sessionId']} " f"open_latency_ms={(time.perf_counter() - stream_started_at) * 1000:.1f}" ) @@ -279,41 +241,34 @@ async def chat( state_filter_buffer = "" return cleaned - async for event in aiter_stream_events(response): + async for event in backend.stream(chat_input): try: if not first_event_logged: first_event_logged = True logger.info( - "FastGPT stream first event " - f"sessionId={json_data['sessionId']} kind={event.kind} " + "Chat backend stream first event " + f"sessionId={json_data['sessionId']} " + f"kind={type(event).__name__} " f"ttfb_ms={(time.perf_counter() - stream_started_at) * 1000:.1f}" ) - if event.kind == "flowResponses" and not module_form_sent: - form_update = extract_form_update_from_flow_nodes(event.data) - if form_update: - form_update_payload = form_update + if isinstance(event, FormUpdate) and not module_form_sent: + if event.data: + form_update_payload = event.data logger.info( - "FastGPT stream formUpdate extracted " + "Chat backend stream formUpdate received " f"sessionId={json_data['sessionId']} " - f"type={type(form_update).__name__} " - f"formUpdate={form_update!r}" + f"type={type(event.data).__name__} " + f"formUpdate={event.data!r}" ) - yield flush_form_update(form_update) + yield flush_form_update(event.data) module_form_sent = True continue - if event.kind not in {"answer", "fastAnswer", "data"}: + if not isinstance(event, TextDelta): continue - data = event.data - if not isinstance(data, dict): - continue - - try: - delta_content = data['choices'][0]['delta'].get('content', '') - except (KeyError, IndexError): - delta_content = '' + delta_content = event.text if not delta_content: continue @@ -400,23 +355,14 @@ async def chat( return StreamingResponse(event_generator(), media_type="text/event-stream") try: - # Use SDK's create_chat_completion - response = await client.create_chat_completion( - messages=[{"role": "user", "content": json_data['text']}], - chatId=json_data['sessionId'], - stream=False, - detail=True, - variables=chat_variables - ) - response.raise_for_status() - data = response.json() + result = await backend.complete(chat_input) logger.info( - "FastGPT non-stream response received " + "Chat backend non-stream response received " f"sessionId={json_data['sessionId']} " f"latency_ms={(time.perf_counter() - request_started_at) * 1000:.1f}" ) - except AuthenticationError as e: + except ChatBackendAuthenticationError as e: logger.error(f"Authentication error: {e}") return ProcessResponse_chat( sessionId=json_data['sessionId'], @@ -427,7 +373,7 @@ async def chat( code="401", msg="认证失败" ) - except RateLimitError as e: + except ChatBackendRateLimitError as e: logger.error(f"Rate limit error: {e}") return ProcessResponse_chat( sessionId=json_data['sessionId'], @@ -438,7 +384,7 @@ async def chat( code="429", msg="请求过于频繁，请稍后重试" ) - except APIError as e: + except ChatBackendAPIError as e: logger.error(f"API error: {e}") return ProcessResponse_chat( sessionId=json_data['sessionId'], @@ -462,39 +408,10 @@ async def chat( ) try: - # Extract content from FastGPT response - content = data['choices'][0]['message']['content'] - logger.info(f"FastGPT服务返回信息content: {content}") + content = result.content + logger.info(f"Chat backend returned content: {content}") - finish_reason = data['choices'][0]['finish_reason'] - - # Extract state variables - state = data.get('newVariables', {}).get('state', {}) - if isinstance(state, str): - state = json.loads(state) - - transfer_to_human = state.get("transfer_to_human", False) - ywrysw = state.get("ywrysw", False) - ywfjdc = state.get("ywfjdc", False) - ywmtc = state.get("ywmtc", False) - jdcsl = state.get("jdcsl", 0) - accident_info_complete = state.get("accident_info_complete", False) - user_is_ready = state.get("user_is_ready", False) - if isinstance(user_is_ready, str): - user_is_ready = user_is_ready.lower() == 'true' - driver_info_complete = state.get("driver_info_complete", False) - drivers_info_complete = state.get("drivers_info_complete", False) - driver_info_check = state.get("drivers_info_check", False) - drivers_info_check = state.get("drivers_info_check", False) - - logger.debug(f"State variables: {data.get('newVariables', {})}") - - # Parse content - sometimes content is a string, sometimes it is a list content_stage_code = None - if isinstance(content, list): - logger.debug("content是一个list") - content = content[0]['text']['content'] - if isinstance(content, str): logger.debug("content是一个str") content_stage_code, content = extract_first_state_and_clean_content(content) @@ -509,10 +426,12 @@ async def chat( logger.error(f"content既不是list也不是str, type: {type(content)}") raise ValueError("大模型回复不是list也不是str") - nextStageCode = content_stage_code or data['newVariables']['status_code'] + nextStageCode = content_stage_code or result.status_code + if not nextStageCode: + raise ValueError("大模型回复中缺少state") nextStageCode = normalize_stage_code(nextStageCode) nextStage = STATUS_CODE_MAP.get(nextStageCode, '') - form_update = extract_form_update_from_flow_nodes(data.get("responseData", [])) + form_update = result.form_update logger.info( "Chat non-stream completed " f"sessionId={json_data['sessionId']} " diff --git a/src/backends/__init__.py b/src/backends/__init__.py new file mode 100644 index 0000000..0f4deb3 --- /dev/null +++ b/src/backends/__init__.py @@ -0,0 +1,27 @@ +"""Backend adapters used by the public API layer.""" + +from .chat import ( + ChatBackend, + ChatBackendAPIError, + ChatBackendAuthenticationError, + ChatBackendError, + ChatBackendRateLimitError, + ChatInput, + ChatResult, + FormUpdate, + TextDelta, +) +from .fastgpt import FastGPTBackend + +__all__ = [ + "ChatBackend", + "ChatBackendAPIError", + "ChatBackendAuthenticationError", + "ChatBackendError", + "ChatBackendRateLimitError", + "ChatInput", + "ChatResult", + "FastGPTBackend", + "FormUpdate", + "TextDelta", +] diff --git a/src/backends/chat.py b/src/backends/chat.py new file mode 100644 index 0000000..afb759b --- /dev/null +++ b/src/backends/chat.py @@ -0,0 +1,72 @@ +"""Backend-neutral chat contract.""" + +from collections.abc import AsyncIterator +from dataclasses import dataclass, field +from typing import Any, Protocol + + +@dataclass(frozen=True) +class ChatInput: + """Input shared by all chat backend implementations.""" + + session_id: str + text: str + need_form_update: bool = False + + +@dataclass(frozen=True) +class TextDelta: + """A piece of raw model text. + + The text may contain a partial ``...`` prefix. Parsing that + public protocol remains the responsibility of the FastAPI layer. + """ + + text: str + + +@dataclass(frozen=True) +class FormUpdate: + """A structured form update produced alongside model text.""" + + data: Any + + +ChatStreamEvent = TextDelta | FormUpdate + + +@dataclass(frozen=True) +class ChatResult: + """Backend-neutral result for a non-streaming chat request.""" + + content: str + status_code: str | None = None + form_update: Any = field(default_factory=dict) + + +class ChatBackend(Protocol): + """Contract implemented by FastGPT today and LangGraph later.""" + + def stream(self, chat_input: ChatInput) -> AsyncIterator[ChatStreamEvent]: + """Stream raw text and structured side-channel events.""" + ... + + async def complete(self, chat_input: ChatInput) -> ChatResult: + """Return one complete backend-neutral chat result.""" + ... + + +class ChatBackendError(Exception): + """Base error raised by a chat backend adapter.""" + + +class ChatBackendAuthenticationError(ChatBackendError): + """The backend rejected its configured credentials.""" + + +class ChatBackendRateLimitError(ChatBackendError): + """The backend rejected the request because of rate limiting.""" + + +class ChatBackendAPIError(ChatBackendError): + """The backend API failed or returned an invalid response.""" diff --git a/src/backends/fastgpt.py b/src/backends/fastgpt.py new file mode 100644 index 0000000..14f8747 --- /dev/null +++ b/src/backends/fastgpt.py @@ -0,0 +1,148 @@ +"""FastGPT implementation of the backend-neutral chat contract.""" + +import json +from collections.abc import AsyncIterator +from typing import Any + +from fastgpt_client import AsyncChatClient, aiter_stream_events +from fastgpt_client.exceptions import APIError, AuthenticationError, RateLimitError + +from .chat import ( + ChatBackendAPIError, + ChatBackendAuthenticationError, + ChatBackendRateLimitError, + ChatInput, + ChatResult, + FormUpdate, + TextDelta, +) + + +FORM_EXTRACT_MODULE_NAME = "文本内容提取事故信息" + + +def _parse_json_value(value: Any) -> Any: + parsed = value + for _ in range(3): + if not isinstance(parsed, str): + return parsed + parsed = parsed.strip() + if not parsed: + return {} + try: + parsed = json.loads(parsed) + except json.JSONDecodeError: + return parsed + return parsed + + +def _extract_form_update(nodes: Any) -> Any: + if not isinstance(nodes, list): + return {} + + for node in nodes: + if not isinstance(node, dict): + continue + if node.get("moduleName") != FORM_EXTRACT_MODULE_NAME: + continue + + extract_result = node.get("extractResult", {}) + if not isinstance(extract_result, dict): + return {} + + form_update = extract_result.get("formUpdate", "") + return _parse_json_value(form_update) if form_update else {} + + return {} + + +def _extract_content(data: dict[str, Any]) -> str: + try: + content = data["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError) as exc: + raise ChatBackendAPIError("FastGPT response is missing message content") from exc + + if isinstance(content, list): + try: + content = content[0]["text"]["content"] + except (KeyError, IndexError, TypeError) as exc: + raise ChatBackendAPIError( + "FastGPT response contains invalid list content" + ) from exc + + if not isinstance(content, str): + raise ChatBackendAPIError("FastGPT message content is not text") + return content + + +class FastGPTBackend: + """Translate FastGPT SDK calls and events into the neutral chat contract.""" + + def __init__(self, client: AsyncChatClient): + self._client = client + + async def stream(self, chat_input: ChatInput) -> AsyncIterator[TextDelta | FormUpdate]: + try: + response = await self._client.create_chat_completion( + messages=[{"role": "user", "content": chat_input.text}], + chatId=chat_input.session_id, + stream=True, + detail=True, + variables={"needFormUpdate": chat_input.need_form_update}, + ) + + async for event in aiter_stream_events(response): + if event.kind == "flowResponses": + form_update = _extract_form_update(event.data) + if form_update: + yield FormUpdate(form_update) + continue + + if event.kind not in {"answer", "fastAnswer", "data"}: + continue + if not isinstance(event.data, dict): + continue + + try: + content = event.data["choices"][0]["delta"].get("content", "") + except (KeyError, IndexError, TypeError, AttributeError): + content = "" + if content: + yield TextDelta(content) + except AuthenticationError as exc: + raise ChatBackendAuthenticationError(str(exc)) from exc + except RateLimitError as exc: + raise ChatBackendRateLimitError(str(exc)) from exc + except APIError as exc: + raise ChatBackendAPIError(str(exc)) from exc + + async def complete(self, chat_input: ChatInput) -> ChatResult: + try: + response = await self._client.create_chat_completion( + messages=[{"role": "user", "content": chat_input.text}], + chatId=chat_input.session_id, + stream=False, + detail=True, + variables={"needFormUpdate": chat_input.need_form_update}, + ) + response.raise_for_status() + data = response.json() + except AuthenticationError as exc: + raise ChatBackendAuthenticationError(str(exc)) from exc + except RateLimitError as exc: + raise ChatBackendRateLimitError(str(exc)) from exc + except APIError as exc: + raise ChatBackendAPIError(str(exc)) from exc + + if not isinstance(data, dict): + raise ChatBackendAPIError("FastGPT response body is not an object") + + status_code = data.get("newVariables", {}).get("status_code") + if status_code is not None: + status_code = str(status_code) + + return ChatResult( + content=_extract_content(data), + status_code=status_code, + form_update=_extract_form_update(data.get("responseData", [])), + ) diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py index ee582d8..15b82cd 100644 --- a/src/core/fastgpt_client.py +++ b/src/core/fastgpt_client.py @@ -1,17 +1,20 @@ -"""FastGPT client dependency injection.""" +"""FastGPT client and chat backend dependency injection.""" from contextlib import asynccontextmanager from fastapi import FastAPI from fastgpt_client import AsyncChatClient +from ..backends.chat import ChatBackend +from ..backends.fastgpt import FastGPTBackend from .config import Config # Global client instance _fastgpt_client: AsyncChatClient | None = None +_chat_backend: ChatBackend | None = None @asynccontextmanager async def lifespan(app: FastAPI): """Manage FastGPT client lifecycle.""" - global _fastgpt_client + global _chat_backend, _fastgpt_client Config.validate() # Initialize client @@ -24,12 +27,15 @@ async def lifespan(app: FastAPI): enable_logging=Config.DEBUG, ) await _fastgpt_client.__aenter__() - - yield - - # Cleanup - if _fastgpt_client: - await _fastgpt_client.__aexit__(None, None, None) + _chat_backend = FastGPTBackend(_fastgpt_client) + + try: + yield + finally: + _chat_backend = None + if _fastgpt_client: + await _fastgpt_client.__aexit__(None, None, None) + _fastgpt_client = None def get_fastgpt_client() -> AsyncChatClient: @@ -37,3 +43,10 @@ def get_fastgpt_client() -> AsyncChatClient: if _fastgpt_client is None: raise RuntimeError("FastGPT client not initialized") return _fastgpt_client + + +def get_chat_backend() -> ChatBackend: + """Get the backend-neutral chat service.""" + if _chat_backend is None: + raise RuntimeError("Chat backend not initialized") + return _chat_backend diff --git a/test/api/test_chat_backend_boundary.py b/test/api/test_chat_backend_boundary.py new file mode 100644 index 0000000..83135ea --- /dev/null +++ b/test/api/test_chat_backend_boundary.py @@ -0,0 +1,70 @@ +import pytest + +from src.api.endpoints import chat +from src.backends.chat import ChatInput, ChatResult, FormUpdate, TextDelta +from src.schemas.models import ProcessRequest_chat + + +class FakeBackend: + def __init__(self): + self.received = [] + + async def stream(self, chat_input: ChatInput): + self.received.append(chat_input) + yield TextDelta("1002你") + yield FormUpdate({"jdcsl": 2}) + yield TextDelta("好") + + async def complete(self, chat_input: ChatInput): + self.received.append(chat_input) + return ChatResult( + content="1002你好", + status_code="1002", + form_update={"jdcsl": 2}, + ) + + +def make_request(): + return ProcessRequest_chat( + sessionId="session-001", + timeStamp="20260725120000", + text="发生了交通事故", + needFormUpdate=True, + ) + + +@pytest.mark.asyncio +async def test_non_stream_chat_uses_backend_neutral_result(): + backend = FakeBackend() + + response = await chat(make_request(), stream=False, backend=backend) + + assert response.outputText == "你好" + assert response.nextStageCode == "1002" + assert response.formUpdate == {"jdcsl": 2} + assert backend.received == [ + ChatInput( + session_id="session-001", + text="发生了交通事故", + need_form_update=True, + ) + ] + + +@pytest.mark.asyncio +async def test_stream_chat_keeps_state_prefix_buffering_in_fastapi_layer(): + backend = FakeBackend() + + response = await chat(make_request(), stream=True, backend=backend) + chunks = [] + async for chunk in response.body_iterator: + chunks.append(chunk.decode() if isinstance(chunk, bytes) else chunk) + body = "".join(chunks) + + assert body.index("event: stage_code") < body.index("event: text_delta") + assert '"nextStageCode": "1002"' in body + assert '"text": "你"' in body + assert '"text": "好"' in body + assert "event: formUpdate" in body + assert "event: done" in body diff --git a/test/api/test_public_schema_contract.py b/test/api/test_public_schema_contract.py new file mode 100644 index 0000000..6fe4bc6 --- /dev/null +++ b/test/api/test_public_schema_contract.py @@ -0,0 +1,195 @@ +"""Characterization tests for the public HTTP API schemas. + +These tests freeze the current contract before the FastGPT backend is replaced. +They should change only when the teams integrating with this service agree to a +contract change. +""" + +from collections.abc import Callable +from typing import Any + +import pytest +from pydantic import BaseModel, ValidationError + +from src.schemas.models import ( + ProcessRequest_chat, + ProcessRequest_get, + ProcessRequest_set, + ProcessResponse_chat, + ProcessResponse_get, + ProcessResponse_set, +) + + +SESSION_ID = "session-001" +TIMESTAMP = "20260725120000" + + +def chat_request_payload() -> dict[str, Any]: + return { + "sessionId": SESSION_ID, + "timeStamp": TIMESTAMP, + "text": "发生了交通事故", + } + + +def chat_response_payload() -> dict[str, Any]: + return { + "sessionId": SESSION_ID, + "timeStamp": TIMESTAMP, + "outputText": "请描述事故经过。", + "nextStage": "通话中", + "nextStageCode": "1002", + "code": "200", + } + + +def get_request_payload() -> dict[str, Any]: + return { + "sessionId": SESSION_ID, + "timeStamp": TIMESTAMP, + "key": "acdinfo", + } + + +def get_response_payload() -> dict[str, Any]: + return { + "sessionId": SESSION_ID, + "timeStamp": TIMESTAMP, + "value": '{"jdcsl": "2"}', + "code": "200", + } + + +def set_request_payload() -> dict[str, Any]: + return { + "sessionId": SESSION_ID, + "timeStamp": TIMESTAMP, + "key": "hphm1", + "value": "沪A12345", + } + + +def set_response_payload() -> dict[str, Any]: + return { + "sessionId": SESSION_ID, + "timeStamp": TIMESTAMP, + "code": "200", + } + + +def test_chat_request_defaults_are_backward_compatible() -> None: + request = ProcessRequest_chat(**chat_request_payload()) + + assert request.model_dump() == { + **chat_request_payload(), + "needFormUpdate": False, + "useTextChunk": False, + } + + +def test_chat_response_shape_is_backward_compatible() -> None: + response = ProcessResponse_chat(**chat_response_payload()) + + assert response.model_dump() == { + **chat_response_payload(), + "formUpdate": {}, + "msg": None, + } + + +def test_get_info_shapes_are_backward_compatible() -> None: + request = ProcessRequest_get(**get_request_payload()) + response = ProcessResponse_get(**get_response_payload()) + + assert request.model_dump() == { + **get_request_payload(), + "includeInputInfo": False, + } + assert response.model_dump() == { + **get_response_payload(), + "msg": None, + } + + +def test_set_info_shapes_are_backward_compatible() -> None: + request = ProcessRequest_set(**set_request_payload()) + response = ProcessResponse_set(**set_response_payload()) + + assert request.model_dump() == { + **set_request_payload(), + "includeInputInfo": False, + } + assert response.model_dump() == { + **set_response_payload(), + "msg": None, + } + + +@pytest.mark.parametrize( + ("model", "payload_factory", "required_field"), + [ + (ProcessRequest_chat, chat_request_payload, "sessionId"), + (ProcessRequest_chat, chat_request_payload, "timeStamp"), + (ProcessRequest_chat, chat_request_payload, "text"), + (ProcessRequest_get, get_request_payload, "key"), + (ProcessRequest_set, set_request_payload, "key"), + (ProcessRequest_set, set_request_payload, "value"), + (ProcessResponse_chat, chat_response_payload, "nextStageCode"), + (ProcessResponse_get, get_response_payload, "value"), + ], +) +def test_required_fields_remain_required( + model: type[BaseModel], + payload_factory: Callable[[], dict[str, Any]], + required_field: str, +) -> None: + payload = payload_factory() + payload.pop(required_field) + + with pytest.raises(ValidationError): + model(**payload) + + +@pytest.mark.parametrize( + ("model", "payload_factory", "field_name", "invalid_value"), + [ + (ProcessRequest_chat, chat_request_payload, "sessionId", "s" * 65), + (ProcessRequest_chat, chat_request_payload, "timeStamp", "t" * 33), + (ProcessRequest_get, get_request_payload, "sessionId", "s" * 65), + (ProcessRequest_set, set_request_payload, "timeStamp", "t" * 33), + (ProcessResponse_chat, chat_response_payload, "nextStage", "n" * 33), + (ProcessResponse_chat, chat_response_payload, "nextStageCode", "10020"), + (ProcessResponse_chat, chat_response_payload, "code", "10000"), + (ProcessResponse_get, get_response_payload, "code", "10000"), + (ProcessResponse_set, set_response_payload, "code", "10000"), + ], +) +def test_public_length_limits_are_enforced( + model: type[BaseModel], + payload_factory: Callable[[], dict[str, Any]], + field_name: str, + invalid_value: str, +) -> None: + payload = payload_factory() + payload[field_name] = invalid_value + + with pytest.raises(ValidationError): + model(**payload) + + +def test_get_info_value_remains_a_json_encoded_string() -> None: + response = ProcessResponse_get(**get_response_payload()) + + assert isinstance(response.value, str) + assert response.value == '{"jdcsl": "2"}' + + +def test_chat_form_update_remains_unstructured_for_compatibility() -> None: + form_update = {"jdcsl": 2, "ywrysw": False} + response = ProcessResponse_chat( + **chat_response_payload(), + formUpdate=form_update, + ) + + assert response.formUpdate == form_update diff --git a/test/backends/test_fastgpt_backend.py b/test/backends/test_fastgpt_backend.py new file mode 100644 index 0000000..db876d0 --- /dev/null +++ b/test/backends/test_fastgpt_backend.py @@ -0,0 +1,123 @@ +from types import SimpleNamespace + +import pytest + +from src.backends.chat import ChatInput, FormUpdate, TextDelta +from src.backends.fastgpt import FastGPTBackend + + +class FakeResponse: + def __init__(self, data): + self._data = data + self.raise_for_status_called = False + + def raise_for_status(self): + self.raise_for_status_called = True + + def json(self): + return self._data + + +class FakeClient: + def __init__(self, response): + self.response = response + self.calls = [] + + async def create_chat_completion(self, **kwargs): + self.calls.append(kwargs) + return self.response + + +@pytest.mark.asyncio +async def test_complete_translates_fastgpt_response_to_neutral_result(): + response = FakeResponse( + { + "choices": [ + { + "message": { + "content": "1002请描述事故经过。" + } + } + ], + "newVariables": {"status_code": "1002"}, + "responseData": [ + { + "moduleName": "文本内容提取事故信息", + "extractResult": { + "formUpdate": '{"jdcsl": 2, "ywrysw": false}' + }, + } + ], + } + ) + client = FakeClient(response) + backend = FastGPTBackend(client) + + result = await backend.complete( + ChatInput( + session_id="session-001", + text="发生了交通事故", + need_form_update=True, + ) + ) + + assert response.raise_for_status_called is True + assert result.content == "1002请描述事故经过。" + assert result.status_code == "1002" + assert result.form_update == {"jdcsl": 2, "ywrysw": False} + assert client.calls == [ + { + "messages": [{"role": "user", "content": "发生了交通事故"}], + "chatId": "session-001", + "stream": False, + "detail": True, + "variables": {"needFormUpdate": True}, + } + ] + + +@pytest.mark.asyncio +async def test_stream_translates_fastgpt_events_to_neutral_events(monkeypatch): + response = object() + client = FakeClient(response) + backend = FastGPTBackend(client) + + async def fake_aiter_stream_events(actual_response): + assert actual_response is response + yield SimpleNamespace( + kind="answer", + data={"choices": [{"delta": {"content": "1002"}}]}, + ) + yield SimpleNamespace( + kind="flowResponses", + data=[ + { + "moduleName": "文本内容提取事故信息", + "extractResult": {"formUpdate": '{"jdcsl": 2}'}, + } + ], + ) + yield SimpleNamespace( + kind="answer", + data={"choices": [{"delta": {"content": "你好"}}]}, + ) + yield SimpleNamespace(kind="ignored", data={}) + + monkeypatch.setattr( + "src.backends.fastgpt.aiter_stream_events", + fake_aiter_stream_events, + ) + + events = [ + event + async for event in backend.stream( + ChatInput(session_id="session-001", text="你好") + ) + ] + + assert events == [ + TextDelta("1002"), + FormUpdate({"jdcsl": 2}), + TextDelta("你好"), + ] + assert client.calls[0]["stream"] is True