Add stream text output

.gitignore

@@ -1,2 +1,5 @@
 *.pyc
 __pycache__/
+logs/
+*.log
+.env

examples/stream_chat.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""
+Simple CLI script to interact with /chat endpoint in stream mode.
+"""
+
+import asyncio
+import aiohttp
+import json
+import sys
+from datetime import datetime
+
+
+API_BASE_URL = "http://localhost:8000"
+
+
+async def stream_chat(session_id: str, text: str):
+    """Send a streaming chat request."""
+    timestamp = datetime.now().isoformat()
+
+    payload = {
+        "sessionId": session_id,
+        "timeStamp": timestamp,
+        "text": text
+    }
+
+    async with aiohttp.ClientSession() as http_session:
+        async with http_session.post(
+            f"{API_BASE_URL}/chat",
+            json=payload,
+            params={"stream": "true"},
+        ) as response:
+            print(f"Status: {response.status}")
+            print("-" * 50)
+
+            buffer = ""
+            # Use async iterator on response.content
+            async for chunk in response.content.iter_chunked(1024):
+                chunk_str = chunk.decode("utf-8")
+                buffer += chunk_str
+
+                # Process complete SSE messages
+                while "\n\n" in buffer:
+                    message, buffer = buffer.split("\n\n", 1)
+                    if message.startswith("event: "):
+                        event_type = message[7:].split("\n")[0]
+                        data_line = message.split("data: ", 1)[-1]
+                        try:
+                            data = json.loads(data_line)
+                            print(f"[{event_type}] {json.dumps(data, ensure_ascii=False)}")
+                        except json.JSONDecodeError:
+                            print(f"[{event_type}] {data_line}")
+                    elif message.startswith("data: "):
+                        data_str = message[6:]
+                        try:
+                            data = json.loads(data_str)
+                            print(f"[data] {json.dumps(data, ensure_ascii=False)}")
+                        except json.JSONDecodeError:
+                            print(f"[data] {data_str}")
+
+
+async def main():
+    if len(sys.argv) < 3:
+        print("Usage: python stream_chat.py <session_id> <message>")
+        print("Example: python stream_chat.py test-session-123 '发生了交通事故'")
+        sys.exit(1)
+
+    session_id = sys.argv[1]
+    text = " ".join(sys.argv[2:])
+
+    print(f"Session ID: {session_id}")
+    print(f"Message: {text}")
+    print("-" * 50)
+
+    await stream_chat(session_id, text)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

requirements.txt

@@ -6,3 +6,15 @@ httpx>=0.25.0
 pytest>=7.4.0
 pytest-asyncio>=0.21.0
 pytest-cov>=4.1.0
+pillow>=10.4.0
+paho-mqtt>=2.1.0
+pydantic-settings==2.1.0
+python-multipart==0.0.6
+python-jose[cryptography]==3.3.0
+passlib[bcrypt]==1.7.4
+openai==1.55.3
+loguru>=0.7.0
+pandas
+requests
+sqlalchemy
+pymysql

src/.env.example

@@ -0,0 +1,10 @@
+DATABASE_URL=sqlite:///./test.db
+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

src/api/endpoints.py

@@ -1,37 +1,22 @@
-from fastapi import APIRouter, HTTPException
+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
-import httpx
+from fastgpt_client import AsyncChatClient
-from datetime import datetime
+from fastgpt_client.exceptions import (
-import os
+    APIError, AuthenticationError, RateLimitError, ValidationError
-from dotenv import load_dotenv
+)
-from urllib import parse as parse
+from ..core.fastgpt_client import get_fastgpt_client
+from ..core.config import Config
+from loguru import logger
 import json
-
+import re
-load_dotenv()
 
 router = APIRouter()
-
-ANALYSIS_SERVICE_URL = os.getenv("ANALYSIS_SERVICE_URL", None)
-if ANALYSIS_SERVICE_URL is None:
-    raise ValueError("ANALYSIS_SERVICE_URL environment variable not set")
-ANALYSIS_AUTH_TOKEN = os.getenv("ANALYSIS_AUTH_TOKEN", None)
-if ANALYSIS_AUTH_TOKEN is None:
-    raise ValueError("ANALYSIS_AUTH_TOKEN environment variable not set")
-DELETE_SESSION_URL = os.getenv("DELETE_SESSION_URL", None)
-if DELETE_SESSION_URL is None:
-    raise ValueError("DELETE_SESSION_URL environment variable not set")
-APP_ID = os.getenv("APP_ID", None)
-if APP_ID is None:
-    raise ValueError("APP_ID environment variable not set")
-DELETE_CHAT_URL = os.getenv("DELETE_CHAT_URL", None)
-if DELETE_CHAT_URL is None:
-    raise ValueError("DELETE_CHAT_URL environment variable not set")
-GET_CHAT_RECORDS_URL = os.getenv("GET_CHAT_RECORDS_URL", None)
-if GET_CHAT_RECORDS_URL is None:
-    raise ValueError("GET_CHAT_RECORDS_URL environment variable not set")
 STATUS_CODE_MAP = {
     '0000': '结束通话',
     '0001': '转接人工',
+    '0002': '语义无法识别转接人工',
+    '0003': '有人伤转接人工',
     '1001': '未准备好通话',
     '1002': '通话中',
     '2000': '进入单车拍照',
@@ -49,44 +34,64 @@ STATUS_CODE_MAP = {
     '2016': '确认双车中的车牌'
 }
 
+def extract_state_and_content(data1: str) -> dict | None:
+    """
+    Extracts the state and content from a string in the format <state>STATE</state>content.
 
