mirror of
https://github.com/langgenius/dify.git
synced 2025-10-29 18:03:13 +00:00
63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
from contextlib import contextmanager
|
|
from datetime import datetime
|
|
from typing import Optional, Union
|
|
|
|
from extensions.ext_database import db
|
|
from models.model import Message
|
|
|
|
|
|
def filter_none_values(data: dict):
|
|
new_data = {}
|
|
for key, value in data.items():
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, datetime):
|
|
new_data[key] = value.isoformat()
|
|
else:
|
|
new_data[key] = value
|
|
return new_data
|
|
|
|
|
|
def get_message_data(message_id: str):
|
|
return db.session.query(Message).filter(Message.id == message_id).first()
|
|
|
|
|
|
@contextmanager
|
|
def measure_time():
|
|
timing_info = {"start": datetime.now(), "end": None}
|
|
try:
|
|
yield timing_info
|
|
finally:
|
|
timing_info["end"] = datetime.now()
|
|
|
|
|
|
def replace_text_with_content(data):
|
|
if isinstance(data, dict):
|
|
new_data = {}
|
|
for key, value in data.items():
|
|
if key == "text":
|
|
new_data["content"] = value
|
|
else:
|
|
new_data[key] = replace_text_with_content(value)
|
|
return new_data
|
|
elif isinstance(data, list):
|
|
return [replace_text_with_content(item) for item in data]
|
|
else:
|
|
return data
|
|
|
|
|
|
def generate_dotted_order(
|
|
run_id: str, start_time: Union[str, datetime], parent_dotted_order: Optional[str] = None
|
|
) -> str:
|
|
"""
|
|
generate dotted_order for langsmith
|
|
"""
|
|
start_time = datetime.fromisoformat(start_time) if isinstance(start_time, str) else start_time
|
|
timestamp = start_time.strftime("%Y%m%dT%H%M%S%f")[:-3] + "Z"
|
|
current_segment = f"{timestamp}{run_id}"
|
|
|
|
if parent_dotted_order is None:
|
|
return current_segment
|
|
|
|
return f"{parent_dotted_order}.{current_segment}"
|