gmakstutis/LightRAG
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor MD5 hash functions and consolidate Unicode error handling

Browse source
This commit is contained in:
yangdx 2025-08-19 17:49:23 +08:00
parent 64058c771f
commit 64015548df
1 changed files with 11 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
lightrag/utils.py
View file

@ -272,19 +272,26 @@ def compute_args_hash(*args: Any) -> str:
Returns: Returns:
str: Hash string str: Hash string
""" """
import hashlib
# Convert all arguments to strings and join them # Convert all arguments to strings and join them
args_str = "".join([str(arg) for arg in args]) args_str = "".join([str(arg) for arg in args])
# Use 'replace' error handling to safely encode problematic Unicode characters # Use 'replace' error handling to safely encode problematic Unicode characters
# This replaces invalid characters with Unicode replacement character (U+FFFD) # This replaces invalid characters with Unicode replacement character (U+FFFD)
try: try:
return hashlib.md5(args_str.encode("utf-8")).hexdigest() return md5(args_str.encode("utf-8")).hexdigest()
except UnicodeEncodeError: except UnicodeEncodeError:
# Handle surrogate characters and other encoding issues # Handle surrogate characters and other encoding issues
safe_bytes = args_str.encode("utf-8", errors="replace") safe_bytes = args_str.encode("utf-8", errors="replace")
return hashlib.md5(safe_bytes).hexdigest() return md5(safe_bytes).hexdigest()
def compute_mdhash_id(content: str, prefix: str = "") -> str:
"""
Compute a unique ID for a given content string.
The ID is a combination of the given prefix and the MD5 hash of the content string.
"""
return prefix + compute_args_hash(content)
def generate_cache_key(mode: str, cache_type: str, hash_value: str) -> str: def generate_cache_key(mode: str, cache_type: str, hash_value: str) -> str:
@ -316,15 +323,6 @@ def parse_cache_key(cache_key: str) -> tuple[str, str, str] | None:
return None return None
def compute_mdhash_id(content: str, prefix: str = "") -> str:
"""
Compute a unique ID for a given content string.
The ID is a combination of the given prefix and the MD5 hash of the content string.
"""
return prefix + md5(content.encode()).hexdigest()
# Custom exception class # Custom exception class
class QueueFullError(Exception): class QueueFullError(Exception):
"""Raised when the queue is full and the wait times out""" """Raised when the queue is full and the wait times out"""