-async def delete_last_two_chat_records(sessionId):
+    Args:
-    # 获取最后两条聊天记录
+        data1: The input string.
-    headers = {
+
-        "Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
+    Returns:
-        "Content-Type": "application/json"
+        A dictionary with 'state' and 'content' keys if a match is found,
-    }
+        otherwise None.
-    chat_data = {
+    """
-        "appId": APP_ID,
+    data1 = data1.strip()
-        "chatId": sessionId,
+    regex = r"<state>(.*?)</state>(.*)"
-        "offset": 0,
+    match = re.search(regex, data1)
-    }
+    
-    async with httpx.AsyncClient() as client:
+    if match:
-        response = await client.post(
+        return {
-            GET_CHAT_RECORDS_URL, 
+            "state": match.group(1),
-            json=chat_data,
+            "content": match.group(2),
-            headers=headers,
+        }
-            timeout=None
+    return None
+
+async def delete_last_two_chat_records(
+    client: AsyncChatClient,
+    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,
+            chatId=session_id,
+            offset=0,
+            pageSize=10
         )
         response.raise_for_status()
         data = response.json()
-        last_two_dataId = [x['dataId'] for x in data['data']['list'][-2:]]
         
-    # 删除最后聊天聊天记录
+        records = data.get('data', {}).get('list', [])
-    for contentId in last_two_dataId:
+        if len(records) < 2:
-        headers = {
+            logger.warning(f"Less than 2 records found for session {session_id}")
-            "Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
+            return
-            "Content-Type": "application/json"
+            
-        }
+        last_two_data_ids = [record['dataId'] for record in records[-2:]]
-        url = DELETE_CHAT_URL.format(contentId=contentId, chatId=sessionId, appId=APP_ID)
+        logger.info(f"last_two_data_ids: {last_two_data_ids}")
-        async with httpx.AsyncClient() as client:
+        
-            response = await client.delete(
+        # Delete records using SDK
-                url,
+        for content_id in last_two_data_ids:
-                headers=headers,
+            delete_response = await client.delete_chat_record(
-                timeout=None
+                appId=Config.FASTGPT_APP_ID,
+                chatId=session_id,
+                contentId=content_id
             )
-            response.raise_for_status()
+            delete_response.raise_for_status()
-            data = response.json()
+            
+    except Exception as e:
+        logger.error(f"Error deleting chat records: {e}")
+        raise
 
 
 # @app.exception_handler(RequestValidationError)
@@ -94,38 +99,150 @@ async def delete_last_two_chat_records(sessionId):
 #     logging.error(f"Validation Error: {exc.errors()}")
 #     raise HTTPException(status_code=422, detail=exc.errors())
 
-@router.post("/chat", response_model=ProcessResponse_chat)
+def create_sse_event(event: str, data: dict) -> str:
-async def chat(request: ProcessRequest_chat):
+    """Format data as an SSE event."""
-    try:
+    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
-        headers = {
-            "Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
-            "Content-Type": "application/json"
-        }
-        json_data = request.model_dump()
-        chat_data = {
-            "chatId": json_data['sessionId'],
-            "stream": False,
-            "detail": True,
-            "messages": [
-                {
-                    "role": "user",
-                    "content": json_data['text']
-                }
-            ]
-        }
 
-        # Call external analysis service
+@router.post("/chat", response_model=ProcessResponse_chat)
-        async with httpx.AsyncClient() as client:
+async def chat(
-            response = await client.post(
+    request: ProcessRequest_chat,
-                ANALYSIS_SERVICE_URL, 
+    stream: bool = False,
-                json=chat_data,
+    client: AsyncChatClient = Depends(get_fastgpt_client)
-                headers=headers,
+):
-                timeout=None
+    """Handle chat completion request."""
-            )
+    json_data = request.model_dump()
-            response.raise_for_status()
+    logger.info(f"用户请求信息ProcessRequest_chat: {json_data}, stream={stream}")
-            data = response.json()
+    
+    if stream:
+        async def event_generator():
+            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
+                )
+                
+                buffer = ""
+                state_code_found = False
+                
+                async for chunk in response.aiter_lines():
+                    if chunk.startswith('data: '):
+                        data_str = chunk[6:].strip()
+                        if data_str == '[DONE]':
+                            break
+                        try:
+                            data = json.loads(data_str)
+                            try:
+                                delta_content = data['choices'][0]['delta'].get('content', '')
+                            except (KeyError, IndexError):
+                                delta_content = ''
+                            if delta_content:
+                                buffer += delta_content
+                                
+                                if not state_code_found:
+                                    # Check for <state>XXXX</state> pattern
+                                    match = re.search(r"<state>(.*?)</state>", buffer)
+                                    if match:
+                                        state_code = match.group(1)
+                                        
+                                        # Apply logic to map/adjust state code
+                                        nextStageCode = state_code
+                                        if nextStageCode in ['3001', '3002', '1002']:
+                                            nextStageCode = '1002'
+                                        elif nextStageCode == '2006':
+                                            nextStageCode = '2004'
+                                        elif nextStageCode == '2017':
+                                            nextStageCode = '2016'
+                                        elif nextStageCode == '2020':
+                                            nextStageCode = '0002'
+                                        nextStage = STATUS_CODE_MAP.get(nextStageCode, '')
+                                        
+                                        # Send stage code event
+                                        yield create_sse_event("stage_code", {
+                                            "nextStageCode": nextStageCode,
+                                            "nextStage": nextStage
+                                        })
+                                        
+                                        state_code_found = True
+                                        
+                                        # Send remaining content as text_delta
+                                        remaining_content = buffer[match.end():]
+                                        if remaining_content:
+                                            yield create_sse_event("text_delta", {"text": remaining_content})
+                                        buffer = "" # Clear buffer after extracting state
+                                else:
+                                    # State code already found, just stream text
+                                    yield create_sse_event("text_delta", {"text": delta_content})
+                                    buffer = "" # Do not buffer text after state found
+                                    
+                        except json.JSONDecodeError:
+                            continue
+                        except Exception as e:
+                            print(data)
+                            logger.error(f"Error processing chunk: {e}")
+                            continue
+
+                # If stream ends and no state code found (unlikely if format is strict), 
+                # we might want to send what we have
+                if not state_code_found and buffer:
+                     yield create_sse_event("text_delta", {"text": buffer})
+
+                yield create_sse_event("done", {"status": "completed"})
+                
+            except Exception as e:
+                logger.error(f"Streaming error: {e}")
+                yield create_sse_event("error", {"msg": str(e), "code": "500"})
+
+        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
+        )
+        response.raise_for_status()
+        data = response.json()
+        
+    except AuthenticationError as e:
+        logger.error(f"Authentication error: {e}")
+        return ProcessResponse_chat(
+            sessionId=json_data['sessionId'],
+            timeStamp=json_data['timeStamp'],
+            outputText="",
+            nextStage="",
+            nextStageCode="",
+            code="401",
+            msg="认证失败"
+        )
+    except RateLimitError as e:
+        logger.error(f"Rate limit error: {e}")
+        return ProcessResponse_chat(
+            sessionId=json_data['sessionId'],
+            timeStamp=json_data['timeStamp'],
+            outputText="",
+            nextStage="",
+            nextStageCode="",
+            code="429",
+            msg="请求过于频繁，请稍后重试"
+        )
+    except APIError as e:
+        logger.error(f"API error: {e}")
+        return ProcessResponse_chat(
+            sessionId=json_data['sessionId'],
+            timeStamp=json_data['timeStamp'],
+            outputText="",
+            nextStage="",
+            nextStageCode="",
+            code="500",
+            msg="大模型服务器无响应"
+        )
     except Exception as e:
-        print(e)
+        logger.error(f"Unexpected error: {e}")
         return ProcessResponse_chat(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -139,26 +256,59 @@ async def chat(request: ProcessRequest_chat):
     try:
         # Extract content from FastGPT response
         content = data['choices'][0]['message']['content']
+        logger.info(f"FastGPT服务返回信息content: {content}")
+
         finish_reason = data['choices'][0]['finish_reason']
 
-        transfer_to_human = data['newVariables']['state'].get("transfer_to_human", False)
+        # Extract state variables
-        ywrysw = data['newVariables']['state'].get("ywrysw", False)
+        state = data.get('newVariables', {}).get('state', {})
-        ywfjdc = data['newVariables']['state'].get("ywfjdc", False)
+        if isinstance(state, str):
-        ywmtc = data['newVariables']['state'].get("ywmtc", False)
+            state = json.loads(state)
-        jdcsl = data['newVariables']['state'].get("jdcsl", 0)
+
-        accident_info_complete = data['newVariables']['state'].get("accident_info_complete", False)
+        transfer_to_human = state.get("transfer_to_human", False)
-        user_is_ready = data['newVariables']['state'].get("user_is_ready", 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 = data['newVariables']['state'].get("driver_info_complete", False)
+        driver_info_complete = state.get("driver_info_complete", False)
-        drivers_info_complete = data['newVariables']['state'].get("drivers_info_complete", False)        
+        drivers_info_complete = state.get("drivers_info_complete", False)
-        driver_info_check = data['newVariables']['state'].get("drivers_info_check", False)  
+        driver_info_check = state.get("drivers_info_check", False)
-        drivers_info_check = data['newVariables']['state'].get("drivers_info_check", False)  
+        drivers_info_check = state.get("drivers_info_check", False)
 
-        print(data['newVariables'])
+        logger.debug(f"State variables: {data.get('newVariables', {})}")
 
-        nextStageCode=data['newVariables']['status_code']
+        nextStageCode = data['newVariables']['status_code']
-        nextStage=STATUS_CODE_MAP[nextStageCode]
+        
+        # 有一些情况需要调整nextStageCode
+        if nextStageCode in ['3001', '3002', '1002']:
+            nextStageCode = '1002'
+        elif nextStageCode == '2006':
+            nextStageCode = '2004'
+        elif nextStageCode == '2017':
+            nextStageCode = '2016'
+        elif nextStageCode == '2020':
+            nextStageCode = '0002'
+        nextStage = STATUS_CODE_MAP.get(nextStageCode, '')
+
+        # Parse content - sometimes content is a string, sometimes it is a list
+        if isinstance(content, list):
+            logger.debug("content是一个list")
+            content = content[0]['text']['content']
+        elif isinstance(content, str):
+            logger.debug("content是一个str")
+            state_and_content = extract_state_and_content(content)
+            if state_and_content:
+                logger.debug(f"解析后的state和content为: {state_and_content}")
+                content = state_and_content['content']
+            else:
+                raise ValueError("大模型回复中的state解析失败")
+        else:
+            logger.error(f"content既不是list也不是str, type: {type(content)}")
+            raise ValueError("大模型回复不是list也不是str")
 
         return ProcessResponse_chat(
             sessionId=json_data['sessionId'],
@@ -170,7 +320,8 @@ async def chat(request: ProcessRequest_chat):
             msg="",
         )
     except Exception as e:
-        print(e)
+        logger.error(f"解析信息发生错误: {e}")
+        logger.error(f"content: {content}, type: {type(content)}")
         return ProcessResponse_chat(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -183,42 +334,31 @@ async def chat(request: ProcessRequest_chat):
 
 
 @router.post("/set_info", response_model=ProcessResponse_set)
-async def set_info(request: ProcessRequest_set):
+async def set_info(
-    # 获取当前state
+    request: ProcessRequest_set,
-    try:
+    client: AsyncChatClient = Depends(get_fastgpt_client)
-        headers = {
+):
-            "Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
+    """Set information in chat state."""
-            "Content-Type": "application/json"
+    json_data = request.model_dump()
-        }
+    
-        json_data = request.model_dump()
+    try:
-        chat_data = {
+        # Get current state
-            "chatId": json_data['sessionId'],
+        response = await client.create_chat_completion(
-            "stream": False,
+            messages=[{"role": "user", "content": ""}],
-            "detail": True,
+            chatId=json_data['sessionId'],
-            "messages": [
+            stream=False,
-                {
+            detail=True
-                    "role": "user",
+        )
-                    "content": ""
+        response.raise_for_status()
-                }
+        data = response.json()
-            ]
+        
-        }
+        current_state = data['newVariables']['state']
+        if isinstance(current_state, str):
+            current_state = json.loads(current_state)
+        logger.debug(f"Current state: {current_state}")
             
-        # Call external analysis service
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                ANALYSIS_SERVICE_URL, 
-                json=chat_data,
-                headers=headers,
-                timeout=None
-            )
-            response.raise_for_status()
-            data = response.json()
-            current_state = data['newVariables']['state']
-            if isinstance(current_state, str):
-                current_state = json.loads(current_state)
-            print(current_state)
     except Exception as e:
-        print(e)
+        logger.error(f"Error getting current state: {e}")
         return ProcessResponse_set(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -227,9 +367,9 @@ async def set_info(request: ProcessRequest_set):
         )
     
     try:
-        await delete_last_two_chat_records(json_data['sessionId'])
+        await delete_last_two_chat_records(client, json_data['sessionId'])
     except Exception as e:
-        print(e)
+        logger.error(f"Error deleting chat records: {e}")
         return ProcessResponse_set(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -237,45 +377,25 @@ async def set_info(request: ProcessRequest_set):
             msg="大模型后台无响应"
         )
     
-    # 修改当前state
     try:
-        headers = {
+        # Update state
-            "Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
-            "Content-Type": "application/json"
-        }
-        json_data = request.model_dump()
-        # key_map = {"xm1": "xm1", "hpzl1": "hpzl1", "hphm1": "driver1License", "sfzmhm1": "driver1ID", "sfzmwh1": "sfzmwh1", "sjhm1": "driver1Phonenumber","sjwh1": "sjwh1", "xm2":"xm2", "hpzl2": "hpzl2", "hphm2": "driver2License", "sfzmhm2": "driver2ID", "sfzmwh2": "sfzmwh2", "sjhm2": "driver2Phonenumber", "sjwh2": "sjwh2"}  
-        # if json_data['key'] not in key_map:
-            # raise Exception("check your key")
-        # key = key_map.get(json_data['key'])
         key = json_data['key']
         value = json_data['value']
-        current_state[json_data['key']] = value
+        current_state[key] = value
-        print(f'即将上传 {current_state}')
+        logger.info(f'即将上传 {current_state}')
-        # if key == 'driver1ID' or key == 'driver2ID' or key == 'driver1Phonenumber' or key == 'driver2Phonenumber':
-            # current_state[json_data['key'].replace('hm','wh')] = value[-4:]
         
-        chat_data = {
+        # Update state using SDK
-            "chatId": json_data['sessionId'],
+        response = await client.create_chat_completion(
-            "stream": False,
+            messages=[{"role": "user", "content": ""}],
-            "detail": True,
+            chatId=json_data['sessionId'],
-            "variables": {'state':current_state},
+            stream=False,
-            "messages": [
+            detail=True,
-                {
+            variables={'state': current_state}
-                    "role": "user",
+        )
-                    "content": ""
+        response.raise_for_status()
-                }
+        
-            ]
+        # Delete records again after update
-        }            
+        await delete_last_two_chat_records(client, json_data['sessionId'])
-        # Call external analysis service
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                ANALYSIS_SERVICE_URL, 
-                json=chat_data,
-                headers=headers,
-                timeout=None
-            )
-            response.raise_for_status()
         
         return ProcessResponse_set(
             sessionId=json_data['sessionId'],
@@ -285,18 +405,7 @@ async def set_info(request: ProcessRequest_set):
         )
         
     except Exception as e:
-        print(e)
+        logger.error(f"Error setting info: {e}")
-        return ProcessResponse_set(
-            sessionId=json_data['sessionId'],
-            timeStamp=json_data['timeStamp'],
-            code="500",
-            msg="大模型后台无响应"
-        )
-
-    try:
-        await delete_last_two_chat_records(json_data['sessionId'])
-    except Exception as e:
-        print(e)
         return ProcessResponse_set(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -305,42 +414,31 @@ async def set_info(request: ProcessRequest_set):
         )
     
 @router.post("/get_info", response_model=ProcessResponse_get)
-async def get_info(request: ProcessRequest_get):
+async def get_info(
-    # 获取当前state
+    request: ProcessRequest_get,
-    try:
+    client: AsyncChatClient = Depends(get_fastgpt_client)
-        headers = {
+):
-            "Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
+    """Get information from chat state."""
-            "Content-Type": "application/json"
+    json_data = request.model_dump()
-        }
+    
-        json_data = request.model_dump()
+    try:
-        chat_data = {
+        # Get current state
-            "chatId": json_data['sessionId'],
+        response = await client.create_chat_completion(
-            "stream": False,
+            messages=[{"role": "user", "content": ""}],
-            "detail": True,
+            chatId=json_data['sessionId'],
-            "messages": [
+            stream=False,
-                {
+            detail=True
-                    "role": "user",
+        )
-                    "content": ""
+        response.raise_for_status()
-                }
+        data = response.json()
-            ]
+        
-        }
+        current_state = data['newVariables']['state']
+        if isinstance(current_state, str):
+            current_state = json.loads(current_state)
+        logger.debug(f"Current state: {current_state}")
             
-        # Call external analysis service
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                ANALYSIS_SERVICE_URL, 
-                json=chat_data,
-                headers=headers,
-                timeout=None
-            )
-            response.raise_for_status()
-            data = response.json()
-            current_state = data['newVariables']['state']
-            if isinstance(current_state, str):
-                current_state = json.loads(current_state)
-            print(current_state)
     except Exception as e:
-        print(e)
+        logger.error(f"Error getting state: {e}")
         return ProcessResponse_get(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -350,9 +448,9 @@ async def get_info(request: ProcessRequest_get):
         )
     
     try:
-        await delete_last_two_chat_records(json_data['sessionId'])
+        await delete_last_two_chat_records(client, json_data['sessionId'])
     except Exception as e:
-        print(e)
+        logger.error(f"Error deleting records: {e}")
         return ProcessResponse_get(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -363,66 +461,36 @@ async def get_info(request: ProcessRequest_get):
     
     try:
         key = json_data['key']
-        # key_map = {"hpzl1": "hpzl1", "hphm1": "driver1License", "sfzmhm1": "driver1ID", "sfzmwh1": "sfzmwh1", "sjhm1": "driver1Phonenumber","sjwh1": "sjwh1", "hpzl2": "hpzl2", "hphm2": "driver2License", "sfzmhm2": "driver2ID", "sfzmwh2": "sfzmwh2", "sjhm2": "driver2Phonenumber", "sjwh2": "sjwh2"} 
         acd_keys = ['ywrysw', 'ywfjdc', 'ywmtc', 'bjrjs', 'sgfssj', 'sfsgxc', 'jdcsl', 'sgyy']
         human1_keys = ['xm1', 'hpzl1', 'hphm1', 'sfzmhm1', 'sfzmwh1', 'sjhm1', 'sjwh1', 'csbw1']
         human2_keys = ['xm2', 'hpzl2', 'hphm2', 'sfzmhm2', 'sfzmwh2', 'sjhm2', 'sjwh2', 'csbw2']
+        
+        def bool_to_str(v):
+            """Convert boolean to string representation."""
+            return '1' if v is True else '0' if v is False else v
+        
         if key == 'all':
-            acd_values = {}
+            acd_values = {k: bool_to_str(current_state.get(k, '')) for k in acd_keys}
-            for k in acd_keys:
+            human1_values = {k: bool_to_str(current_state.get(k, '')) for k in human1_keys}
-                v = current_state.get(k, '')
+            human2_values = {k: bool_to_str(current_state.get(k, '')) for k in human2_keys}
-                if isinstance(v, bool):
+            value = json.dumps({
-                    v = '1' if v==True else '0'
+                'acdinfo': acd_values,
-                acd_values[k] = v
+                'acdhuman1': human1_values,
-            human1_values = {}
+                'acdhuman2': human2_values
-            for k in human1_keys:
+            })
-                v = current_state.get(k, '')
-                if isinstance(v, bool):
-                    v = '1' if v==True else '0'
-                human1_values[k] = v
-            human2_values = {}
-            for k in human2_keys:
-                v = current_state.get(k, '')
-                if isinstance(v, bool):
-                    v = '1' if v==True else '0'
-                human2_values[k] = v
-            value = {'acdinfo': acd_values, 'acdhuman1': human1_values, 'acdhuman2': human2_values}
-            value = json.dumps(value)
         elif key == "acdinfo":
-            acd_values = {}
+            acd_values = {k: bool_to_str(current_state.get(k, '')) for k in acd_keys}
-            for k in acd_keys:
-                v = current_state.get(k, '')
-                if isinstance(v, bool):
-                    v = '1' if v==True else '0'
-                acd_values[k] = v
             value = json.dumps(acd_values)
         elif key == 'acdhuman1':
-            human1_values = {}
+            human1_values = {k: bool_to_str(current_state.get(k, '')) for k in human1_keys}
-            for k in human1_keys:
+            value = json.dumps(human1_values)
-                v = current_state.get(k, '')
-                if isinstance(v, bool):
-                    v = '1' if v==True else '0'
-                human1_values[k] = v
-            value = json.dumps(human1_value)
         elif key == 'acdhuman2':
-            human2_values = {}
+            human2_values = {k: bool_to_str(current_state.get(k, '')) for k in human2_keys}
-            for k in human1_keys:
+            value = json.dumps(human2_values)
-                v = current_state.get(k, '')
-                if isinstance(v, bool):
-                    v = '1' if v==True else '0'
-                human2_values[k] = v
-            value = json.dumps(human2_value)
         else:
-            # if key in key_map:
+            value = json.dumps(current_state.get(key, ''))
-            #     real_key = key_map.get(key)
-            # else:
-            #     real_key = key
-            value = current_state.get(key, '')
-            value = json.dumps(value)
-            # if value == 'null':
-                # raise Exception("check your key")
         
-        print(json_data)
+        logger.debug(f"Returning value for key '{key}': {value}")
         return ProcessResponse_get(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -430,8 +498,9 @@ async def get_info(request: ProcessRequest_get):
             code="200",
             msg=""
         )
+        
     except Exception as e:
-        print(e)
+        logger.error(f"Error getting info: {e}")
         return ProcessResponse_get(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
@@ -442,39 +511,32 @@ async def get_info(request: ProcessRequest_get):
     
 
 @router.delete("/delete_session", response_model=ProcessResponse_delete_session)
-async def delete_session(request: ProcessRequest_delete_session):
+async def delete_session(
-    try:
+    request: ProcessRequest_delete_session,
-        headers = {
+    client: AsyncChatClient = Depends(get_fastgpt_client)
-            "Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
+):
-            "Content-Type": "application/json"
+    """Delete a chat session."""
-        }
+    json_data = request.model_dump()
-        json_data = request.model_dump()
+    chat_id = json_data.get('sessionId')
-        chatId = json_data.get('sessionId', None)
-        if chatId is None:
-            raise ValueError(f'Error in chatId.')
-        url = DELETE_SESSION_URL.format(appId=APP_ID, chatId=chatId)
     
-        # Call external analysis service
+    if not chat_id:
-        async with httpx.AsyncClient() as client:
+        return ProcessResponse_delete_session(
-            response = await client.delete(
+            sessionId="",
-                url, 
-                headers=headers,
-                timeout=None
-            )
-            response.raise_for_status()
-            data = response.json()
-            # print(data)
-    except Exception as e:
-        print(e)
-        return ProcessResponse_chat(
-            sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],
-            code="500",
+            code="400",
-            msg="大模型服务器无响应"
+            msg="sessionId is required"
         )
     
     try:
-        if data['code'] == 200:
+        # Use SDK's delete_chat_history
+        response = await client.delete_chat_history(
+            appId=Config.FASTGPT_APP_ID,
+            chatId=chat_id
+        )
+        response.raise_for_status()
+        data = response.json()
+        
+        if data.get('code') == 200:
             return ProcessResponse_delete_session(
                 sessionId=json_data['sessionId'],
                 timeStamp=json_data['timeStamp'],
@@ -483,8 +545,9 @@ async def delete_session(request: ProcessRequest_delete_session):
             )
         else:
             raise ValueError("删除会话失败")
+            
     except Exception as e:
-        print(e)
+        logger.error(f"Error deleting session: {e}")
         return ProcessResponse_delete_session(
             sessionId=json_data['sessionId'],
             timeStamp=json_data['timeStamp'],

src/core/config.py

@@ -9,3 +9,22 @@ class Config:
     SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key")
     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")
+    
+    @classmethod
+    def validate(cls):
+        """Validate required configuration."""
+        required = [
+            ("FASTGPT_API_KEY", cls.FASTGPT_API_KEY),
+            ("FASTGPT_APP_ID", cls.FASTGPT_APP_ID),
+        ]
+        missing = [name for name, value in required if not value]
+        if missing:
+            raise ValueError(f"Missing required environment variables: {', '.join(missing)}")

src/core/fastgpt_client.py

@@ -0,0 +1,39 @@
+"""FastGPT client dependency injection."""
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastgpt_client import AsyncChatClient
+from .config import Config
+
+# Global client instance
+_fastgpt_client: AsyncChatClient | None = None
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Manage FastGPT client lifecycle."""
+    global _fastgpt_client
+    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)
+
+
+def get_fastgpt_client() -> AsyncChatClient:
+    """Get the FastGPT client instance."""
+    if _fastgpt_client is None:
+        raise RuntimeError("FastGPT client not initialized")
+    return _fastgpt_client

