src/generation/rag.py

from langchain_ollama import ChatOllama
# pyrefly: ignore [missing-import]
from config.settings import SETTINGS
from src.retrieval.retrieve_tool import Retrieve_Tool
from src.generation.llm_client import generate
from src.embedding.cache import SemanticCache
from src.api.schemas import RetrievalInput
from langchain_groq import ChatGroq
import httpx
from src.generation.tools import search_tool
from langfuse import observe, propagate_attributes, get_client

langfuse = get_client()
def is_ollama_available(base_url="http://localhost:11434"):
    try:
        response = httpx.get(f"{base_url}/api/tags", timeout=2.0)
        print("Ollama is available")
        return response.status_code == 200
    except Exception:
        print("Ollama is not available")
        return False

class Rag():
    def __init__(self):
        if is_ollama_available():
            self.llm = ChatOllama(
                model="llama3.2",
                base_url="http://localhost:11434",
                temperature=0.5,
            )
        else:
            self.llm = ChatGroq(
                model="llama-3.1-8b-instant",
                api_key=SETTINGS.API_KEY.get_secret_value(),
                temperature=0.5,
            ) 
        self.retrieve = Retrieve_Tool()
        self.search_tool = search_tool(self.retrieve)
        self.llm_with_tools = self.llm.bind_tools([self.search_tool])
        self.response_cache = SemanticCache(
            embeddings=self.retrieve.embeddings,
            key_prefix="rag:response:",
        )
        
    @observe(name="RAG Systems")
    async def get_sse_response(self, query: RetrievalInput):
        # Thiết lập session và user cho Trace hiện tại
        with propagate_attributes(
            session_id=query.session_id,
            user_id=query.user_id
        ):        
            result = self.response_cache.get(query.user_input)
        if result is not None:
            yield result
            return
 
        full_response = ""
        async for chunk in generate(self.llm_with_tools, query, self.search_tool):
            full_response += chunk
            yield f"{chunk} "

        if full_response:
            self.response_cache.set(query.user_input, full_response)
            langfuse.set_current_trace_io(input={query.user_input}, output={full_response})