wx44wx/ZNJJ-api-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor chat backend integration and introduce new backend-neutral architecture

Browse Source 
- 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.
This commit is contained in:
Eric Wang
2026-07-26 07:25:24 +08:00
parent 893eb2f2f7
commit 5c719ed2ea
8 changed files with 697 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/endpoints.py
+41 -122
View File
@@ -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"<state>\s*(\d+)\s*</state>", 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']} "
src/backends/__init__.py
+27
View File
@@ -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",
]
src/backends/chat.py
+72
View File
@@ -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 ``<state>...</state>`` 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."""
src/backends/fastgpt.py
+148
View File
@@ -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", [])),
)
src/core/fastgpt_client.py
+21 -8
View File
@@ -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