src/core/logging_config.py

@@ -0,0 +1,79 @@
+"""Loguru logging configuration."""
+import sys
+import logging
+import inspect
+from loguru import logger
+from pathlib import Path
+from .config import Config
+
+# Remove default logger
+logger.remove()
+
+# Add console handler with color
+logger.add(
+    sys.stderr,
+    format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
+    level="DEBUG" if Config.DEBUG else "INFO",
+    colorize=True,
+)
+
+# Add file handler with rotation
+log_dir = Path("logs")
+log_dir.mkdir(exist_ok=True)
+
+logger.add(
+    log_dir / "server_{time:YYYY-MM-DD}.log",
+    rotation="00:00",  # Rotate at midnight
+    retention="30 days",  # Keep logs for 30 days
+    compression="zip",  # Compress old logs
+    format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}",
+    level="DEBUG",
+    encoding="utf-8",
+)
+
+# Add error file handler
+logger.add(
+    log_dir / "error_{time:YYYY-MM-DD}.log",
+    rotation="00:00",
+    retention="90 days",
+    compression="zip",
+    format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message}",
+    level="ERROR",
+    encoding="utf-8",
+)
+
+# Configure uvicorn logging to use loguru
+class InterceptHandler(logging.Handler):
+    """Intercept standard logging messages toward loguru."""
+    
+    def emit(self, record: logging.LogRecord) -> None:
+        # Get corresponding Loguru level if it exists
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = str(record.levelno)
+
+        # Find caller from where originated the logged message
+        frame, depth = inspect.currentframe(), 2
+        while frame and frame.f_code.co_filename == logging.__file__:
+            frame = frame.f_back
+            depth += 1
+
+        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+
+
+def setup_logging():
+    """Configure logging for the application."""
+    # Intercept uvicorn and fastapi loggers
+    logging.getLogger("uvicorn").handlers = [InterceptHandler()]
+    logging.getLogger("uvicorn.access").handlers = [InterceptHandler()]
+    logging.getLogger("fastapi").handlers = [InterceptHandler()]
+    
+    # Set log levels
+    logging.getLogger("uvicorn").setLevel(logging.INFO)
+    logging.getLogger("uvicorn.access").setLevel(logging.INFO)
+    logging.getLogger("fastapi").setLevel(logging.INFO)
+    
+    # Intercept httpx logs if needed
+    logging.getLogger("httpx").handlers = [InterceptHandler()]
+    logging.getLogger("httpx").setLevel(logging.WARNING)

