Skip to content
GitLab
Commit 9e70c543 authored by minhnguyengp1's avatar minhnguyengp1
Browse files

eval pipelne WIP

parent 77c9ccb6
Expand all Hide whitespace changes
Inline Side-by-side
agents/agent_decision.py deleted 100644 → 0
View file @ 77c9ccb6
from typing import Dict, Optional, Union, List, TypedDict, Any, Literal
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage
from langgraph.graph import MessagesState, StateGraph, END
from agents.message_filter import MessageFilter
from agents.rag_agent.buffer_memory import WindowBufferMemory
from config import Config
from agents.rag_agent import ResearchAssistantRAG
from langgraph.checkpoint.memory import MemorySaver
load_dotenv()
config = Config()
memory = MemorySaver()
thread_config = {"configurable": {"thread_id": "1"}}
class AgentState(MessagesState):
"""State maintained across the workflow."""
messages: List[BaseMessage]
current_input: Optional[Union[str, Dict]]
output: Optional[str]
retrieval_confidence: float
sources: Optional[List[str]] # retrieved_chunks
def create_agent_graph():
message_filter = MessageFilter(config.rag.llm)
def run_rag_agent(state: AgentState) -> AgentState:
print(f"Selected agent: RAG_AGENT")
new_state = state.copy()
rag_agent = ResearchAssistantRAG(config)
# messages: List[BaseMessage] = state["messages"]
# query: Union[str, Dict] = state["current_input"]
# rag_context_limit: int = config.rag.context_limit
#
# print(f"DEBUG messages: {messages}")
#
# buffer_memory = WindowBufferMemory(context_limit=rag_context_limit)
# chat_history: List[Dict[str, str]] = buffer_memory.build_context(messages)
#
# print("DEBUG chat_history input:", chat_history)
# print("DEBUG type of chat_history:", type(chat_history))
messages = state["messages"]
query = state["current_input"]
rag_context_limit = config.rag.context_limit
print(f"checkdd")
recent_context = ""
for msg in messages[-rag_context_limit:]: # limit controlled from config
if isinstance(msg, HumanMessage):
# print("######### DEBUG 1:", msg)
recent_context += f"User: {msg.content}\n"
elif isinstance(msg, AIMessage):
# print("######### DEBUG 2:", msg)
recent_context += f"Assistant: {msg.content}\n"
# if chat_history:
# for i, msg in enumerate(chat_history):
# print(f"DEBUG chat_history[{i}]: {msg} (type: {type(msg)})")
print("Before processing query:", recent_context)
response: Dict[str, Any] = rag_agent.process_query(query, chat_history=recent_context)
### CHECKPOINT
print("Final response from RAG agent:", response)
retrieval_confidence: float = response.get("confidence", 0.0)
print(f"Retrieval Confidence: {retrieval_confidence}")
print(f"Sources: {len(response['sources'])}")
insufficient_info = False
response_content = response["response"]
if hasattr(response_content, 'content'):
response_text = response_content.content
else:
response_text = response_content
print(f"Response text type: {type(response_text)}")
print(f"Response text preview: {response_text[:100]}...")
# TODO: need to be improved because the logic is still "lord"
if isinstance(response_text, str) and (
"I don't have enough information to answer this question based on the provided context" in response_text or
"I don't have enough information" in response_text or
"don't have enough information" in response_text.lower() or
"not enough information" in response_text.lower() or
"insufficient information" in response_text.lower() or
"cannot answer" in response_text.lower() or
"unable to answer" in response_text.lower()
):
print("RAG response indicates insufficient information")
print(f"Response text that triggered insufficient_info: {response_text[:100]}...")
# Store RAG output ONLY if confidence is high
if retrieval_confidence >= config.rag.min_retrieval_confidence:
response_output = AIMessage(content=response_text)
else:
response_output = AIMessage(content=response_text) # should be = ""
updated_messages = messages + [AIMessage(content=response_text)]
new_state.update({
**state,
"messages": updated_messages,
"output": response_output,
"retrieval_confidence": retrieval_confidence,
"sources": response.get("chunks", [])
})
print(f"new_state in run_rag_agent: {new_state}")
return new_state
def apply_message_filter(state: AgentState) -> AgentState:
"""Apply output guardrails to the generated response."""
new_state = state.copy()
output = state["output"]
current_input = state["current_input"]
# Check if output is valid
if not output or not isinstance(output, (str, AIMessage)):
return state
# Get the original input text
input_text = ""
if isinstance(current_input, str):
input_text = current_input
elif isinstance(current_input, dict):
input_text = current_input.get("text", "")
output_text = ""
if isinstance(output, str):
output_text = output
elif isinstance(output, AIMessage):
output_text = output.content
sanitized_output = message_filter.validate_output(output_text, input_text)
# For non-validation cases, add the sanitized output to messages
sanitized_message = AIMessage(content=sanitized_output) if isinstance(output, AIMessage) else sanitized_output
updated_messages = state.get("messages", [])
if isinstance(updated_messages, list):
updated_messages = updated_messages + [sanitized_message]
else:
updated_messages = [sanitized_message] # ✅ Thêm fallback nếu messages bị sai kiểu
new_state.update({
**state,
"messages": updated_messages,
"output": sanitized_message
})
print(f"new_state in apply_message_filter: {new_state}")
return new_state
# === Create LangGraph ===
graph = StateGraph(AgentState)
graph.add_node("RAG_AGENT", run_rag_agent)
graph.add_node("MESSAGE_FILTER", apply_message_filter)
graph.set_entry_point("RAG_AGENT")
graph.add_edge("RAG_AGENT", "MESSAGE_FILTER")
graph.add_edge("MESSAGE_FILTER", END)
return graph.compile(checkpointer=memory)
def init_agent_state() -> AgentState:
"""Initialize the agent state with default values."""
return {
"messages": [],
"current_input": None,
"output": None,
"retrieval_confidence": 0.0,
"sources": [],
}
def process_query(query: Union[str, Dict], conversation_history: List[BaseMessage] = None) -> Dict[str, Any]:
graph = create_agent_graph()
state = init_agent_state()
print(f"conversation_history {conversation_history}")
input_text = query if isinstance(query, str) else query.get("text", "")
if conversation_history:
state["messages"] = conversation_history
else:
state["messages"] = [HumanMessage(content=input_text)]
state["current_input"] = query
print(f"state first: {state}")
state = graph.invoke(state, thread_config)
output = state.get("output")
if hasattr(output, "content"):
response_text = output.content
else:
response_text = str(output)
# # Keep history to reasonable size (ANOTHER OPTION: summarize and store before truncating history)
# if len(result["messages"]) > config.max_conversation_history: # Keep last config.max_conversation_history messages
# result["messages"] = result["messages"][-config.max_conversation_history:]
#
# # visualize conversation history in console
# for m in result["messages"]:
# m.pretty_print()
print(f"state after: {state}")
return {
"response": response_text,
"messages": state.get("messages", []),
"chunks": state.get("sources", []),
}
\ No newline at end of file
agents/rag_agent/__init__.py deleted 100644 → 0
View file @ 77c9ccb6
import logging
import time
from typing import List, Dict, Any, Optional
import uuid
from pathlib import Path
import json
from .content_processor import ContentProcessor
from .doc_parser import DocParser
from .query_expander import QueryExpander
from .query_processor import QueryProcessor
from .reranker import Reranker
from .response_generator import ResponseGenerator
from .multi_query_generator import MultiQueryGenerator
from .query_rewriter import QueryRewriter
from .vectorstore_qdrant import VectorStore
class ResearchAssistantRAG:
def __init__(self, config):
self.config = config
# self._initialize()
self.logger = logging.getLogger(f"{self.__module__}")
self.logger.info("Initializing Research Assistant RAG system")
self.doc_parser = DocParser()
self.content_processor = ContentProcessor(config)
self.parsed_content_dir = self.config.rag.parsed_content_dir
self.vector_store = VectorStore(config)
self.reranker = Reranker(config)
self.response_generator = ResponseGenerator(config)
self.parsed_content_dir = self.config.rag.parsed_content_dir
def ingest_file(self, document_path: str) -> Dict[str, Any]:
"""
Ingest a single file into the RAG system.
Args:
document_path: Path to the file to ingest
Returns:
Dictionary with ingestion results
"""
start_time = time.time()
self.logger.info(f"Ingesting file: {document_path}")
try:
# Step 1: Parse document
self.logger.info("1. Parsing document and extracting images...")
parsed_document, images = self.doc_parser.parse_document(document_path, self.parsed_content_dir)
self.logger.info(f" Parsed document and extracted {len(images)} images")
# Step 2: Summarize images
self.logger.info("2. Summarizing images...")
image_summaries = self.content_processor.summarize_images(images)
self.logger.info(f" Generated {len(image_summaries)} image summaries")
# Step 3: Format document with image summaries
self.logger.info("3. Formatting document with image summaries...")
formatted_document = self.content_processor.format_document_with_images(parsed_document, image_summaries)
# Save parsed_document to data/parsed
parsed_dir = Path("data/parsed")
parsed_dir.mkdir(exist_ok=True, parents=True)
parsed_path = parsed_dir / (Path(document_path).stem + "_parsed.json")
with open(parsed_path, "w", encoding="utf-8") as f:
import json
json.dump(formatted_document, f, ensure_ascii=False, indent=2, default=str)
# Step 4: Chunk document into semantic sections
self.logger.info("4. Chunking document into semantic sections...")
document_chunks: List[str] = self.content_processor.chunk_document(formatted_document)
self.logger.info(f" Document split into {len(document_chunks)} chunks")
with open("document_chunks.json", "w", encoding="utf-8") as f:
json.dump(document_chunks, f, ensure_ascii=False, indent=2)
# Step 5: Create vector store and document store
self.logger.info("5. Creating vector store knowledge base...")
self.vector_store.create_vectorstore(
document_chunks=document_chunks,
document_path=document_path
)
return {
"success": True,
"documents_ingested": 1,
"chunks_processed": len(document_chunks),
"processing_time": time.time() - start_time
}
except Exception as e:
self.logger.error(f"Error ingesting file: {e}")
return {
"success": False,
"error": str(e),
"processing_time": time.time() - start_time
}
def process_query(self, query: str, chat_history: Optional[List[Dict[str, str]]] = None) -> Dict[str, Any]:
"""
Process a query with the RAG system.
Args:
query: The query string
chat_history: Optional chat history for context
Returns:
Response dictionary
"""
start_time = time.time()
self.logger.info(f"RAG Agent processing query: {query}")
# Process query and return result, passing chat_history
try:
# Step 1: Expand query - REWRITE QUERY
# self.logger.info(f"1. Expanding query: '{query}'")
# expansion_result = self.query_expander.expand_query(query)
# expanded_query = expansion_result["expanded_query"]
# self.logger.info(f" Original: '{query}'")
# self.logger.info(f" Expanded: '{expanded_query}'")
# query = expanded_query
# Step 2: Retrieval
self.logger.info(f"2. Retrieving relevant documents for the query: '{query}'")
vectorstore, docstore = self.vector_store.load_vectorstore()
retrieved_documents = self.vector_store.retrieve_relevant_chunks(
query=query,
vectorstore=vectorstore,
docstore=docstore,
)
self.logger.info(f" Retrieved {len(retrieved_documents)} relevant document chunks")
print(f"type of retrieved_documents = {type(retrieved_documents)}")
print(f"retrieved_documents = {retrieved_documents}")
# Step 3: Rerank the retrieved documents if we have a reranker and enough documents
self.logger.info(f"3. Reranking the retrieved documents")
if self.reranker and len(retrieved_documents) > 1:
# reranked_documents, reranked_top_k_picture_paths = self.reranker.rerank(query, retrieved_documents,
# self.parsed_content_dir)
reranked_documents = self.reranker.rerank(query, retrieved_documents,
self.parsed_content_dir)
self.logger.info(f" Reranked retrieved documents and chose top {len(reranked_documents)}")
# self.logger.info(f" Found {len(reranked_top_k_picture_paths)} referenced images")
else:
self.logger.info(f" Could not rerank the retrieved documents, falling back to original scores")
reranked_documents = retrieved_documents
reranked_top_k_picture_paths = []
# Step 4: Generate response
self.logger.info("4. Generating response...")
response = self.response_generator.generate_response(
query=query,
retrieved_docs=reranked_documents,
# picture_paths=reranked_top_k_picture_paths,
chat_history=chat_history
)
print(f"type(reranked_documents) = {type(reranked_documents)}")
print(f"reranked_documents = {reranked_documents}")
# Add timing information
processing_time = time.time() - start_time
response["processing_time"] = processing_time
response["chunks"] = [doc["content"] for doc in reranked_documents]
return response
except Exception as e:
self.logger.error(f"Error processing query: {e}")
import traceback
self.logger.error(traceback.format_exc())
# Return error response
return {
"response": f"I encountered an error while processing your query: {str(e)}",
"sources": [],
"confidence": 0.0,
"processing_time": time.time() - start_time
}
def _ensure_json_serializable(self, obj):
if isinstance(obj, dict):
return {k: self._ensure_json_serializable(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [self._ensure_json_serializable(item) for item in obj]
elif isinstance(obj, (str, int, float, bool, type(None))):
return obj
else:
return str(obj)
def ingest_directory(self, directory_path: str, file_extension: Optional[str] = None) -> Dict[str, Any]:
pass
agents/rag_agent/multi_query_generator.py deleted 100644 → 0
View file @ 77c9ccb6
from typing import List, Dict, Optional
from langchain_core.language_models import BaseLanguageModel
class MultiQueryGenerator:
def __init__(self, llm: BaseLanguageModel, max_queries: int = 5):
"""
MQ (Multi-query) Generator: Sinh nhiều truy vấn từ câu hỏi gốc và ngữ cảnh.
Có thể nâng cấp sang MQA dễ dàng trong tương lai.
Args:
llm: LLM dùng để sinh truy vấn (GPT-3.5, GPT-4...)
max_queries: Số truy vấn tối đa muốn sinh
"""
self.llm = llm
self.max_queries = max_queries
def generate_queries(
self,
user_question: str,
chat_history: Optional[List[Dict[str, str]]] = None,
persona: Optional[List[str]] = None,
initial_answer: Optional[str] = None # nếu sau này dùng MQA, có thể truyền vào đây
) -> List[str]:
"""
Sinh ra các truy vấn tìm kiếm từ câu hỏi và ngữ cảnh hội thoại.
Args:
user_question: câu hỏi cuối cùng của người dùng
chat_history: lịch sử hội thoại (list gồm {"role": "user"|"ai", "content": ...})
persona: thông tin nền của người dùng (dùng sau cho iKAT)
initial_answer: câu trả lời sơ bộ (dành cho nâng cấp sang MQA)
Returns:
List[str]: Danh sách các truy vấn con
"""
history_text = "\n".join(f"{msg['role'].capitalize()}: {msg['content']}" for msg in chat_history or [])
persona_text = "\n".join(persona or [])
# Prompt template có thể được mở rộng sang MQA nếu truyền initial_answer
prompt = f"""
Du bist ein intelligenter Suchassistent.
Ich werde dir ein Gespräch zwischen einem Nutzer und einem System sowie die letzte Frage des Nutzers geben.
Deine Aufgabe ist es, bis zu {self.max_queries} präzisere Suchanfragen zu generieren, die dabei helfen können, die gesuchte Information besser zu finden.
# Hintergrundinformationen zum Nutzer:
{persona_text}
# Gesprächsverlauf:
{history_text}
# Letzte Nutzerfrage:
{user_question}
# Zu generierende Suchanfragen:
(schreibe jede Anfrage in eine neue Zeile, ohne Nummerierung)
"""
print(f"search-prompt: \n{prompt}")
response = self.llm.invoke(prompt)
if hasattr(response, "content"):
response_text = response.content
else:
response_text = response # fallback nếu là str
queries = [line.strip() for line in response_text.strip().splitlines() if line.strip()]
return queries[:self.max_queries]
agents/rag_agent/query_expander.py deleted 100644 → 0
View file @ 77c9ccb6
import logging
from typing import List, Dict, Any
class QueryExpander:
def __init__(self, config):
self.logger = logging.getLogger(f"{self.__module__}")
self.config = config
self.model = config.rag.llm
def expand_query(self, original_query: str) -> Dict[str, Any]:
"""
Expand the original query with relevant academic and scientific terms.
Args:
original_query: The user's original query
Returns:
Dictionary with original and expanded queries
"""
self.logger.info(f"Expanding query: {original_query}")
# Generate expansions
# Step 1: expand the query using synonyms and related terms (using a predefined thesaurus or external source)
expanded_query = self._expand_with_synonyms_and_related_terms(original_query)
# Step 2: use llm to expand query semantically
expanded_query = self._expand_with_semantics(expanded_query)
return {
"original_query": original_query,
"expanded_query": expanded_query
}
def _expand_with_synonyms_and_related_terms(self, query: str) -> str:
"""Expand the query by adding synonyms and related terms using a predefined thesaurus or domain knowledge."""
# Define a dictionary for basic synonyms/related terms (this can be expanded based on your domain)
synonym_dict = {
"artificial intelligence": ["machine learning", "deep learning", "neural networks", "AI"],
"data science": ["big data", "data analysis", "machine learning", "data mining"],
}
expanded_query = query.lower() # Convert to lowercase for matching
# Expand using synonyms from the dictionary
for key, synonyms in synonym_dict.items():
if key.lower() in expanded_query:
expanded_query += " OR " + " OR ".join(synonyms)
self.logger.info(f"Expanded query after synonym addition: {expanded_query}")
# return expanded_query
### For now: do nothing and return original query
return query
def _expand_with_semantics(self, query: str) -> str:
"""Use LLM to expand query with academic and scientific terminology."""
prompt = f"""
As a research assistant, expand the following query with relevant scientific terminology,
synonyms, related concepts, and keywords that would help in retrieving relevant academic papers and research articles:
User Query: {query}
Expand the query only if you feel like it is required, otherwise keep the user query intact.
Be specific to the field of study mentioned, do not add other irrelevant domains.
If the user query asks about answering in tabular format, include that in the expanded query and do not answer in tabular format yourself.
Provide only the expanded query without explanations.
"""
expansion = self.model.invoke(prompt)
return expansion
\ No newline at end of file
agents/rag_agent/query_processor.py deleted 100644 → 0
View file @ 77c9ccb6
import logging
import re
import uuid
from typing import List, Dict, Any, Optional, Tuple, Union
from datetime import datetime
class QueryProcessor:
def __init__(self, config, embedding_model, sparse_embedder):
self.logger = logging.getLogger(__name__)
self.embedding_model = embedding_model
self.sparse_embedder = sparse_embedder
self.synonyms = {
"zusammenfassung": ["summary", "zusammenfassen", "kurzfassung"],
"artikel": ["article", "studie", "paper"],
"projekt": ["project", "forschung", "studie"],
"aktuellen": ["aktuellen", "neuesten", "recent", "latest"],
}
self.intent_patterns = {
"summary": re.compile(r"\b(zusammenfassung|summary)\b", re.IGNORECASE),
"project_search": re.compile(r"\b(projekt|forschung|study|project)\b", re.IGNORECASE),
"article_navigation": re.compile(r"\b(link|artikel|paper|artikel)\b", re.IGNORECASE),
}
def process_query(self, query: str) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""
Process the query to generate embedding and extract metadata filters.
Args:
query: User query string
Returns:
Tuple of (query_embedding, extracted_filters)
"""
try:
query_id = str(uuid.uuid4())
expanded_query = self._expand_query(query)
query_intent = self._detect_query_intent(query)
query_dense_embedding = self.embedding_model.embed_query(expanded_query)
if query_dense_embedding is None:
raise ValueError("Embedding model returned None for query.")
query_dense_embedding = query_dense_embedding.tolist() if hasattr(query_dense_embedding, "tolist") else query_dense_embedding
query_sparse_embeddings = list(self.sparse_embedder.query_embed(query))
if query_sparse_embeddings:
query_sparse_embedding = query_sparse_embeddings[0]
else:
query_sparse_embedding = {}
filters: Dict[str, Any] = {
"query_id": query_id,
"timestamp": datetime.now().isoformat(),
"query_intent": query_intent,
}
filters = {k: v for k, v in filters.items() if v is not None}
self.logger.info(f"Processed query with filters: {filters}")
return {
"dense": query_dense_embedding,
"sparse": query_sparse_embedding
}, filters
except Exception as e:
self.logger.error(f"Error processing query: {e}")
fallback_dense_embedding = self.embedding_model.embed_query(query)
if fallback_dense_embedding is None:
fallback_dense_embedding = []
fallback_dense_embedding = fallback_dense_embedding.tolist() if hasattr(fallback_dense_embedding, "tolist") else fallback_dense_embedding
try:
fallback_sparse_embeddings = list(self.sparse_embedder.query_embed(query))
fallback_sparse_embedding = fallback_sparse_embeddings[0] if fallback_sparse_embeddings else {}
except Exception as se:
self.logger.error(f"Error generating sparse fallback: {se}")
fallback_sparse_embedding = {}
fallback_filters: Dict[str, Any] = {
"query_id": str(uuid.uuid4()),
"timestamp": datetime.now().isoformat(),
"is_fallback": True
}
return {
"dense": fallback_dense_embedding,
"sparse": fallback_sparse_embedding
}, fallback_filters
def _expand_query(self, query: str) -> str:
parts = [query]
low = query.lower()
for key, syns in self.synonyms.items():
if key in low:
parts.extend(syns)
return " ".join(dict.fromkeys(parts))
def _detect_query_intent(self, query: str) -> str:
for intent, pattern in self.intent_patterns.items():
if pattern.search(query):
return intent
return "general_information"
agents/rag_agent/vector_store.py deleted 100644 → 0
View file @ 77c9ccb6
# from typing import List, Dict, Any, Optional, Union
# import logging
# from qdrant_client import QdrantClient
# from qdrant_client.http import models as qdrant_models
# from qdrant_client.http.models import ContextQuery, ContextPair, SearchParams, SparseVectorParams
# from qdrant_client.http.exceptions import UnexpectedResponse
# from qdrant_client.http.models import Distance, VectorParams, PointStruct, Prefetch, FusionQuery, Fusion
# from sklearn.feature_extraction.text import TfidfVectorizer
#
# class QdrantClientManager:
# _instance = None
# _client = None
#
# @classmethod
# def get_client(cls, config):
# if cls._client is None:
# if config.rag.use_local:
# logging.info(f"Connecting to local Qdrant at: {config.rag.url}")
# cls._client = QdrantClient(url=config.rag.url) # Dùng URL, không phải path
# else:
# logging.info(f"Connecting to remote Qdrant at: {config.rag.url}")
# cls._client = QdrantClient(url=config.rag.url, api_key=config.rag.api_key)
#
# logging.info("Initialized Qdrant client singleton")
#
# return cls._client
#
#
# class QdrantRetriever:
#
# def __init__(self, config):
# self.logger = logging.getLogger(__name__)
# self.collection_name = config.rag.collection_name
# self.embedding_dim = config.rag.embedding_dim
# self.distance_metric = config.rag.distance_metric
# self.tfidf_vectorizer = TfidfVectorizer()
# self.sparse_vocab = {}
# self.is_fitted = False
#
# self.client = QdrantClientManager.get_client(config)
#
# self._ensure_collection()
#
# def _ensure_collection(self):
# collection_info = self.client.get_collections()
# print(f"Collection info: {collection_info}")
# collection_names = [collection.name for collection in collection_info.collections]
# print("Collection names:", collection_names)
#
# # self.client.delete_collection(collection_name=self.collection_name)
#
# if self.collection_name not in collection_names:
# self.logger.info(f"Creating new collection {self.collection_name}")
# try:
# self.client.create_collection(
# collection_name=self.collection_name,
# vectors_config={
# "dense": VectorParams(
# size=self.embedding_dim,
# distance=self.distance_metric, # =Distance.COSINE
# )
# },
# sparse_vectors_config={
# "sparse": SparseVectorParams(
# modifier=qdrant_models.Modifier.IDF
# )
# }
# # vectors_config=VectorParams(
# # size=self.embedding_dim,
# # distance=Distance.COSINE
# # ),
# # optimizers_config=qdrant_models.OptimizersConfigDiff(
# # indexing_threshold=10000,
# # ),
# )
# self.logger.info(f"Created new collection: {self.collection_name}")
# except Exception as e:
# self.logger.error(f"Error creating collection: {e}")
# raise e
# else:
# self.logger.info(f"Collection {self.collection_name} already exists")
#
# def upsert_chunks(self, chunks: List[Dict[str, Any]]):
# try:
# print(f"DEBUG Upserting chunks: {chunks}")
# points: List[PointStruct] = []
#
# for chunk in chunks:
# payload = chunk["metadata"].copy()
#
# payload["content"] = chunk["content"]
#
# print(f"DEBUG Upserting chunk: {chunk['id']}")
# print(f"DEBUG Embedding keys: {chunk['embedding']}")
# print(f"DEBUG Payload: {payload}")
#
# points.append(
# PointStruct(
# id=chunk["id"],
# # vector=chunk["embedding"],
# vector={
# "dense": chunk["embedding"]["dense"],
# "sparse": chunk["embedding"]["sparse"].as_object()
# },
# #Giả định chunk["embedding"] là dict kiểu:
# # {
# # "dense": [...], # list[float]
# # "sparse": {...} # dict[int -> float]
# # }
# payload=payload
# )
# )
#
# batch_size = 100
# # Nghĩa là mỗi lần gửi lên database sẽ gửi tối đa 100 points (vector + metadata).
# for i in range(0, len(points), batch_size):
# batch = points[i:i + batch_size]
# self.client.upsert(
# collection_name=self.collection_name,
# points=batch,
# wait=True
# )
# self.logger.info(f"Successfully upserted {len(chunks)} chunks into '{self.collection_name}'")
# except Exception as e:
# self.logger.error(f"Error upserting documents: {e}")
# raise
#
#
# def count_documents(self) -> int:
# try:
# collection_info = self.client.get_collection(self.collection_name)
# count = collection_info.vectors_count
# self.logger.info(f"Collection {self.collection_name} has {count} documents")
# return count
# except Exception as e:
# self.logger.error(f"Error getting collection info: {str(e)}")
# return 0
#
#
# def fit_sparse_vectorizer(self, corpus: List[str]):
# try:
# if not self.is_fitted:
# self.tfidf_vectorizer.fit(corpus)
# self.is_fitted = True
# self.logger.info("TF-IDF Vectorizer fitted successfully")
# else:
# self.logger.info("TF-IDF Vectorizer is already fitted")
# except Exception as e:
# self.logger.error(f"Error fitting TF-IDF Vectorizer: {e}")
#
#
# def text_to_sparse_vector(self, query_text: str) -> Dict[str, List]:
# if not self.is_fitted:
# self.logger.warning("TF-IDF Vectorizer is not fitted yet. Skipping transformation")
# return {}
#
# vec = self.tfidf_vectorizer.transform([query_text])
# indices = vec.indices.tolist()
# values = vec.data.tolist()
# return {
# "indices": indices,
# "values": values
# }
#
# def retrieve(
# self,
# query_embedding: Dict[str, Any],
# filters: Optional[Dict[str, Any]] = None,
# top_k: int = 5,
# include_metadata: bool = True,
# ) -> List[Dict[str, Any]]:
# """
# Perform hybrid retrieval using both dense and sparse embeddings.
#
# Args:
# query_embedding: Dict with keys "dense" (List[float]) and "sparse" (SparseEmbedding)
# filters: Optional metadata filters
# top_k: Number of documents to retrieve
# include_metadata: Whether to include payload in results
#
# Returns:
# List of retrieved documents with score
# """
# dense_vector = query_embedding.get("dense")
# sparse_vector = query_embedding.get("sparse")
#
# self.logger.info(f"""
# [RETRIEVE PARAMS]
#
# dense_dim={len(dense_vector) if dense_vector else 'None'},
# filters={filters},
# top_k={top_k},
# include_metadata={include_metadata}
# """)
#
# if sparse_vector:
# self.logger.info("Sparse vector detected → using hybrid search")
#
# prefetch = [
# qdrant_models.Prefetch(query=dense_vector, using="dense", limit=top_k),
# qdrant_models.Prefetch(
# query=qdrant_models.SparseVector(
# indices=sparse_vector.indices.tolist(),
# values=sparse_vector.values.tolist()
# ),
# using="sparse",
# limit=top_k,
# ),
# ]
#
# try:
# response = self.client.query_points(
# collection_name=self.collection_name,
# prefetch=prefetch,
# query=qdrant_models.FusionQuery(fusion=qdrant_models.Fusion.RRF),
# with_payload=include_metadata,
# query_filter=None, # TODO: build filter from `filters` if needed
# )
# retrieved_points = response.points
#
# results = []
# for point in retrieved_points:
# print("DEBUG point:", point)
#
# result_item = dict(point.payload or {})
# result_item["score"] = point.score
# results.append(result_item)
#
# self.logger.info(f"Retrieved {len(results)} results (hybrid search)")
# return results
#
# except Exception as e:
# self.logger.error(f"Error during hybrid retrieval: {e}", exc_info=True)
# return []
#
# else:
# self.logger.info("No sparse vector → fallback to dense-only search")
#
# try:
# response = self.client.query_points(
# collection_name=self.collection_name,
# prefetch=[
# qdrant_models.Prefetch(query=dense_vector, using="dense", limit=top_k)
# ],
# query=qdrant_models.FusionQuery(fusion=qdrant_models.Fusion.RRF),
# with_payload=include_metadata,
# query_filter=None,
# )
# retrieved_points = response.points
#
# results = []
# for point in retrieved_points:
# result_item = dict(point.payload or {})
# result_item["score"] = point.score
# results.append(result_item)
#
# self.logger.info(f"Retrieved {len(results)} results (dense-only search)")
# return results
#
# except Exception as e:
# self.logger.error(f"Error during dense-only retrieval: {e}", exc_info=True)
# return []
#
api/fastapi_server.py
View file @ 9e70c543
import os
import uuid
from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Request, Response
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import uvicorn
from agents.agent_decision import process_query
from typing import List, Union
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from typing import Dict, Optional, Union, List, TypedDict, Any, Literal
from typing import Dict, Optional, List, Any
from config import Config
from dataclasses import is_dataclass
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(name)s | %(levelname)s | %(message)s'
)
logger = logging.getLogger(__name__)
UPLOAD_FOLDER = "uploads/backend"
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
app = FastAPI(title="Transfer-Bot", version="1.0")
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")
def deserialize_messages(serialized: List[dict]) -> List[BaseMessage]:
message_objects = []
for msg in serialized:
print(f"Message deri: {msg}")
msg_type = msg.get("type")
content = msg.get("content", "")
if msg_type == "human":
message_objects.append(HumanMessage(content=content))
elif msg_type == "ai":
message_objects.append(AIMessage(content=content))
elif msg_type == "system":
message_objects.append(SystemMessage(content=content))
return message_objects
class QueryRequest(BaseModel):
query: str
conversation_history: List[dict] = []
@app.post("/chat")
def chat(request: QueryRequest,
response: Response,
request_obj: Request
):
session_id = request_obj.cookies.get("session_id", str(uuid.uuid4()))
try:
print(f"Session id: {session_id}")
history: List[BaseMessage] = deserialize_messages(request.conversation_history)
print(f"History: {history}")
print("Incoming request:", request.model_dump())
response_data: Dict[str, Any] = process_query(
query=request.query,
conversation_history=history
)
@asynccontextmanager
async def lifespan(server: FastAPI):
from rag.hybrid_retrieval import HybridRetrieval
app_config = Config()
server.state.config = app_config
server.state.default_retrieval = HybridRetrieval(app_config)
yield
print("response_data =", response_data)
response_text = response_data['response']
app = FastAPI(title="Transfer-Bot", version="1.0", lifespan=lifespan)
response.set_cookie(key="session_id", value=session_id)
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")
result = {
"response": response_text,
"agent": response_data.get("agent_name", "default_agent")
}
if "chunks" in response_data:
print(f"Chunks: {response_data['chunks']}")
result["context"] = {
"data_points": {
"text": response_data["chunks"]
}
}
class Message(BaseModel):
role: str
content: str
if "result_file" in response_data:
result["result_file_url"] = f"/uploads/{response_data['result_file']}"
print(f"Result final: {result}")
class QueryRequest(BaseModel):
messages: List[Message]
session_id: Optional[str] = None
context: Optional[dict] = {}
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/upload")
async def upload_file(response: Response, request_obj: Request, file: UploadFile = File(...), text: str = Form("")):
def deep_to_dict(obj):
if is_dataclass(obj):
return {k: deep_to_dict(v) for k, v in obj.__dict__.items()}
elif isinstance(obj, list):
return [deep_to_dict(i) for i in obj]
elif isinstance(obj, dict):
return {k: deep_to_dict(v) for k, v in obj.items()}
else:
return obj
file_path = os.path.join(UPLOAD_FOLDER, file.filename)
with open(file_path, "wb") as f:
f.write(await file.read())
session_id = request_obj.cookies.get("session_id", str(uuid.uuid4()))
@app.post("/chat")
async def chat(
request: QueryRequest,
response: Response,
fastapi_request: Request,
):
session_id: str = request.session_id or str(uuid.uuid4())
messages: List[Dict[str, Any]] = [msg.model_dump() for msg in request.messages]
logger.info(f"Received chat | session_id: {session_id} | message count: {len(messages)}")
print(f"type(messages) = {type(messages)}")
print(f"type(session_id) = {type(session_id)}")
print(f"messages = {messages}")
# [
# {
# 'role': 'user',
# 'content': 'Wer ist Herr Thomas Heine?'
# },
# {
# 'role': 'assistant',
# 'content': 'Herr Thomas Heine ist e'
# }
# ]
print(f"session_id = {session_id}") # 3f391a3a-f25b-4a96-9d6e-f63cc3eabb07
try:
query = {"text": text, "file_path": file_path}
response_data = process_query(query)
response_text = response_data['messages'][-1].content
retrieval = fastapi_request.app.state.default_retrieval
logger.info(f"Using retrieval class: {retrieval.__class__.__name__}")
result = await retrieval.run(messages, session_id)
response.set_cookie(key="session_id", value=session_id)
result = {
"response": response_text,
"agent": response_data.get("agent_name", "default_agent")
}
if "result_file" in response_data:
result["result_file_url"] = f"/uploads/{response_data['result_file']}"
try:
os.remove(file_path)
except Exception as e:
print(f"Failed to delete temp file: {str(e)}")
logger.info(f"Response returned for session_id: {session_id}")
result = deep_to_dict(result)
logger.info(f"Chat response (session_id={session_id}): {result}")
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
logger.error("Error in /chat endpoint", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000, log_config="logging_config.yaml"
)
uvicorn.run(app, host="0.0.0.0", port=8000)
chat_gui.py
View file @ 9e70c543
......@@ -10,8 +10,12 @@ if "chat_history" not in st.session_state:
st.session_state.chat_history = []
def get_serialized_history():
# return [
# {"type": "human" if role == "user" else "ai", "content": msg}
# for role, msg in st.session_state.chat_history
# ]
return [
{"type": "human" if role == "user" else "ai", "content": msg}
{"role": "user" if role == "user" else "assistant", "content": msg}
for role, msg in st.session_state.chat_history
]
......@@ -51,33 +55,59 @@ with tab1:
st.session_state.pending_query = user_query
st.rerun()
# if "pending_query" in st.session_state:
# with st.spinner("Wird verarbeitet..."):
# data = {}
# try:
# payload = {
# "query": st.session_state.pending_query,
# "conversation_history": get_serialized_history() # ✅ Thêm
# }
#
# print(f"payload: {payload}")
#
# response = requests.post(f"{API_URL}/chat", json=payload)
#
# data = response.json()
# bot_reply = data.get("response", "Keine Antwort erhalten.")
#
# # st.session_state.chat_history.append(("bot", bot_reply))
# if "messages" in data:
# st.session_state.chat_history = [
# ("user" if m["type"] == "human" else "bot", m["content"])
# for m in data["messages"]
# ]
# else:
# st.session_state.chat_history.append(("bot", bot_reply))
# except Exception as e:
# bot_reply = f"Error: {str(e)}"
# st.session_state.chat_history.append(("bot", bot_reply))
#
# del st.session_state.pending_query
# st.rerun()
if "pending_query" in st.session_state:
with st.spinner("Wird verarbeitet..."):
data = {}
try:
payload = {
"query": st.session_state.pending_query,
"conversation_history": get_serialized_history() # ✅ Thêm
"messages": get_serialized_history(),
"session_id": st.session_state.get("session_id", None),
}
print(f"payload: {payload}")
response = requests.post(f"{API_URL}/chat", json=payload)
data = response.json()
bot_reply = data.get("response", "Keine Antwort erhalten.")
# st.session_state.chat_history.append(("bot", bot_reply))
if "messages" in data:
st.session_state.chat_history = [
("user" if m["type"] == "human" else "bot", m["content"])
for m in data["messages"]
]
else:
st.session_state.chat_history.append(("bot", bot_reply))
bot_reply = data.get("message", {}).get("content", "Keine Antwort erhalten.")
# Nếu muốn lấy lại full history từ server
# if "messages" in data:
# st.session_state.chat_history = [
# (m["role"], m["content"])
# for m in data["messages"]
# ]
# else:
st.session_state.chat_history.append(("assistant", bot_reply))
except Exception as e:
bot_reply = f"Error: {str(e)}"
st.session_state.chat_history.append(("bot", bot_reply))
st.session_state.chat_history.append(("assistant", bot_reply))
del st.session_state.pending_query
st.rerun()
......
config.py
View file @ 9e70c543
import os
from dotenv import load_dotenv
from fastembed import SparseTextEmbedding
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from openai import OpenAI
from agents.rag_agent.local_embedding_model import LocalEmbeddingModel
from dotenv import load_dotenv
load_dotenv()
class AgentDecisoinConfig:
DEFAULTS = {
"VECTORSTORE_LOCAL_PATH": "./data/qdrant_db",
"DOCUMENT_DB_LOCAL_DIR": "./data/document_db",
"PARSED_CONTENT_DIR": "./data/parsed_documents",
"QDRANT_COLLECTION_NAME": "pro_assistant_rag",
"RETRIEVAL_TOP_K": 5,
"RERANKER_MODEL": "cross-encoder/ms-marco-TinyBERT-L-6",
"RERANKER_TOP_K": 3,
"EMBEDDING_PROVIDER": "openai",
"EMBEDDING_MODEL_NAME": "text-embedding-ada-002",
"EMBEDDING_DIM": 1536,
"SEMANTIC_CHUNKING_MODEL_NAME": "paraphrase-multilingual-MiniLM-L12-v2",
"API_HOST": "0.0.0.0",
"API_PORT": 8000,
"API_DEBUG": True,
"API_RATE_LIMIT": 10,
}
def getenv(name, default=None, type_cast=None):
value = os.getenv(name, default)
if type_cast and value is not None:
return type_cast(value)
return value
class BaseConfig:
def __init__(self):
self.vectorstore_local_path = getenv("VECTORSTORE_LOCAL_PATH", DEFAULTS["VECTORSTORE_LOCAL_PATH"])
self.document_db_local_dir = getenv("DOCUMENT_DB_LOCAL_DIR", DEFAULTS["DOCUMENT_DB_LOCAL_DIR"])
self.parsed_content_dir = getenv("PARSED_CONTENT_DIR", DEFAULTS["PARSED_CONTENT_DIR"])
self.collection_name = getenv("QDRANT_COLLECTION_NAME", DEFAULTS["QDRANT_COLLECTION_NAME"])
self.qdrant_url = getenv("QDRANT_URL", "")
self.qdrant_api_key = getenv("QDRANT_API_KEY", "")
self.retrieval_top_k = getenv("RETRIEVAL_TOP_K", DEFAULTS["RETRIEVAL_TOP_K"], int)
self.embedding_provider = getenv("EMBEDDING_PROVIDER", DEFAULTS["EMBEDDING_PROVIDER"])
self.embedding_model_name = getenv("EMBEDDING_MODEL_NAME", DEFAULTS["EMBEDDING_MODEL_NAME"])
self.embedding_dim = getenv("EMBEDDING_DIM", DEFAULTS["EMBEDDING_DIM"], int)
self.openai_api_key = getenv("OPENAI_API_KEY", "")
self.huggingface_api_token = getenv("HUGGINGFACE_TOKEN", "")
class RetrievalConfig(BaseConfig):
def __init__(self):
super().__init__()
self.llm = ChatOpenAI(
model=os.getenv("model_name"),
api_key=os.getenv("openai_api_key"),
model=getenv("CHAT_MODEL_NAME"),
api_key=self.openai_api_key,
temperature=0.3
)
self.reranker_model = getenv("RERANKER_MODEL", DEFAULTS["RERANKER_MODEL"])
self.reranker_top_k = getenv("RERANKER_TOP_K", DEFAULTS["RERANKER_TOP_K"], int)
class RAGConfig:
class FileIngestorConfig(BaseConfig):
def __init__(self):
self.vector_db_type = "qdrant"
self.embedding_dim = 1536
self.distance_metric = "Cosine"
self.use_local = True
self.vector_local_path = "./data/qdrant_db" # Add this with a default value
self.doc_local_path = "./data/docs_db"
self.parsed_content_dir = "./data/parsed_docs"
self.url = os.getenv("QDRANT_URL")
self.api_key = os.getenv("QDRANT_API_KEY")
self.collection_name = "pro_assistant_rag"
self.chunk_size = 512
self.chunk_overlap = 50
self.processed_docs_dir = "./data/processed"
self.embedding_model = OpenAIEmbeddings(
model=os.getenv("embedding_model_name"),
api_key=os.getenv("openai_api_key")
)
self.embedding_for_chunk_model = LocalEmbeddingModel('paraphrase-multilingual-MiniLM-L12-v2')
super().__init__()
self.summarizer_model = ChatOpenAI(
model=os.getenv("model_name"),
api_key=os.getenv("openai_api_key"),
model=getenv("CHAT_MODEL_NAME"),
api_key=self.openai_api_key,
temperature=0.5
)
self.chunker_model = ChatOpenAI(
model=os.getenv("model_name"),
api_key=os.getenv("openai_api_key"),
temperature=0.0
)
self.llm = ChatOpenAI(
model=os.getenv("model_name"),
api_key=os.getenv("openai_api_key"),
temperature=0.3
)
self.sparse_embedder = SparseTextEmbedding("Qdrant/bm25")
self.max_queries = 3
self.top_k = 5
self.similarity_threshold = 0.75
self.huggingface_token = os.getenv("HUGGINGFACE_TOKEN")
self.vector_search_type = 'similarity' # or 'mmr'
self.reranker_model = "cross-encoder/ms-marco-TinyBERT-L-6"
self.reranker_top_k = 3
# self.chunking_strategy = "semantic"
self.max_context_length = 8192
self.include_sources = True
self.response_format_instructions = """Instructions:
1. Answer the query based ONLY on the information provided in the context.
2. If the context doesn't contain relevant information to answer the query, state: "I don't have enough information to answer this question based on the provided context."
3. Do not use prior knowledge not contained in the context.
5. Be concise and accurate.
6. Provide a well-structured response based on retrieved knowledge."""
self.min_retrieval_confidence = 0.8 # the auto routing from RAG agent to WEB_SEARCH agent is dependent on this value
self.context_limit = 20 # include last 20 messsages (10 Q&A pairs) in history
self.semantic_chunking_model_name = getenv("SEMANTIC_CHUNKING_MODEL_NAME",
DEFAULTS["SEMANTIC_CHUNKING_MODEL_NAME"])
self.local_model_path = getenv("LOCAL_EMBEDDING_MODEL", "")
class APIConfig:
def __init__(self):
self.host = "0.0.0.0"
self.port = 8000
self.debug = True
self.rate_limit = 10
self.max_image_upload_size = 5 # max upload size in MB
self.host = getenv("API_HOST", DEFAULTS["API_HOST"])
self.port = getenv("API_PORT", DEFAULTS["API_PORT"], int)
self.debug = getenv("API_DEBUG", DEFAULTS["API_DEBUG"])
self.rate_limit = getenv("API_RATE_LIMIT", DEFAULTS["API_RATE_LIMIT"], int)
class Config:
def __init__(self):
self.agent_decision = AgentDecisoinConfig()
self.rag = RAGConfig()
self.rag = RetrievalConfig()
self.api = APIConfig()
self.max_conversation_history = 40
# config = Config()
\ No newline at end of file
self.file_ingestor = FileIngestorConfig()
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Dies ist die Gitlab-Instanz des Transferportals der Hochschule für Technik Stuttgart. Hier geht es zurück zum Portal