From 3e9385b0ca9efb633b58cbe9db60bdb0ec5ff3bc Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Tue, 28 Jul 2026 08:41:34 +0800 Subject: [PATCH] Add client-mode FastGPT routing --- src/.env.example | 13 +++-- src/api/endpoints.py | 72 ++++++++++++++++------- src/core/config.py | 42 +++++++++---- src/core/fastgpt_client.py | 91 ++++++++++++++++++++--------- src/schemas/models.py | 10 +++- test/api/fastapi-browser-addon.http | 85 +++++++++++++++++++++++++++ test/test_client_mode.py | 76 ++++++++++++++++++++++++ 7 files changed, 325 insertions(+), 64 deletions(-) create mode 100644 test/api/fastapi-browser-addon.http create mode 100644 test/test_client_mode.py diff --git a/src/.env.example b/src/.env.example index f7c789d..5ea0222 100644 --- a/src/.env.example +++ b/src/.env.example @@ -3,11 +3,14 @@ SECRET_KEY=your_secret_key DEBUG=True ANALYSIS_SERVICE_URL=http://127.0.0.1:3030 -ANALYSIS_AUTH_TOKEN=fastgpt-hSPnXMoBNGVAEpTLkQT3YfAnN26gQSyvLd4ABL1MRDoh68nL4RDlopFHXqmH8 -APP_ID=683ea1bc86197e19f71fc1ae -DELETE_SESSION_URL=http://127.0.0.1:3030/api/core/chat/delHistory?chatId={chatId}&appId={appId} -DELETE_CHAT_URL=http://127.0.0.1:3030/api/core/chat/item/delete?contentId={contentId}&chatId={chatId}&appId={appId} -GET_CHAT_RECORDS_URL=http://127.0.0.1:3030/api/core/chat/getPaginationRecords +ANALYSIS_AUTH_TOKEN=replace-with-direct-fastgpt-api-key +APP_ID=replace-with-direct-fastgpt-app-id + +# Browser Addon FastGPT application. It shares ANALYSIS_SERVICE_URL unless +# FASTGPT_BROWSER_ADDON_BASE_URL is set. +FASTGPT_BROWSER_ADDON_API_KEY=replace-with-browser-addon-fastgpt-api-key +FASTGPT_BROWSER_ADDON_APP_ID=replace-with-browser-addon-fastgpt-app-id +# FASTGPT_BROWSER_ADDON_BASE_URL=http://127.0.0.1:3030 # Voice demo (Pipecat /ws-product). Relative to project root, or an absolute path. VOICE_CONFIG=config/voice.json diff --git a/src/api/endpoints.py b/src/api/endpoints.py index 1268f18..833b190 100644 --- a/src/api/endpoints.py +++ b/src/api/endpoints.py @@ -1,12 +1,11 @@ -from fastapi import APIRouter, HTTPException, Depends +from fastapi import APIRouter, HTTPException 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 ..core.fastgpt_client import get_fastgpt_client -from ..core.config import Config +from ..core.fastgpt_client import get_fastgpt_profile from ..utils.text_chunker import SentenceTextChunker, SentenceTextChunkerConfig from loguru import logger import json @@ -52,6 +51,13 @@ def normalize_stage_code(stage_code: str) -> str: return stage_code +def external_stage_code(client_mode: str, stage_code: str) -> str: + """Apply the external namespace required by the client mode.""" + if client_mode == "browser_addon": + return f"browser_addon.{stage_code}" + return stage_code + + def extract_first_state_and_clean_content(text: str) -> tuple[str | None, str]: """Return the first state code and content with all state tags removed.""" match = STATE_TAG_PATTERN.search(text) @@ -107,13 +113,14 @@ def format_set_info_input(payload: dict, include_input_info: bool) -> str: async def delete_last_two_chat_records( client: AsyncChatClient, - session_id: str + app_id: str, + session_id: str, ) -> None: """Delete the last two chat records.""" try: # Get chat records using SDK response = await client.get_chat_records( - appId=Config.FASTGPT_APP_ID, + appId=app_id, chatId=session_id, offset=0, pageSize=10 @@ -132,7 +139,7 @@ async def delete_last_two_chat_records( # Delete records using SDK for content_id in last_two_data_ids: delete_response = await client.delete_chat_record( - appId=Config.FASTGPT_APP_ID, + appId=app_id, chatId=session_id, contentId=content_id ) @@ -156,17 +163,19 @@ def create_sse_event(event: str, data: dict) -> str: async def chat( request: ProcessRequest_chat, stream: bool = False, - client: AsyncChatClient = Depends(get_fastgpt_client) ): """Handle chat completion request.""" json_data = request.model_dump() + client_mode = json_data['clientMode'] + profile = get_fastgpt_profile(client_mode) + client = profile.client need_form_update = json_data.get('needFormUpdate', False) use_text_chunk = json_data.get('useTextChunk', False) chat_variables = {'needFormUpdate': need_form_update} request_started_at = time.perf_counter() logger.info( "Chat request received " - f"sessionId={json_data['sessionId']} stream={stream} " + f"sessionId={json_data['sessionId']} clientMode={client_mode} stream={stream} " f"needFormUpdate={need_form_update} useTextChunk={use_text_chunk} " f"text_len={len(json_data.get('text', ''))} " f"input={json_data.get('text', '')!r}" @@ -325,11 +334,17 @@ async def chat( if state_code: # Apply logic to map/adjust state code - nextStageCode = normalize_stage_code(state_code) - nextStage = STATUS_CODE_MAP.get(nextStageCode, '') + normalized_stage_code = normalize_stage_code(state_code) + nextStageCode = external_stage_code( + client_mode, + normalized_stage_code, + ) + nextStage = STATUS_CODE_MAP.get(normalized_stage_code, '') logger.info( "FastGPT stream stage_code parsed " f"sessionId={json_data['sessionId']} " + f"clientMode={client_mode} " + f"rawStageCode={state_code} " f"nextStageCode={nextStageCode} nextStage={nextStage}" ) @@ -509,13 +524,15 @@ 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 = normalize_stage_code(nextStageCode) - nextStage = STATUS_CODE_MAP.get(nextStageCode, '') + raw_stage_code = content_stage_code or data['newVariables']['status_code'] + normalized_stage_code = normalize_stage_code(raw_stage_code) + nextStageCode = external_stage_code(client_mode, normalized_stage_code) + nextStage = STATUS_CODE_MAP.get(normalized_stage_code, '') form_update = extract_form_update_from_flow_nodes(data.get("responseData", [])) logger.info( "Chat non-stream completed " f"sessionId={json_data['sessionId']} " + f"clientMode={client_mode} rawStageCode={raw_stage_code} " f"duration_ms={(time.perf_counter() - request_started_at) * 1000:.1f} " f"nextStageCode={nextStageCode} nextStage={nextStage} " f"output_len={len(content)} formUpdate_type={type(form_update).__name__} " @@ -549,10 +566,11 @@ async def chat( @router.post("/set_info", response_model=ProcessResponse_set) async def set_info( request: ProcessRequest_set, - client: AsyncChatClient = Depends(get_fastgpt_client) ): """Set information in chat state.""" json_data = request.model_dump() + profile = get_fastgpt_profile(json_data['clientMode']) + client = profile.client set_info_payload = {'key': json_data['key'], 'value': json_data['value']} set_info_input = format_set_info_input( set_info_payload, @@ -585,7 +603,11 @@ async def set_info( ) try: - await delete_last_two_chat_records(client, json_data['sessionId']) + await delete_last_two_chat_records( + client, + profile.app_id, + json_data['sessionId'], + ) except Exception as e: logger.error(f"Error deleting chat records: {e}") return ProcessResponse_set( @@ -614,7 +636,11 @@ async def set_info( response.raise_for_status() # Delete records again after update - await delete_last_two_chat_records(client, json_data['sessionId']) + await delete_last_two_chat_records( + client, + profile.app_id, + json_data['sessionId'], + ) return ProcessResponse_set( sessionId=json_data['sessionId'], @@ -635,10 +661,11 @@ async def set_info( @router.post("/get_info", response_model=ProcessResponse_get) async def get_info( request: ProcessRequest_get, - client: AsyncChatClient = Depends(get_fastgpt_client) ): """Get information from chat state.""" json_data = request.model_dump() + profile = get_fastgpt_profile(json_data['clientMode']) + client = profile.client get_info_payload = {'key': json_data['key']} get_info_input = format_set_info_input( get_info_payload, @@ -672,7 +699,11 @@ async def get_info( ) try: - await delete_last_two_chat_records(client, json_data['sessionId']) + await delete_last_two_chat_records( + client, + profile.app_id, + json_data['sessionId'], + ) except Exception as e: logger.error(f"Error deleting records: {e}") return ProcessResponse_get( @@ -737,10 +768,11 @@ async def get_info( @router.delete("/delete_session", response_model=ProcessResponse_delete_session) async def delete_session( request: ProcessRequest_delete_session, - client: AsyncChatClient = Depends(get_fastgpt_client) ): """Delete a chat session.""" json_data = request.model_dump() + profile = get_fastgpt_profile(json_data['clientMode']) + client = profile.client chat_id = json_data.get('sessionId') if not chat_id: @@ -754,7 +786,7 @@ async def delete_session( try: # Use SDK's delete_chat_history response = await client.delete_chat_history( - appId=Config.FASTGPT_APP_ID, + appId=profile.app_id, chatId=chat_id ) response.raise_for_status() diff --git a/src/core/config.py b/src/core/config.py index 3692aba..df70931 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -10,21 +10,43 @@ class Config: DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db") DEBUG = os.getenv("DEBUG", "false").lower() in ("true", "1", "t") - # FastGPT Configuration - FASTGPT_API_KEY = os.getenv("ANALYSIS_AUTH_TOKEN") - FASTGPT_BASE_URL = os.getenv("ANALYSIS_SERVICE_URL") - FASTGPT_APP_ID = os.getenv("APP_ID") - DELETE_SESSION_URL = os.getenv("DELETE_SESSION_URL") - DELETE_CHAT_URL = os.getenv("DELETE_CHAT_URL") - GET_CHAT_RECORDS_URL = os.getenv("GET_CHAT_RECORDS_URL") + # FastGPT Configuration. Direct mode falls back to the legacy environment + # variable names so existing deployments keep working unchanged. + FASTGPT_DIRECT_API_KEY = ( + os.getenv("FASTGPT_DIRECT_API_KEY") + or os.getenv("ANALYSIS_AUTH_TOKEN") + ) + FASTGPT_DIRECT_BASE_URL = ( + os.getenv("FASTGPT_DIRECT_BASE_URL") + or os.getenv("ANALYSIS_SERVICE_URL") + ) + FASTGPT_DIRECT_APP_ID = ( + os.getenv("FASTGPT_DIRECT_APP_ID") + or os.getenv("APP_ID") + ) + + FASTGPT_BROWSER_ADDON_API_KEY = os.getenv("FASTGPT_BROWSER_ADDON_API_KEY") + FASTGPT_BROWSER_ADDON_BASE_URL = ( + os.getenv("FASTGPT_BROWSER_ADDON_BASE_URL") + or FASTGPT_DIRECT_BASE_URL + ) + FASTGPT_BROWSER_ADDON_APP_ID = os.getenv("FASTGPT_BROWSER_ADDON_APP_ID") + + # Backward-compatible aliases for code outside the mode-aware API layer. + FASTGPT_API_KEY = FASTGPT_DIRECT_API_KEY + FASTGPT_BASE_URL = FASTGPT_DIRECT_BASE_URL + FASTGPT_APP_ID = FASTGPT_DIRECT_APP_ID @classmethod def validate(cls): """Validate required configuration.""" required = [ - ("FASTGPT_API_KEY", cls.FASTGPT_API_KEY), - ("FASTGPT_APP_ID", cls.FASTGPT_APP_ID), + ("FASTGPT_DIRECT_BASE_URL (or ANALYSIS_SERVICE_URL)", cls.FASTGPT_DIRECT_BASE_URL), + ("FASTGPT_DIRECT_API_KEY (or ANALYSIS_AUTH_TOKEN)", cls.FASTGPT_DIRECT_API_KEY), + ("FASTGPT_DIRECT_APP_ID (or APP_ID)", cls.FASTGPT_DIRECT_APP_ID), + ("FASTGPT_BROWSER_ADDON_API_KEY", cls.FASTGPT_BROWSER_ADDON_API_KEY), + ("FASTGPT_BROWSER_ADDON_APP_ID", cls.FASTGPT_BROWSER_ADDON_APP_ID), ] missing = [name for name, value in required if not value] if missing: - raise ValueError(f"Missing required environment variables: {', '.join(missing)}") \ No newline at end of file + raise ValueError(f"Missing required environment variables: {', '.join(missing)}") diff --git a/src/core/fastgpt_client.py b/src/core/fastgpt_client.py index ee582d8..ee6dfed 100644 --- a/src/core/fastgpt_client.py +++ b/src/core/fastgpt_client.py @@ -1,39 +1,76 @@ """FastGPT client dependency injection.""" -from contextlib import asynccontextmanager +from contextlib import AsyncExitStack, asynccontextmanager +from dataclasses import dataclass from fastapi import FastAPI from fastgpt_client import AsyncChatClient from .config import Config -# Global client instance -_fastgpt_client: AsyncChatClient | None = None + +@dataclass(frozen=True) +class FastGPTProfile: + """FastGPT client and app metadata for one client mode.""" + + client: AsyncChatClient + app_id: str + + +_fastgpt_profiles: dict[str, FastGPTProfile] = {} + + +def _profile_settings() -> dict[str, dict[str, str | None]]: + return { + "direct": { + "api_key": Config.FASTGPT_DIRECT_API_KEY, + "base_url": Config.FASTGPT_DIRECT_BASE_URL, + "app_id": Config.FASTGPT_DIRECT_APP_ID, + }, + "browser_addon": { + "api_key": Config.FASTGPT_BROWSER_ADDON_API_KEY, + "base_url": Config.FASTGPT_BROWSER_ADDON_BASE_URL, + "app_id": Config.FASTGPT_BROWSER_ADDON_APP_ID, + }, + } @asynccontextmanager async def lifespan(app: FastAPI): - """Manage FastGPT client lifecycle.""" - global _fastgpt_client + """Manage all mode-specific FastGPT clients.""" Config.validate() - - # Initialize client - _fastgpt_client = AsyncChatClient( - api_key=Config.FASTGPT_API_KEY, - base_url=Config.FASTGPT_BASE_URL, - timeout=60.0, - max_retries=3, - retry_delay=1.0, - enable_logging=Config.DEBUG, - ) - await _fastgpt_client.__aenter__() - - yield - - # Cleanup - if _fastgpt_client: - await _fastgpt_client.__aexit__(None, None, None) + + async with AsyncExitStack() as stack: + for client_mode, settings in _profile_settings().items(): + client = AsyncChatClient( + api_key=settings["api_key"], + base_url=settings["base_url"], + timeout=60.0, + max_retries=3, + retry_delay=1.0, + enable_logging=Config.DEBUG, + ) + await stack.enter_async_context(client) + _fastgpt_profiles[client_mode] = FastGPTProfile( + client=client, + app_id=settings["app_id"], + ) + + try: + yield + finally: + _fastgpt_profiles.clear() -def get_fastgpt_client() -> AsyncChatClient: - """Get the FastGPT client instance.""" - if _fastgpt_client is None: - raise RuntimeError("FastGPT client not initialized") - return _fastgpt_client +def get_fastgpt_profile(client_mode: str) -> FastGPTProfile: + """Get the initialized FastGPT profile for a client mode.""" + try: + return _fastgpt_profiles[client_mode] + except KeyError as exc: + if client_mode not in _profile_settings(): + raise ValueError(f"Unsupported clientMode: {client_mode}") from exc + raise RuntimeError( + f"FastGPT profile is not initialized for clientMode={client_mode}" + ) from exc + + +def get_fastgpt_client(client_mode: str = "direct") -> AsyncChatClient: + """Get a FastGPT client, defaulting to the legacy direct mode.""" + return get_fastgpt_profile(client_mode).client diff --git a/src/schemas/models.py b/src/schemas/models.py index a7425fa..592c710 100644 --- a/src/schemas/models.py +++ b/src/schemas/models.py @@ -1,10 +1,13 @@ from pydantic import BaseModel, Field -from typing import Any, Optional +from typing import Any, Literal, Optional + +ClientMode = Literal["direct", "browser_addon"] class ProcessRequest_chat(BaseModel): sessionId: str = Field(..., max_length=64) timeStamp: str = Field(..., max_length=32) text: str = Field(...) + clientMode: ClientMode = "direct" needFormUpdate: bool = False useTextChunk: bool = False @@ -14,7 +17,7 @@ class ProcessResponse_chat(BaseModel): outputText: str = Field(...) formUpdate: Any = Field(default_factory=dict) nextStage: str = Field(..., max_length=32) - nextStageCode: str = Field(..., max_length=4) + nextStageCode: str = Field(..., max_length=32) code: str = Field(..., max_length=4) msg: Optional[str] = None @@ -22,6 +25,7 @@ class ProcessRequest_get(BaseModel): sessionId: str = Field(..., max_length=64) timeStamp: str = Field(..., max_length=32) key: str = Field(...) + clientMode: ClientMode = "direct" includeInputInfo: bool = False class ProcessResponse_get(BaseModel): @@ -36,6 +40,7 @@ class ProcessRequest_set(BaseModel): timeStamp: str = Field(..., max_length=32) key: str = Field(...) value: str = Field(...) + clientMode: ClientMode = "direct" includeInputInfo: bool = False class ProcessResponse_set(BaseModel): @@ -47,6 +52,7 @@ class ProcessResponse_set(BaseModel): class ProcessRequest_delete_session(BaseModel): sessionId: str = Field(..., max_length=64) timeStamp: str = Field(..., max_length=32) + clientMode: ClientMode = "direct" class ProcessResponse_delete_session(BaseModel): sessionId: str = Field(..., max_length=64) diff --git a/test/api/fastapi-browser-addon.http b/test/api/fastapi-browser-addon.http new file mode 100644 index 0000000..932d780 --- /dev/null +++ b/test/api/fastapi-browser-addon.http @@ -0,0 +1,85 @@ +@baseUrl = http://127.0.0.1:8000 +### +@sessionId = a1123 +@timeStamp = 202603310303 +@clientMode = browser_addon +### +GET {{baseUrl}} + +HTTP/1.1 200 - OK +date: Tue, 28 Jul 2026 00:36:56 GMT +server: uvicorn +content-length: 32 +content-type: application/json +connection: close +### +DELETE {{baseUrl}}/delete_session +content-type: application/json + +{ + "sessionId": "{{sessionId}}", + "timeStamp": "{{$timestamp}}", + "clientMode": "{{clientMode}}" +} + +HTTP/1.1 200 - OK +date: Tue, 28 Jul 2026 00:36:57 GMT +server: uvicorn +content-length: 71 +content-type: application/json +connection: close +### +POST {{baseUrl}}/chat?stream=true +content-type: application/json + +{ + "sessionId": "{{sessionId}}", + "timeStamp": "{{timeStamp}}", + "text": "hi", + "clientMode": "{{clientMode}}", + "needFormUpdate": true, + "useTextChunk": true +} + +HTTP/1.1 200 - OK +date: Tue, 28 Jul 2026 00:36:58 GMT +server: uvicorn +content-type: text/event-stream; charset=utf-8 +connection: close +transfer-encoding: chunked +### +POST {{baseUrl}}/get_info +content-type: application/json + +{ + "sessionId": "{{sessionId}}", + "timeStamp": "{{timeStamp}}", + "key": "hphm1", + "clientMode": "{{clientMode}}" +} + +HTTP/1.1 200 - OK +date: Tue, 28 Jul 2026 00:37:07 GMT +server: uvicorn +content-length: 97 +content-type: application/json +connection: close +### +POST {{baseUrl}}/set_info +content-type: application/json + +{ + "sessionId": "{{sessionId}}", + "timeStamp": "{{timeStamp}}", + "key": "hphm1", + "value": "沪A8939", + "clientMode": "{{clientMode}}" +} + +HTTP/1.1 200 - OK +date: Tue, 28 Jul 2026 00:37:00 GMT +server: uvicorn +content-length: 70 +content-type: application/json +connection: close +### diff --git a/test/test_client_mode.py b/test/test_client_mode.py new file mode 100644 index 0000000..a3a5dbb --- /dev/null +++ b/test/test_client_mode.py @@ -0,0 +1,76 @@ +import pytest +from pydantic import ValidationError + +from src.api.endpoints import external_stage_code +from src.schemas.models import ( + ProcessRequest_chat, + ProcessRequest_delete_session, + ProcessRequest_get, + ProcessRequest_set, +) + + +@pytest.mark.parametrize( + ("request_model", "payload"), + [ + ( + ProcessRequest_chat, + {"sessionId": "session-1", "timeStamp": "1", "text": "你好"}, + ), + ( + ProcessRequest_set, + { + "sessionId": "session-1", + "timeStamp": "1", + "key": "name", + "value": "张三", + }, + ), + ( + ProcessRequest_get, + {"sessionId": "session-1", "timeStamp": "1", "key": "name"}, + ), + ( + ProcessRequest_delete_session, + {"sessionId": "session-1", "timeStamp": "1"}, + ), + ], +) +def test_client_mode_defaults_to_direct(request_model, payload): + request = request_model(**payload) + + assert request.clientMode == "direct" + + +def test_browser_addon_client_mode_is_accepted(): + request = ProcessRequest_chat( + sessionId="session-1", + timeStamp="1", + text="你好", + clientMode="browser_addon", + ) + + assert request.clientMode == "browser_addon" + + +def test_unknown_client_mode_is_rejected(): + with pytest.raises(ValidationError): + ProcessRequest_chat( + sessionId="session-1", + timeStamp="1", + text="你好", + clientMode="unknown", + ) + + +@pytest.mark.parametrize( + ("client_mode", "raw_code", "expected"), + [ + ("direct", "0001", "0001"), + ("direct", "1002", "1002"), + ("browser_addon", "0001", "browser_addon.0001"), + ("browser_addon", "1002", "browser_addon.1002"), + ], +) +def test_external_stage_code(client_mode, raw_code, expected): + assert external_stage_code(client_mode, raw_code) == expected