src/main.py

@@ -1,11 +1,17 @@
 from fastapi import FastAPI
 import sys
 from .api.endpoints import router as api_router
+from .core.fastgpt_client import lifespan
+from .core.logging_config import setup_logging
+
+# Setup logging first
+setup_logging()
 
 app = FastAPI(
     title="AI Accident Information Collection API",
     description="AI Accident Information Collection API",
-    version="1.0.0"
+    version="1.0.0",
+    lifespan=lifespan
 )
 
 @app.get("/")

test/api/fastapi.http

@@ -1,200 +1,74 @@
-GET http://127.0.0.1:8000
+@baseUrl = http://127.0.0.1:8080
+###
+GET http://127.0.0.1:8080
 
 HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:07:21 GMT
+connection: close
-server: uvicorn
 content-length: 32
 content-type: application/json
-connection: close
+date: Thu, 08 Jan 2026 08:58:09 GMT
-###
-POST http://127.0.0.1:8000/chat
-content-type: application/json
-
-{
-    "sessionId": "a0009",
-    "timeStamp": "202503310303",
-    "text": "可以，继续"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:18:14 GMT
 server: uvicorn
-content-length: 169
-content-type: application/json
-connection: close
 ###
-POST http://127.0.0.1:8000/chat
+POST http://127.0.0.1:8080/chat
 content-type: application/json
 
 {
-    "sessionId": "a0009",
+    "sessionId": "a1002",
-    "timeStamp": "202503310303",
-    "text": "没有"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:18:24 GMT
-server: uvicorn
-content-length: 187
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/chat
-content-type: application/json
-
-{
-    "sessionId": "a0009",
-    "timeStamp": "202503310303",
-    "text": "没有"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:18:29 GMT
-server: uvicorn
-content-length: 178
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/chat
-content-type: application/json
-
-{
-    "sessionId": "a0009",
-    "timeStamp": "202503310303",
-    "text": "是的"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:18:38 GMT
-server: uvicorn
-content-length: 196
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/chat
-content-type: application/json
-
-{
-    "sessionId": "a0009",
-    "timeStamp": "202503310303",
-    "text": "十点半"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:19:07 GMT
-server: uvicorn
-content-length: 247
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/chat
-content-type: application/json
-
-{
-    "sessionId": "a0009",
-    "timeStamp": "202503310303",
-    "text": "我在现场"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:19:27 GMT
-server: uvicorn
-content-length: 208
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/chat
-content-type: application/json
-
-{
-    "sessionId": "a0009",
-    "timeStamp": "202503310303",
-    "text": "我闯红灯了"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:19:40 GMT
-server: uvicorn
-content-length: 231
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/chat
-content-type: application/json
-
-{
-    "sessionId": "a0009",
     "timeStamp": "202503310303",
     "text": "【拍摄完成】"
 }
 
 HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 03:20:10 GMT
-server: uvicorn
-content-length: 210
-content-type: application/json
 connection: close
+content-length: 205
+content-type: application/json
+date: Thu, 08 Jan 2026 08:59:37 GMT
+server: uvicorn
 ###
-POST http://127.0.0.1:8000/get_info
+POST http://127.0.0.1:8080/get_info
 content-type: application/json
 
 {
-    "sessionId": "a0009",
+    "sessionId": "a1002",
-    "timeStamp": "202503310303",
-    "key": "acdinfo"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 05:19:05 GMT
-server: uvicorn
-content-length: 271
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/set_info
-content-type: application/json
-
-{
-    "sessionId": "a0009",
-    "timeStamp": "202503310303",
-    "key": "hphm1",
-    "value": "沪A8938"
-}
-
-HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 05:30:03 GMT
-server: uvicorn
-content-length: 70
-content-type: application/json
-connection: close
-###
-POST http://127.0.0.1:8000/get_info
-content-type: application/json
-
-{
-    "sessionId": "a0009",
     "timeStamp": "202503310303",
     "key": "hphm1"
 }
 
 HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 05:30:27 GMT
+connection: close
-server: uvicorn
 content-length: 97
 content-type: application/json
-connection: close
+date: Thu, 08 Jan 2026 09:27:05 GMT
+server: uvicorn
 ###
-DELETE http://127.0.0.1:8000/delete_session
+POST http://127.0.0.1:8080/set_info
 content-type: application/json
 
 {
-    "sessionId": "a0009",
+    "sessionId": "a1002",
+    "timeStamp": "202503310303",
+    "key": "hphm1",
+    "value": "沪A8939"
+}
+
+HTTP/1.1 200  - OK
+connection: close
+content-length: 70
+content-type: application/json
+date: Thu, 08 Jan 2026 09:27:00 GMT
+server: uvicorn
+###
+DELETE http://127.0.0.1:8080/delete_session
+content-type: application/json
+
+{
+    "sessionId": "a1002",
     "timeStamp": "202503310303"
 }
 
 HTTP/1.1 200  - OK
-date: Fri, 20 Jun 2025 05:56:25 GMT
+connection: close
-server: uvicorn
 content-length: 70
 content-type: application/json
-connection: close
+date: Thu, 08 Jan 2026 09:27:24 GMT
+server: uvicorn