utils.py 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190
  1. import re
  2. import logging
  3. from typing import List
  4. from config import SRC_LOG_LEVELS, CHROMA_CLIENT
  5. log = logging.getLogger(__name__)
  6. log.setLevel(SRC_LOG_LEVELS["RAG"])
  7. def query_doc(collection_name: str, query: str, k: int, embedding_function):
  8. try:
  9. # if you use docker use the model from the environment variable
  10. collection = CHROMA_CLIENT.get_collection(
  11. name=collection_name,
  12. embedding_function=embedding_function,
  13. )
  14. result = collection.query(
  15. query_texts=[query],
  16. n_results=k,
  17. )
  18. return result
  19. except Exception as e:
  20. raise e
  21. def merge_and_sort_query_results(query_results, k):
  22. # Initialize lists to store combined data
  23. combined_ids = []
  24. combined_distances = []
  25. combined_metadatas = []
  26. combined_documents = []
  27. # Combine data from each dictionary
  28. for data in query_results:
  29. combined_ids.extend(data["ids"][0])
  30. combined_distances.extend(data["distances"][0])
  31. combined_metadatas.extend(data["metadatas"][0])
  32. combined_documents.extend(data["documents"][0])
  33. # Create a list of tuples (distance, id, metadata, document)
  34. combined = list(
  35. zip(combined_distances, combined_ids, combined_metadatas, combined_documents)
  36. )
  37. # Sort the list based on distances
  38. combined.sort(key=lambda x: x[0])
  39. # Unzip the sorted list
  40. sorted_distances, sorted_ids, sorted_metadatas, sorted_documents = zip(*combined)
  41. # Slicing the lists to include only k elements
  42. sorted_distances = list(sorted_distances)[:k]
  43. sorted_ids = list(sorted_ids)[:k]
  44. sorted_metadatas = list(sorted_metadatas)[:k]
  45. sorted_documents = list(sorted_documents)[:k]
  46. # Create the output dictionary
  47. merged_query_results = {
  48. "ids": [sorted_ids],
  49. "distances": [sorted_distances],
  50. "metadatas": [sorted_metadatas],
  51. "documents": [sorted_documents],
  52. "embeddings": None,
  53. "uris": None,
  54. "data": None,
  55. }
  56. return merged_query_results
  57. def query_collection(
  58. collection_names: List[str], query: str, k: int, embedding_function
  59. ):
  60. results = []
  61. for collection_name in collection_names:
  62. try:
  63. # if you use docker use the model from the environment variable
  64. collection = CHROMA_CLIENT.get_collection(
  65. name=collection_name,
  66. embedding_function=embedding_function,
  67. )
  68. result = collection.query(
  69. query_texts=[query],
  70. n_results=k,
  71. )
  72. results.append(result)
  73. except:
  74. pass
  75. return merge_and_sort_query_results(results, k)
  76. def rag_template(template: str, context: str, query: str):
  77. template = template.replace("[context]", context)
  78. template = template.replace("[query]", query)
  79. return template
  80. def rag_messages(docs, messages, template, k, embedding_function):
  81. log.debug(f"docs: {docs}")
  82. last_user_message_idx = None
  83. for i in range(len(messages) - 1, -1, -1):
  84. if messages[i]["role"] == "user":
  85. last_user_message_idx = i
  86. break
  87. user_message = messages[last_user_message_idx]
  88. if isinstance(user_message["content"], list):
  89. # Handle list content input
  90. content_type = "list"
  91. query = ""
  92. for content_item in user_message["content"]:
  93. if content_item["type"] == "text":
  94. query = content_item["text"]
  95. break
  96. elif isinstance(user_message["content"], str):
  97. # Handle text content input
  98. content_type = "text"
  99. query = user_message["content"]
  100. else:
  101. # Fallback in case the input does not match expected types
  102. content_type = None
  103. query = ""
  104. relevant_contexts = []
  105. for doc in docs:
  106. context = None
  107. try:
  108. if doc["type"] == "collection":
  109. context = query_collection(
  110. collection_names=doc["collection_names"],
  111. query=query,
  112. k=k,
  113. embedding_function=embedding_function,
  114. )
  115. elif doc["type"] == "text":
  116. context = doc["content"]
  117. else:
  118. context = query_doc(
  119. collection_name=doc["collection_name"],
  120. query=query,
  121. k=k,
  122. embedding_function=embedding_function,
  123. )
  124. except Exception as e:
  125. log.exception(e)
  126. context = None
  127. relevant_contexts.append(context)
  128. log.debug(f"relevant_contexts: {relevant_contexts}")
  129. context_string = ""
  130. for context in relevant_contexts:
  131. if context:
  132. context_string += " ".join(context["documents"][0]) + "\n"
  133. ra_content = rag_template(
  134. template=template,
  135. context=context_string,
  136. query=query,
  137. )
  138. if content_type == "list":
  139. new_content = []
  140. for content_item in user_message["content"]:
  141. if content_item["type"] == "text":
  142. # Update the text item's content with ra_content
  143. new_content.append({"type": "text", "text": ra_content})
  144. else:
  145. # Keep other types of content as they are
  146. new_content.append(content_item)
  147. new_user_message = {**user_message, "content": new_content}
  148. else:
  149. new_user_message = {
  150. **user_message,
  151. "content": ra_content,
  152. }
  153. messages[last_user_message_idx] = new_user_message
  154. return messages