删除非流式
This commit is contained in:
parent
04ce6c6019
commit
0a2cc4f474
@ -140,28 +140,22 @@ async def converse_with_chat_assistant(
|
||||
cache_similarity = static_sim
|
||||
cache_source = 'static_qa'
|
||||
logger.info(f'[RAG_SOURCE] 命中静态FAQ | chat_id={converse_params.chat_id} | question={converse_params.question} | similarity={cache_similarity:.2f} | answer_length={len(cached_answer)}')
|
||||
|
||||
# 非流式：直接返回静态问答答案
|
||||
if not converse_params.stream:
|
||||
return ResponseUtil.success(data={'answer': cached_answer, 'from_cache': True, 'similarity': cache_similarity, 'source': 'static_qa'})
|
||||
|
||||
# 流式：使用静态问答答案进行流式响应
|
||||
if converse_params.stream:
|
||||
logger.info(f'[StaticQA] 流式响应使用静态问答答案，chat_id={converse_params.chat_id}')
|
||||
return StreamingResponse(
|
||||
stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='static_qa'),
|
||||
media_type='text/event-stream',
|
||||
headers={
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
)
|
||||
logger.info(f'[StaticQA] 流式响应使用静态问答答案，chat_id={converse_params.chat_id}')
|
||||
return StreamingResponse(
|
||||
stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='static_qa'),
|
||||
media_type='text/event-stream',
|
||||
headers={
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f'[StaticQA] 静态问答匹配失败: {e}')
|
||||
|
||||
# ========== 2. RAG历史缓存查找 ==========
|
||||
logger.info(f'[SemanticCache] 准备执行RAG历史缓存查找 | redis={redis is not None} | chat_id={converse_params.chat_id}')
|
||||
if redis:
|
||||
lookup_hash = _get_question_hash(converse_params.question)
|
||||
logger.info(f'[SemanticCache] 开始查找 | chat_id={converse_params.chat_id} | question={converse_params.question} | hash={lookup_hash} | threshold=0.60')
|
||||
@ -175,24 +169,17 @@ async def converse_with_chat_assistant(
|
||||
cached_answer, cache_similarity = cache_result
|
||||
cache_source = 'rag_history'
|
||||
logger.info(f'[RAG_SOURCE] 命中RAG会话历史 | chat_id={converse_params.chat_id} | question={converse_params.question} | similarity={cache_similarity:.2f} | answer_length={len(cached_answer)}')
|
||||
|
||||
# 非流式：直接返回缓存答案
|
||||
if not converse_params.stream:
|
||||
return ResponseUtil.success(data={'answer': cached_answer, 'from_cache': True, 'similarity': cache_similarity, 'source': 'rag_history'})
|
||||
|
||||
# 流式：使用缓存答案进行流式响应
|
||||
if converse_params.stream:
|
||||
logger.info(f'[SemanticCache] 流式响应使用RAG历史缓存答案，chat_id={converse_params.chat_id}')
|
||||
return StreamingResponse(
|
||||
stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='rag_history'),
|
||||
media_type='text/event-stream',
|
||||
headers={
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
)
|
||||
logger.info(f'[SemanticCache] 流式响应使用RAG历史缓存答案，chat_id={converse_params.chat_id}')
|
||||
return StreamingResponse(
|
||||
stream_cached_response(cached_answer, converse_params.chat_id, start_time, cache_source='rag_history'),
|
||||
media_type='text/event-stream',
|
||||
headers={
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
)
|
||||
|
||||
# 检查是否应该使用搜索服务
|
||||
try:
|
||||
@ -204,17 +191,6 @@ async def converse_with_chat_assistant(
|
||||
except Exception as e:
|
||||
logger.warning(f"意图分类失败，使用RAG服务: {e}")
|
||||
|
||||
# 缓存检查（原有的精确缓存）- 非流式
|
||||
if not converse_params.stream and redis:
|
||||
cache_key = build_chat_cache_key(converse_params.chat_id, converse_params.question)
|
||||
try:
|
||||
cached = await redis.get(cache_key)
|
||||
if cached:
|
||||
logger.info('ragflow对话命中缓存: chat=%s', converse_params.chat_id)
|
||||
return ResponseUtil.success(json.loads(cached))
|
||||
except Exception as e:
|
||||
logger.warning(f"缓存获取失败: {e}")
|
||||
|
||||
# 直接使用同步RAGFlow服务
|
||||
try:
|
||||
# 恢复原始问题（包含语言风格提示词）
|
||||
@ -225,36 +201,32 @@ async def converse_with_chat_assistant(
|
||||
logger.info(f'[RAG_SOURCE] 调用原生RAG服务 | chat_id={converse_params.chat_id} | question={converse_params.question}')
|
||||
result = RAGFlowService.converse_with_chat_assistant_services(converse_params)
|
||||
|
||||
# 流式响应
|
||||
if converse_params.stream:
|
||||
# 注意：存储时使用清理后的问题（与查找时保持一致）
|
||||
cache_question = cleaned_question if style_removed else converse_params.question
|
||||
store_hash = _get_question_hash(cache_question)
|
||||
logger.info(f'[RAG_CACHE] 准备存储 | chat_id={converse_params.chat_id} | question={cache_question} | hash={store_hash}')
|
||||
|
||||
# 创建缓存存储回调函数
|
||||
async def make_cache_store(chat_id: str, question: str):
|
||||
async def store_answer(answer: str):
|
||||
if redis and answer and len(answer.strip()) >= 10:
|
||||
try:
|
||||
await _async_store_qa(chat_id, question, answer, redis)
|
||||
logger.info(f'[RAG_CACHE] 语义缓存存储成功 | chat_id={chat_id} | answer_length={len(answer)}')
|
||||
except Exception as e:
|
||||
logger.warning(f'语义缓存存储失败: {e}')
|
||||
return store_answer
|
||||
|
||||
cache_store = await make_cache_store(converse_params.chat_id, cache_question)
|
||||
|
||||
return StreamingResponse(
|
||||
stream_ragflow_response(result, converse_params.chat_id, start_time, cache_store_func=cache_store),
|
||||
media_type='text/event-stream',
|
||||
headers={
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
)
|
||||
cache_question = cleaned_question if style_removed else converse_params.question
|
||||
store_hash = _get_question_hash(cache_question)
|
||||
logger.info(f'[RAG_CACHE] 准备存储 | chat_id={converse_params.chat_id} | question={cache_question} | hash={store_hash}')
|
||||
|
||||
async def make_cache_store(chat_id: str, question: str):
|
||||
async def store_answer(answer: str):
|
||||
if redis and answer and len(answer.strip()) >= 10:
|
||||
try:
|
||||
await _async_store_qa(chat_id, question, answer, redis)
|
||||
logger.info(f'[RAG_CACHE] 语义缓存存储成功 | chat_id={chat_id} | answer_length={len(answer)}')
|
||||
except Exception as e:
|
||||
logger.warning(f'语义缓存存储失败: {e}')
|
||||
return store_answer
|
||||
|
||||
cache_store = await make_cache_store(converse_params.chat_id, cache_question)
|
||||
|
||||
return StreamingResponse(
|
||||
stream_ragflow_response(result, converse_params.chat_id, start_time, cache_store_func=cache_store),
|
||||
media_type='text/event-stream',
|
||||
headers={
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'X-Accel-Buffering': 'no',
|
||||
'Transfer-Encoding': 'chunked'
|
||||
}
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f'[RAG_PROFILE] ragflow对话异常: {e}')
|
||||
|
||||
@ -419,10 +419,13 @@ def get_semantic_cache_service() -> SemanticCacheService:
|
||||
# 便捷函数
|
||||
async def lookup_question(chat_id: str, question: str, redis_client=None) -> Optional[Tuple[str, float]]:
|
||||
"""查找缓存问题"""
|
||||
logger.info(f"[SemanticCache.lookup_question] 入参 | chat_id={chat_id} | question={question}")
|
||||
logger.info(f"[SemanticCache.lookup_question] 🔍 函数被调用 | chat_id={chat_id} | question={question} | redis_client={redis_client is not None}")
|
||||
service = get_semantic_cache_service()
|
||||
logger.info(f"[SemanticCache.lookup_question] 获取服务实例完成 | service={service is not None}")
|
||||
result = await service.lookup(chat_id, question, redis_client)
|
||||
logger.info(f"[SemanticCache.lookup_question] 返回 | result={result is not None}")
|
||||
if result:
|
||||
logger.info(f"[SemanticCache.lookup_question] 命中结果 | answer_length={len(result[0])} | similarity={result[1]:.2f}")
|
||||
return result
|
||||
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user