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