增加日志
This commit is contained in:
parent
b9fc511806
commit
52951871c0
@ -123,32 +123,46 @@ def stream_ragflow_response(result: Generator, chat_id: str, start_time: float)
|
||||
"""
|
||||
流式处理RAGFlow响应 - 同步版本，修复首token延迟
|
||||
"""
|
||||
import time
|
||||
|
||||
server_stream_start = time.time()
|
||||
logger.info(f"[RAG_SERVER {server_stream_start:.3f}] 🚀 开始流式响应处理，chat_id: {chat_id}")
|
||||
|
||||
last_answer = ""
|
||||
first_token_received = False
|
||||
start_stream_time = time.perf_counter()
|
||||
|
||||
# 立即发送连接建立消息，解决首token延迟
|
||||
yield "event: ping\ndata: {\"status\": \"connected\"}\n\n"
|
||||
connection_time = time.time()
|
||||
ping_message = "event: ping\ndata: {\"status\": \"connected\"}\n\n"
|
||||
yield ping_message
|
||||
logger.info(f"[RAG_SERVER {connection_time:.3f}] 📡 连接建立ping发送，耗时: {connection_time - server_stream_start:.3f}s")
|
||||
|
||||
chunk_count = 0
|
||||
try:
|
||||
for chunk in result:
|
||||
chunk_time = time.time()
|
||||
chunk_count += 1
|
||||
|
||||
# 检查第一个token的延迟
|
||||
if not first_token_received:
|
||||
first_token_received = True
|
||||
latency = time.perf_counter() - start_stream_time
|
||||
logger.info(f"[RAG_PROFILE] Time to First Token ({chat_id}): {latency:.3f}s")
|
||||
logger.info(f"[RAG_SERVER {chunk_time:.3f}] 🎯 首token到达，耗时: {latency:.3f}s")
|
||||
|
||||
# 处理chunk数据
|
||||
payload = chunk.get('data') if isinstance(chunk, dict) else chunk
|
||||
if not payload:
|
||||
logger.info(f"[RAG_SERVER {chunk_time:.3f}] ⚠️ 空chunk跳过")
|
||||
continue
|
||||
|
||||
body = payload if isinstance(payload, dict) else {'data': payload}
|
||||
|
||||
# 处理错误
|
||||
if isinstance(chunk, dict) and chunk.get('code') and chunk.get('code') != 0:
|
||||
logger.error(f"RAGFlow Stream Error: {chunk}")
|
||||
yield format_sse({'message': chunk.get('message', '流式处理异常')}, event='error')
|
||||
logger.error(f"[RAG_SERVER {chunk_time:.3f}] ❌ RAGFlow Stream Error: {chunk}")
|
||||
error_message = format_sse({'message': chunk.get('message', '流式处理异常')}, event='error')
|
||||
yield error_message
|
||||
continue
|
||||
|
||||
# 处理answer字段的增量更新
|
||||
@ -161,23 +175,32 @@ def stream_ragflow_response(result: Generator, chat_id: str, start_time: float)
|
||||
if delta:
|
||||
body['answer'] = delta
|
||||
last_answer = current_answer
|
||||
logger.info(f"[RAG_SERVER {chunk_time:.3f}] 📝 Chunk #{chunk_count} 处理完成，delta长度: {len(delta)}")
|
||||
yield format_sse(body)
|
||||
else:
|
||||
# 上下文重置的备用处理
|
||||
last_answer = current_answer
|
||||
logger.info(f"[RAG_SERVER {chunk_time:.3f}] 🔄 Chunk #{chunk_count} 上下文重置")
|
||||
yield format_sse(body)
|
||||
else:
|
||||
logger.info(f"[RAG_SERVER {chunk_time:.3f}] 📦 Chunk #{chunk_count} 其他数据: {body}")
|
||||
yield format_sse(body)
|
||||
|
||||
# 流结束
|
||||
yield format_sse({'status': 'completed'}, event='end')
|
||||
stream_end_time = time.time()
|
||||
end_message = format_sse({'status': 'completed'}, event='end')
|
||||
yield end_message
|
||||
logger.info(f"[RAG_SERVER {stream_end_time:.3f}] 🏁 流式响应完成，chat_id: {chat_id}")
|
||||
logger.info(f"[RAG_SERVER {stream_end_time:.3f}] 📊 总共处理chunk数量: {chunk_count}")
|
||||
|
||||
except Exception as exc:
|
||||
logger.exception(f'[RAG_PROFILE] ragflow流式对话异常: {exc}')
|
||||
yield format_sse({'message': str(exc)}, event='error')
|
||||
error_time = time.time()
|
||||
logger.exception(f'[RAG_SERVER {error_time:.3f}] ragflow流式对话异常: {exc}')
|
||||
error_message = format_sse({'message': str(exc)}, event='error')
|
||||
yield error_message
|
||||
finally:
|
||||
total_time = time.perf_counter() - start_time
|
||||
logger.info(f'[RAG_PROFILE] Total Stream Duration ({chat_id}): {total_time:.3f}s')
|
||||
logger.info(f'[RAG_SERVER {time.time():.3f}] ⏱️ Total Stream Duration ({chat_id}): {total_time:.3f}s')
|
||||
|
||||
|
||||
# 聊天助手列表
|
||||
|
||||
@ -11,51 +11,78 @@ import requests # 同步 HTTP 客户端
|
||||
|
||||
# 下面的默认参数可以通过环境变量覆盖，避免把机密写死在代码里。
|
||||
API_URL = os.environ.get(
|
||||
"RAGFLOW_URL",
|
||||
"http://10.0.0.202:9099/system/ragflow/converse_with_chat_assistant",
|
||||
"LOCAL_API_URL",
|
||||
"http://localhost:9099/system/ragflow/converse_with_chat_assistant",
|
||||
)
|
||||
AUTH_TOKEN = os.environ.get(
|
||||
"RAGFLOW_TOKEN",
|
||||
"LOCAL_API_TOKEN",
|
||||
"Bearer ",
|
||||
)
|
||||
DEFAULT_CHAT_ID = os.environ.get(
|
||||
"RAGFLOW_CHAT_ID", "db4bb966895b11f08cda0242ac130006"
|
||||
"LOCAL_CHAT_ID", "db4bb966895b11f08cda0242ac130006"
|
||||
)
|
||||
DEFAULT_SESSION_ID = os.environ.get(
|
||||
"RAGFLOW_SESSION_ID", "38d765e48a3811f0be310242ac130006"
|
||||
"LOCAL_SESSION_ID", "38d765e48a3811f0be310242ac130006"
|
||||
)
|
||||
DEFAULT_QUESTION = os.environ.get(
|
||||
"RAGFLOW_QUESTION", "你好，请用简洁的语言介绍你自己"
|
||||
"LOCAL_QUESTION", "你好，请用简洁的语言介绍你自己"
|
||||
)
|
||||
LOGIN_URL = os.environ.get(
|
||||
"RAGFLOW_LOGIN_URL",
|
||||
"http://10.0.0.202:9099/login",
|
||||
"LOCAL_LOGIN_URL",
|
||||
"http://localhost:9099/login",
|
||||
)
|
||||
|
||||
|
||||
def print_log(message: str):
|
||||
"""打印带时间戳的日志"""
|
||||
current_time = time.time()
|
||||
print(f"[CLIENT {current_time:.3f}] {message}")
|
||||
|
||||
|
||||
async def login_get_token() -> str:
|
||||
"""
|
||||
获取本地服务token
|
||||
"""
|
||||
print_log("开始登录获取token...")
|
||||
login_start = time.time()
|
||||
|
||||
payload = {
|
||||
"username": "admin",
|
||||
"password": "admin123"
|
||||
}
|
||||
headers = {
|
||||
"Authorization": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiMSIsInVzZXJfbmFtZSI6ImFkbWluIiwiZGVwdF9uYW1lIjoiXHU3ODE0XHU1M2QxXHU5MGU4XHU5NWU4Iiwic2Vzc2lvbl9pZCI6IjYwNjdlMzkxLTRiNGMtNGUwYy1hZjM5LTVmNzczZjIzMmE3OSIsImxvZ2luX2luZm8iOnsiaXBhZGRyIjpudWxsLCJsb2dpbkxvY2F0aW9uIjoiXHU2NzJhXHU3N2U1IiwiYnJvd3NlciI6Ik90aGVyIiwib3MiOiJPdGhlciIsImxvZ2luVGltZSI6IjIwMjUtMTItMDIgMTA6NTk6NDAifSwiZXhwIjoxNzY3MjM2MzgwfQ.NM9j0emNz8trxU1DX87PVZZaWWHlFFwF7o0Pxop6B5c"
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
async with aiohttp.ClientSession() as session:
|
||||
# 使用data参数发送form-data而不是json
|
||||
async with session.post(
|
||||
LOGIN_URL, data=payload, headers=headers, timeout=60
|
||||
LOGIN_URL, json=payload, headers=headers, timeout=60
|
||||
) as resp:
|
||||
response_data = await resp.json() # 将响应转换为JSON
|
||||
token = response_data.get("token") # 获取token字段
|
||||
return token
|
||||
if resp.status == 200:
|
||||
response_data = await resp.json()
|
||||
token = response_data.get("token") or response_data.get("access_token")
|
||||
login_end = time.time()
|
||||
print_log(f"登录成功，耗时: {login_end - login_start:.3f}s")
|
||||
if token:
|
||||
return token
|
||||
else:
|
||||
print_log("登录响应无token，返回空字符串")
|
||||
return ""
|
||||
else:
|
||||
print_log(f"登录失败，状态码: {resp.status}")
|
||||
return ""
|
||||
|
||||
|
||||
async def stream_chat_async(question: str, chat_id: str, session_id: str) -> str:
|
||||
"""
|
||||
异步版（aiohttp），适合已有事件循环的应用，不阻塞主线程。
|
||||
"""
|
||||
client_start = time.time()
|
||||
print_log(f"开始SSE流式请求 - 问题: {question}")
|
||||
print_log(f"chat_id: {chat_id}, session_id: {session_id}")
|
||||
|
||||
token = await login_get_token()
|
||||
token_received = time.time()
|
||||
print_log(f"Token获取完成，耗时: {token_received - client_start:.3f}s")
|
||||
|
||||
payload = {
|
||||
"chatId": chat_id,
|
||||
@ -68,15 +95,24 @@ async def stream_chat_async(question: str, chat_id: str, session_id: str) -> str
|
||||
"Accept": "text/event-stream",
|
||||
}
|
||||
|
||||
print_log("发送HTTP请求到服务器...")
|
||||
request_sent = time.time()
|
||||
print_log(f"请求耗时: {request_sent - token_received:.3f}s")
|
||||
|
||||
answer_parts = []
|
||||
event_type = "message"
|
||||
first_chunk_received = False
|
||||
chunk_count = 0
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(
|
||||
API_URL, json=payload, headers=headers, timeout=60
|
||||
) as resp:
|
||||
print(
|
||||
f"# status={resp.status} content-type={resp.headers.get('content-type')}")
|
||||
response_received = time.time()
|
||||
print_log(f"HTTP响应接收完成，状态码: {resp.status}")
|
||||
print_log(f"连接建立耗时: {response_received - request_sent:.3f}s")
|
||||
print_log(f"Content-Type: {resp.headers.get('content-type')}")
|
||||
|
||||
resp.raise_for_status()
|
||||
|
||||
while True:
|
||||
@ -90,35 +126,44 @@ async def stream_chat_async(question: str, chat_id: str, session_id: str) -> str
|
||||
event_type = line.split(":", 1)[1].strip() or "message"
|
||||
continue
|
||||
if not line.startswith("data:"):
|
||||
print(f"[skip] {line}")
|
||||
continue
|
||||
data_str = line[len("data:"):].strip()
|
||||
if not data_str:
|
||||
continue
|
||||
|
||||
current_time = time.time()
|
||||
|
||||
if not first_chunk_received:
|
||||
first_chunk_received = True
|
||||
first_chunk_time = current_time
|
||||
print_log(f"🎯 首块数据到达，耗时: {first_chunk_time - response_received:.3f}s")
|
||||
|
||||
try:
|
||||
payload_obj = json.loads(data_str)
|
||||
except json.JSONDecodeError:
|
||||
print(f"[{event_type}] {data_str}")
|
||||
print_log(f"[{event_type}] {data_str}")
|
||||
event_type = "message"
|
||||
continue
|
||||
|
||||
if event_type == "end" or payload_obj.get("status") == "completed":
|
||||
print("\n[stream] completed")
|
||||
print_log(f"🏁 流式响应完成，总耗时: {current_time - client_start:.3f}s")
|
||||
print_log(f"📊 接收到的chunk数量: {chunk_count}")
|
||||
break
|
||||
if payload_obj.get("data") is True:
|
||||
event_type = "message"
|
||||
continue
|
||||
if "answer" in payload_obj:
|
||||
piece = payload_obj.get("answer", "")
|
||||
chunk_count += 1
|
||||
answer_parts.append(piece)
|
||||
# 流式输出：每个片段都单独打印，并且添加延时让效果更明显
|
||||
print(piece, end="", flush=True)
|
||||
time.sleep(0.1) # 每个片段间隔100ms，让流式效果更明显
|
||||
else:
|
||||
print(f"[{event_type}] {payload_obj}")
|
||||
print_log(f"[{event_type}] {payload_obj}")
|
||||
event_type = "message"
|
||||
|
||||
print("\n\n[stream completed] - Answer received in real-time via streaming!")
|
||||
print("\n\n[CLIENT] 流式响应完成 - 通过SSE实时接收！")
|
||||
full_answer = "".join(answer_parts)
|
||||
return full_answer
|
||||
|
||||
|
||||
@ -54,6 +54,12 @@ class RAGFlowClient:
|
||||
|
||||
def _stream_request(self, method: str, endpoint: str, **kwargs) -> Generator[Dict[str, Any], None, None]:
|
||||
"""发送流式HTTP请求，修复SSE缓冲问题"""
|
||||
import time
|
||||
|
||||
# 添加详细的时间戳日志
|
||||
server_start_time = time.time()
|
||||
print(f"[RAG_CLIENT {server_start_time:.3f}] 🔄 开始流式请求: {method} {endpoint}")
|
||||
|
||||
url = f"{self.base_url}{endpoint}"
|
||||
headers = kwargs.pop('headers', self.headers.copy())
|
||||
|
||||
@ -64,18 +70,39 @@ class RAGFlowClient:
|
||||
'X-Accel-Buffering': 'no' # Nginx禁用缓冲
|
||||
})
|
||||
|
||||
print(f"[RAG_CLIENT {time.time():.3f}] 📤 发送HTTP请求到: {url}")
|
||||
response = requests.request(method, url, headers=headers, stream=True, **kwargs)
|
||||
|
||||
response_time = time.time()
|
||||
print(f"[RAG_CLIENT {response_time:.3f}] 📥 HTTP响应接收完成，状态码: {response.status_code}")
|
||||
print(f"[RAG_CLIENT {response_time:.3f}] ⏱️ 连接建立耗时: {response_time - server_start_time:.3f}s")
|
||||
|
||||
# 立即处理响应，yield每个chunk
|
||||
chunk_count = 0
|
||||
first_chunk_sent = False
|
||||
|
||||
for line in response.iter_lines():
|
||||
if line:
|
||||
line_time = time.time()
|
||||
line = line.decode('utf-8')
|
||||
if line.startswith('data:'):
|
||||
try:
|
||||
data = json.loads(line[5:].strip())
|
||||
|
||||
if not first_chunk_sent:
|
||||
first_chunk_sent = True
|
||||
print(f"[RAG_CLIENT {line_time:.3f}] 🎯 首块数据开始发送，耗时: {line_time - response_time:.3f}s")
|
||||
|
||||
chunk_count += 1
|
||||
print(f"[RAG_CLIENT {line_time:.3f}] 📦 Chunk #{chunk_count} 发送中... (耗时: {line_time - server_start_time:.3f}s)")
|
||||
yield data
|
||||
except json.JSONDecodeError:
|
||||
print(f"[RAG_CLIENT {line_time:.3f}] ❌ JSON解析错误: {line}")
|
||||
continue
|
||||
|
||||
end_time = time.time()
|
||||
print(f"[RAG_CLIENT {end_time:.3f}] 🏁 流式请求完成，总耗时: {end_time - server_start_time:.3f}s")
|
||||
print(f"[RAG_CLIENT {end_time:.3f}] 📊 总共发送chunk数量: {chunk_count}")
|
||||
|
||||
# ====================
|
||||
# OpenAI兼容API
|
||||
|
||||
Loading…
Reference in New Issue
Block a user