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

use fastgpt python sdk

Browse Source
This commit is contained in:
Xin Wang
2026-01-08 16:22:25 +08:00
parent c6869f773e
commit d5f81ef79f
8 changed files with 472 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

628
src/api/endpoints.py
View File

@@ -1,37 +1,21 @@
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Depends
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 datetime import datetime
import os
from dotenv import load_dotenv
from urllib import parse as parse
from fastgpt_client import AsyncChatClient
from fastgpt_client.exceptions import (
APIError, AuthenticationError, RateLimitError, ValidationError
)
from ..core.fastgpt_client import get_fastgpt_client
from ..core.config import Config
from loguru import logger
import json
load_dotenv()
import re
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 +33,63 @@ 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):
# 获取最后两条聊天记录
headers = {
"Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
"Content-Type": "application/json"
}
chat_data = {
"appId": APP_ID,
"chatId": sessionId,
"offset": 0,
}
async with httpx.AsyncClient() as client:
response = await client.post(
GET_CHAT_RECORDS_URL,
json=chat_data,
headers=headers,
timeout=None
Args:
data1: The input string.
Returns:
A dictionary with 'state' and 'content' keys if a match is found,
otherwise None.
"""
data1 = data1.strip()
regex = r"<state>(.*?)</state>(.*)"
match = re.search(regex, data1)
if match:
return {
"state": match.group(1),
"content": match.group(2),
}
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:]]
# 删除最后聊天聊天记录
for contentId in last_two_dataId:
headers = {
"Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
"Content-Type": "application/json"
}
url = DELETE_CHAT_URL.format(contentId=contentId, chatId=sessionId, appId=APP_ID)
async with httpx.AsyncClient() as client:
response = await client.delete(
url,
headers=headers,
timeout=None
records = data.get('data', {}).get('list', [])
if len(records) < 2:
logger.warning(f"Less than 2 records found for session {session_id}")
return
last_two_data_ids = [record['dataId'] for record in records[-2:]]
# Delete records using SDK
for content_id in last_two_data_ids:
delete_response = await client.delete_chat_record(
appId=Config.FASTGPT_APP_ID,
chatId=session_id,
contentId=content_id
)
response.raise_for_status()
data = response.json()
delete_response.raise_for_status()
except Exception as e:
logger.error(f"Error deleting chat records: {e}")
raise
# @app.exception_handler(RequestValidationError)
@@ -95,37 +98,60 @@ async def delete_last_two_chat_records(sessionId):
# raise HTTPException(status_code=422, detail=exc.errors())
@router.post("/chat", response_model=ProcessResponse_chat)
async def chat(request: ProcessRequest_chat):
async def chat(
request: ProcessRequest_chat,
client: AsyncChatClient = Depends(get_fastgpt_client)
):
"""Handle chat completion request."""
json_data = request.model_dump()
logger.info(f"用户请求信息ProcessRequest_chat: {json_data}")
try:
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']
}
]
}
# 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()
# 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()
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 +165,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)
ywrysw = data['newVariables']['state'].get("ywrysw", False)
ywfjdc = data['newVariables']['state'].get("ywfjdc", False)
ywmtc = data['newVariables']['state'].get("ywmtc", False)
jdcsl = data['newVariables']['state'].get("jdcsl", 0)
accident_info_complete = data['newVariables']['state'].get("accident_info_complete", False)
user_is_ready = data['newVariables']['state'].get("user_is_ready", False)
# 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 = data['newVariables']['state'].get("driver_info_complete", False)
drivers_info_complete = data['newVariables']['state'].get("drivers_info_complete", False)
driver_info_check = data['newVariables']['state'].get("drivers_info_check", False)
drivers_info_check = data['newVariables']['state'].get("drivers_info_check", False)
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)
print(data['newVariables'])
logger.debug(f"State variables: {data.get('newVariables', {})}")
nextStageCode=data['newVariables']['status_code']
nextStage=STATUS_CODE_MAP[nextStageCode]
nextStageCode = data['newVariables']['status_code']
# 有一些情况需要调整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 +229,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'],
@@ -182,121 +242,43 @@ async def chat(request: ProcessRequest_chat):
)
@router.post("/set_info", response_model=ProcessResponse_set)
async def set_info(request: ProcessRequest_set):
# 获取当前state
@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()
try:
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": ""
}
]
}
# Get current state
response = await client.create_chat_completion(
messages=[{"role": "user", "content": ""}],
chatId=json_data['sessionId'],
stream=False,
detail=True
)
response.raise_for_status()
data = response.json()
# 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)
current_state = data['newVariables']['state']
if isinstance(current_state, str):
current_state = json.loads(current_state)
logger.debug(f"Current state: {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'],
code="500",
msg="大模型服务器无响应"
)
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)
return ProcessResponse_set(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
code="500",
msg="大模型后台无响应"
)
# 修改当前state
try:
headers = {
"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
print(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 = {
"chatId": json_data['sessionId'],
"stream": False,
"detail": True,
"variables": {'state':current_state},
"messages": [
{
"role": "user",
"content": ""
}
]
}
# 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'],
timeStamp=json_data['timeStamp'],
code="200",
msg=""
)
except Exception as e:
print(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)
logger.error(f"Error deleting chat records: {e}")
return ProcessResponse_set(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
@@ -304,43 +286,68 @@ async def set_info(request: ProcessRequest_set):
msg="大模型后台无响应"
)
@router.post("/get_info", response_model=ProcessResponse_get)
async def get_info(request: ProcessRequest_get):
# 获取当前state
try:
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": ""
}
]
}
# Update state
key = json_data['key']
value = json_data['value']
current_state[key] = value
logger.info(f'即将上传 {current_state}')
# Update state using SDK
response = await client.create_chat_completion(
messages=[{"role": "user", "content": ""}],
chatId=json_data['sessionId'],
stream=False,
detail=True,
variables={'state': current_state}
)
response.raise_for_status()
# Delete records again after update
await delete_last_two_chat_records(client, json_data['sessionId'])
return ProcessResponse_set(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
code="200",
msg=""
)
# 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 setting info: {e}")
return ProcessResponse_set(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
code="500",
msg="大模型后台无响应"
)
@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()
try:
# Get current state
response = await client.create_chat_completion(
messages=[{"role": "user", "content": ""}],
chatId=json_data['sessionId'],
stream=False,
detail=True
)
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}")
except Exception as e:
logger.error(f"Error getting state: {e}")
return ProcessResponse_get(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
@@ -348,11 +355,11 @@ async def get_info(request: ProcessRequest_get):
code="500",
msg="大模型服务器无响应"
)
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'],
@@ -360,69 +367,39 @@ async def get_info(request: ProcessRequest_get):
code="500",
msg="大模型后台无响应"
)
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 = {}
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
human1_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)
acd_values = {k: bool_to_str(current_state.get(k, '')) for k in acd_keys}
human1_values = {k: bool_to_str(current_state.get(k, '')) for k in human1_keys}
human2_values = {k: bool_to_str(current_state.get(k, '')) for k in human2_keys}
value = json.dumps({
'acdinfo': acd_values,
'acdhuman1': human1_values,
'acdhuman2': human2_values
})
elif key == "acdinfo":
acd_values = {}
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
acd_values = {k: bool_to_str(current_state.get(k, '')) for k in acd_keys}
value = json.dumps(acd_values)
elif key == 'acdhuman1':
human1_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
value = json.dumps(human1_value)
human1_values = {k: bool_to_str(current_state.get(k, '')) for k in human1_keys}
value = json.dumps(human1_values)
elif key == 'acdhuman2':
human2_values = {}
for k in human1_keys:
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)
human2_values = {k: bool_to_str(current_state.get(k, '')) for k in human2_keys}
value = json.dumps(human2_values)
else:
# if key in key_map:
# 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)
value = json.dumps(current_state.get(key, ''))
logger.debug(f"Returning value for key '{key}': {value}")
return ProcessResponse_get(
sessionId=json_data['sessionId'],
timeStamp=json_data['timeStamp'],
@@ -430,8 +407,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 +420,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):
try:
headers = {
"Authorization": f"Bearer {ANALYSIS_AUTH_TOKEN}",
"Content-Type": "application/json"
}
json_data = request.model_dump()
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
async with httpx.AsyncClient() as client:
response = await client.delete(
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'],
async def delete_session(
request: ProcessRequest_delete_session,
client: AsyncChatClient = Depends(get_fastgpt_client)
):
"""Delete a chat session."""
json_data = request.model_dump()
chat_id = json_data.get('sessionId')
if not chat_id:
return ProcessResponse_delete_session(
sessionId="",
timeStamp=json_data['timeStamp'],
code="500",
msg="大模型服务器无响应"
code="400",
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 +454,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'],