mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-07-18 05:37:24 +08:00
de06c9a60b
## Summary This PR passes `session_id` into Langfuse trace observations so multi-turn chat messages can be grouped under the same session in Langfuse. Changes include: - Propagate `session_id` from chat/session APIs into `dialog_service.async_chat`. - Pass `session_id` into Langfuse `start_observation(...)`. - Share Langfuse `trace_context` with chat, embedding, rerank, and TTS model bundles where applicable. - Add unit coverage to verify Langfuse observations receive `session_id`. - Update affected test stubs for the new optional Langfuse context arguments. ## Related Issue Closes: #15636 ## Change Type - [x] Feature - [x] Bug fix - [x] Test - [ ] Refactor - [ ] Documentation - [ ] Breaking change ## Real Behavior Proof Before this change: - Langfuse observations were created without `session_id`. - Multi-turn chat traces could not be grouped by session in Langfuse. After this change: - Chat/session flows pass `session_id` into `async_chat`. - Langfuse observations include `session_id`. - Related model bundles receive shared trace context and session metadata. Validation result: ```bash uv run python -m py_compile \ api/db/services/tenant_llm_service.py \ api/db/services/llm_service.py \ api/db/services/dialog_service.py \ api/db/services/conversation_service.py \ api/apps/restful_apis/chat_api.py \ test/unit_test/api/db/services/test_dialog_service_final_answer.py \ test/unit_test/api/db/services/test_dialog_service_use_sql_source_columns.py ``` Passed. ```bash uv run pytest \ test/unit_test/api/db/services/test_dialog_service_final_answer.py \ test/unit_test/api/db/services/test_dialog_service_use_sql_source_columns.py -q ``` Result: ```text 11 passed in 16.89s ``` ```bash git diff --check ``` Passed. ## Checklist - [x] Analyzed the issue requirement. - [x] Checked existing Langfuse trace integration. - [x] Implemented only the requested session grouping behavior. - [x] Added/updated unit tests. - [x] Ran focused tests successfully. - [x] Ran Python compile validation. - [x] Ran whitespace diff validation.
296 lines
11 KiB
Python
296 lines
11 KiB
Python
#
|
|
# Copyright 2024 The InfiniFlow Authors. All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
#
|
|
import time
|
|
import logging
|
|
from uuid import uuid4
|
|
from common.constants import StatusEnum
|
|
from api.db.db_models import Conversation, DB
|
|
from api.db.services.api_service import API4ConversationService
|
|
from api.db.services.common_service import CommonService
|
|
from api.db.services.dialog_service import DialogService, async_chat
|
|
from common.misc_utils import get_uuid
|
|
import json
|
|
|
|
from rag.prompts.generator import chunks_format
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ConversationService(CommonService):
|
|
model = Conversation
|
|
|
|
@classmethod
|
|
@DB.connection_context()
|
|
def get_list(cls, dialog_id, page_number, items_per_page, orderby, desc, id, name, user_id=None):
|
|
sessions = cls.model.select().where(cls.model.dialog_id == dialog_id)
|
|
if id:
|
|
sessions = sessions.where(cls.model.id == id)
|
|
if name:
|
|
sessions = sessions.where(cls.model.name == name)
|
|
if user_id:
|
|
sessions = sessions.where(cls.model.user_id == user_id)
|
|
if desc:
|
|
sessions = sessions.order_by(cls.model.getter_by(orderby).desc())
|
|
else:
|
|
sessions = sessions.order_by(cls.model.getter_by(orderby).asc())
|
|
|
|
if items_per_page > 0:
|
|
sessions = sessions.paginate(page_number, items_per_page)
|
|
|
|
return list(sessions.dicts())
|
|
|
|
@classmethod
|
|
@DB.connection_context()
|
|
def get_all_conversation_by_dialog_ids(cls, dialog_ids):
|
|
sessions = cls.model.select().where(cls.model.dialog_id.in_(dialog_ids))
|
|
sessions.order_by(cls.model.create_time.asc())
|
|
offset, limit = 0, 100
|
|
res = []
|
|
while True:
|
|
s_batch = sessions.offset(offset).limit(limit)
|
|
_temp = list(s_batch.dicts())
|
|
if not _temp:
|
|
break
|
|
res.extend(_temp)
|
|
offset += limit
|
|
return res
|
|
|
|
|
|
def structure_answer(conv, ans, message_id, session_id):
|
|
reference = ans["reference"]
|
|
if not isinstance(reference, dict):
|
|
reference = {}
|
|
ans["reference"] = {}
|
|
is_final = ans.get("final", True)
|
|
|
|
chunk_list = chunks_format(reference)
|
|
|
|
reference["chunks"] = chunk_list
|
|
ans["id"] = message_id
|
|
ans["session_id"] = session_id
|
|
|
|
if not conv:
|
|
return ans
|
|
|
|
if not conv.message:
|
|
conv.message = []
|
|
content = ans["answer"]
|
|
if ans.get("start_to_think"):
|
|
content = "<think>"
|
|
elif ans.get("end_to_think"):
|
|
content = "</think>"
|
|
|
|
if not conv.message or conv.message[-1].get("role", "") != "assistant":
|
|
conv.message.append({"role": "assistant", "content": content, "created_at": time.time(), "id": message_id})
|
|
else:
|
|
if is_final:
|
|
if ans.get("answer"):
|
|
conv.message[-1] = {"role": "assistant", "content": ans["answer"], "created_at": time.time(), "id": message_id}
|
|
else:
|
|
conv.message[-1]["created_at"] = time.time()
|
|
conv.message[-1]["id"] = message_id
|
|
else:
|
|
conv.message[-1]["content"] = (conv.message[-1].get("content") or "") + content
|
|
conv.message[-1]["created_at"] = time.time()
|
|
conv.message[-1]["id"] = message_id
|
|
if conv.reference:
|
|
should_update_reference = is_final or bool(reference.get("chunks")) or bool(reference.get("doc_aggs"))
|
|
if should_update_reference:
|
|
conv.reference[-1] = reference
|
|
return ans
|
|
|
|
|
|
async def async_completion(tenant_id, chat_id, question, name="New session", session_id=None, stream=True, **kwargs):
|
|
assert name, "`name` can not be empty."
|
|
dia = DialogService.query(id=chat_id, tenant_id=tenant_id, status=StatusEnum.VALID.value)
|
|
assert dia, "You do not own the chat."
|
|
|
|
if not session_id:
|
|
session_id = get_uuid()
|
|
conv = {
|
|
"id": session_id,
|
|
"dialog_id": chat_id,
|
|
"name": name,
|
|
"message": [{"role": "assistant", "content": dia[0].prompt_config.get("prologue"), "created_at": time.time()}],
|
|
"user_id": kwargs.get("user_id", "")
|
|
}
|
|
ConversationService.save(**conv)
|
|
if stream:
|
|
yield "data:" + json.dumps({"code": 0, "message": "",
|
|
"data": {
|
|
"answer": conv["message"][0]["content"],
|
|
"reference": {},
|
|
"audio_binary": None,
|
|
"id": None,
|
|
"session_id": session_id
|
|
}},
|
|
ensure_ascii=False) + "\n\n"
|
|
yield "data:" + json.dumps({"code": 0, "message": "", "data": True}, ensure_ascii=False) + "\n\n"
|
|
return
|
|
else:
|
|
answer = {
|
|
"answer": conv["message"][0]["content"],
|
|
"reference": {},
|
|
"audio_binary": None,
|
|
"id": None,
|
|
"session_id": session_id
|
|
}
|
|
yield answer
|
|
return
|
|
|
|
conv = ConversationService.query(id=session_id, dialog_id=chat_id)
|
|
if not conv:
|
|
raise LookupError("Session does not exist")
|
|
|
|
conv = conv[0]
|
|
msg = []
|
|
question = {
|
|
"content": question,
|
|
"role": "user",
|
|
"id": str(uuid4())
|
|
}
|
|
|
|
# Propagate runtime attachments so downstream chat flow can resolve file content.
|
|
if isinstance(kwargs.get("files"), list) and kwargs["files"]:
|
|
question["files"] = kwargs["files"]
|
|
|
|
conv.message.append(question)
|
|
for m in conv.message:
|
|
if m["role"] == "system":
|
|
continue
|
|
if m["role"] == "assistant" and not msg:
|
|
continue
|
|
msg.append(m)
|
|
message_id = msg[-1].get("id")
|
|
e, dia = DialogService.get_by_id(conv.dialog_id)
|
|
|
|
kb_ids = kwargs.get("kb_ids",[])
|
|
dia.kb_ids = list(set(dia.kb_ids + kb_ids))
|
|
if not conv.reference:
|
|
conv.reference = []
|
|
conv.message.append({"role": "assistant", "content": "", "id": message_id})
|
|
conv.reference.append({"chunks": [], "doc_aggs": []})
|
|
|
|
if stream:
|
|
try:
|
|
async for ans in async_chat(dia, msg, True, session_id=session_id, **kwargs):
|
|
ans = structure_answer(conv, ans, message_id, session_id)
|
|
yield "data:" + json.dumps({"code": 0, "data": ans}, ensure_ascii=False) + "\n\n"
|
|
ConversationService.update_by_id(conv.id, conv.to_dict())
|
|
except Exception as e:
|
|
yield "data:" + json.dumps({"code": 500, "message": str(e),
|
|
"data": {"answer": "**ERROR**: " + str(e), "reference": []}},
|
|
ensure_ascii=False) + "\n\n"
|
|
yield "data:" + json.dumps({"code": 0, "data": True}, ensure_ascii=False) + "\n\n"
|
|
|
|
else:
|
|
answer = None
|
|
async for ans in async_chat(dia, msg, False, session_id=session_id, **kwargs):
|
|
answer = structure_answer(conv, ans, message_id, session_id)
|
|
ConversationService.update_by_id(conv.id, conv.to_dict())
|
|
break
|
|
yield answer
|
|
|
|
async def async_iframe_completion(dialog_id, question, session_id=None, stream=True, tenant_id=None, **kwargs):
|
|
if tenant_id:
|
|
exists, dia = DialogService.get_by_id(dialog_id)
|
|
if (not exists
|
|
or getattr(dia, "tenant_id", None) != tenant_id
|
|
or str(getattr(dia, "status", "")) != StatusEnum.VALID.value):
|
|
logger.warning(
|
|
"Dialog lookup failed for tenant-scoped iframe completion: "
|
|
"tenant_id=%s dialog_id=%s required_status=%s",
|
|
tenant_id,
|
|
dialog_id,
|
|
StatusEnum.VALID.value,
|
|
)
|
|
raise AssertionError("Dialog not found")
|
|
else:
|
|
e, dia = DialogService.get_by_id(dialog_id)
|
|
assert e, "Dialog not found"
|
|
if not session_id:
|
|
session_id = get_uuid()
|
|
conv = {
|
|
"id": session_id,
|
|
"dialog_id": dialog_id,
|
|
"user_id": kwargs.get("user_id", ""),
|
|
"message": [{"role": "assistant", "content": dia.prompt_config["prologue"], "created_at": time.time()}]
|
|
}
|
|
API4ConversationService.save(**conv)
|
|
yield "data:" + json.dumps({"code": 0, "message": "",
|
|
"data": {
|
|
"answer": conv["message"][0]["content"],
|
|
"reference": {},
|
|
"audio_binary": None,
|
|
"id": None,
|
|
"session_id": session_id
|
|
}},
|
|
ensure_ascii=False) + "\n\n"
|
|
yield "data:" + json.dumps({"code": 0, "message": "", "data": True}, ensure_ascii=False) + "\n\n"
|
|
return
|
|
else:
|
|
session_id = session_id
|
|
e, conv = API4ConversationService.get_by_id(session_id)
|
|
assert e, "Session not found!"
|
|
assert conv.dialog_id == dialog_id, "Session does not belong to this dialog"
|
|
|
|
if not conv.message:
|
|
conv.message = []
|
|
messages = conv.message
|
|
question = {
|
|
"role": "user",
|
|
"content": question,
|
|
"id": str(uuid4())
|
|
}
|
|
messages.append(question)
|
|
|
|
msg = []
|
|
for m in messages:
|
|
if m["role"] == "system":
|
|
continue
|
|
if m["role"] == "assistant" and not msg:
|
|
continue
|
|
msg.append(m)
|
|
if not msg[-1].get("id"):
|
|
msg[-1]["id"] = get_uuid()
|
|
message_id = msg[-1]["id"]
|
|
|
|
if not conv.reference:
|
|
conv.reference = []
|
|
conv.reference.append({"chunks": [], "doc_aggs": []})
|
|
|
|
if stream:
|
|
try:
|
|
async for ans in async_chat(dia, msg, True, session_id=session_id, **kwargs):
|
|
ans = structure_answer(conv, ans, message_id, session_id)
|
|
yield "data:" + json.dumps({"code": 0, "message": "", "data": ans},
|
|
ensure_ascii=False) + "\n\n"
|
|
API4ConversationService.append_message(conv.id, conv.to_dict())
|
|
except Exception as e:
|
|
yield "data:" + json.dumps({"code": 500, "message": str(e),
|
|
"data": {"answer": "**ERROR**: " + str(e), "reference": []}},
|
|
ensure_ascii=False) + "\n\n"
|
|
yield "data:" + json.dumps({"code": 0, "message": "", "data": True}, ensure_ascii=False) + "\n\n"
|
|
|
|
else:
|
|
answer = None
|
|
async for ans in async_chat(dia, msg, False, session_id=session_id, **kwargs):
|
|
answer = structure_answer(conv, ans, message_id, session_id)
|
|
API4ConversationService.append_message(conv.id, conv.to_dict())
|
|
break
|
|
yield answer
|