Agentic LightRAG functionality

[nirmitktripathii]

Me and my team are currently working on a project related to creating a Legal Agent that can perform Legal research, Extractive and Abstractive Case Summarization, Document and Contract analysis and drafting and Legal Compliance Requirement checker. We are utilizing the use of Knowledge Graphs to perform LightRAG implementation to perform our tasks.

Link to LightRAG github: https://github.com/HKUDS/LightRAG

Will there be agentic LightRAG/GraphRAG implementation functionality added in the future?

[dirkbrnd]

Hi!
Not on our immediate roadmap, but I have added it to the wishlist! GraphRAG is really powerful and it is on our radar.

[nirmitktripathii]

Noted. Thanks for the update 😁

[edudesouza]

Hi there, I have a RAG implementation using Elastic as a Vector Database.
Is quite simple an effective, I have a function calles docs that query some Elastic indexes

[nirmitktripathii]

Thanks for the reply. Can you guide me how you implemented the rag implementation using elastic vector DB?

[edudesouza]

1. Embeddings

`import os
import warnings

from dotenv import load_dotenv
from typing import List
from elasticsearch import Elasticsearch
from langchain.text_splitter import CharacterTextSplitter, RecursiveCharacterTextSplitter
from langchain_text_splitters.base import Language

import pymupdf4llm
import fitz

import openai

load_dotenv()

warnings.filterwarnings('ignore')

index_name = "a1_convencao"
caminho_pdf = "Nouveaux_Convencao_2015.pdf"

OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')

ELASTICSEARCH_HOST = os.getenv('ELASTICSEARCH_HOST')
ELASTICSEARCH_USER = os.getenv('ELASTICSEARCH_USER')
ELASTICSEARCH_PASS = os.getenv('ELASTICSEARCH_PASS')
ELASTICSEARCH_PORT = 9200

elastic_client = Elasticsearch(
ELASTICSEARCH_HOST,
basic_auth=(ELASTICSEARCH_USER, ELASTICSEARCH_PASS),
verify_certs=False
)

def criar_indice():

index_body = {
    "mappings": {
        "properties": {
            "texto": {"type": "text"},
            "embedding": {
                "type": "dense_vector",
                "dims": 1536,
                "index": True,
                "similarity": "cosine"
            },
            "chunk_id": {"type": "integer"}
        }
    },
    "settings": {
        "number_of_shards": 1,
        "number_of_replicas": 1
    }
}

if not elastic_client.indices.exists(index=index_name):
    elastic_client.indices.create(index=index_name, body=index_body)

def criar_embedding(texto: str) -> List[float]:

client = openai.OpenAI(api_key=OPENAI_API_KEY)
response = client.embeddings.create(
    input=texto,
    model="text-embedding-ada-002"
)

return response.data[0].embedding

c_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200, separator=" ")
r_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200, separators=["\n\n", "\n", " ", ""])
t_splitter = RecursiveCharacterTextSplitter.from_language(language=Language.MARKDOWN,chunk_size=1000,chunk_overlap=100)

full_text = pymupdf4llm.to_markdown(caminho_pdf)

chunks = t_splitter.split_text(full_text)
#print("Chunks: ", chunks)
print("Length of chunks: ", len(chunks))

'''
chunks = c_splitter.split_text(full_text)
#print("Chunks: ", chunks)
print("Length of chunks: ", len(chunks))
'''

criar_indice()

for i, chunk in enumerate(chunks):

embedding = criar_embedding(chunk)

documento = {
    'texto': chunk,
    'embedding': embedding,
    'chunk_id': i
}

elastic_client.index(
    index=index_name,
    document=documento,
    id=str(i)
)

print(f"Chunk {i + 1}/{len(chunks)} processado e armazenado")

print("-> finalizado")
`

[edudesouza]

2. RAG
   `import os
   import warnings
   import base64
   import time

from dotenv import load_dotenv
from typing import List, Dict
from elasticsearch import Elasticsearch
from langchain.text_splitter import CharacterTextSplitter, RecursiveCharacterTextSplitter

from anthropic import Anthropic
from google import genai
from google.genai import types
from together import Together

import openai

load_dotenv()

warnings.filterwarnings('ignore')

#index_name = "a1_convencao"
#query = "Qual a fração ideal do apartamento 64 B?"
#modelo = "o3-mini"

OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
CLAUDE_API_KEY = os.getenv('ANTHROPIC_API_KEY')
GEMINI_API_KEY = os.getenv('GEMINI_API_KEY')
TOGETHER_API_KEY = os.getenv('TOGETHER_API_KEY')

ELASTICSEARCH_HOST = os.getenv('ELASTICSEARCH_HOST')
ELASTICSEARCH_USER = os.getenv('ELASTICSEARCH_USER')
ELASTICSEARCH_PASS = os.getenv('ELASTICSEARCH_PASS')
ELASTICSEARCH_PORT = 9200

elastic_client = Elasticsearch( ELASTICSEARCH_HOST,basic_auth=(ELASTICSEARCH_USER, ELASTICSEARCH_PASS),verify_certs=False)

anthropic = Anthropic(api_key=CLAUDE_API_KEY)
client = openai.OpenAI(api_key=OPENAI_API_KEY)

def buscar_documentos_similares(index_name,query_embedding: List[float], top_k: int = 10) -> List[Dict]:

client = elastic_client

script_query = {
    "size": top_k,
    "_source": ["texto", "metadata"],
    "query": {
        "script_score": {
            "query": {"match_all": {}},
            "script": {
                "source": "cosineSimilarity(params.query_vector, 'embedding') + 1.0",
                "params": {"query_vector": query_embedding}
            }
        }
    }
}

try:
    response = client.search(
        index=index_name,
        body=script_query
    )
    return response['hits']['hits']

except Exception as e:
    print(f"Erro na busca: {str(e)}")
    print(f"Query utilizada: {script_query}")
    return []


context = "\n".join([doc['_source']['texto'] for doc in documentos_similares])

prompt = f"""Você é um assistente especialista em condomínios.
Use o contexto abaixo para responder à pergunta do usuário.
Se a resposta não puder ser encontrada no contexto, diga que não pode responder com base nas informações disponíveis.

Contexto:
{context}

Pergunta: {pergunta}

Resposta:"""

LLM -----------------------------------------------------------------

def resp_claude(query,index_name,modelo):

print('-> claude:',query,index_name,modelo)
inicio = time.time()

response             = client.embeddings.create( input=query,model="text-embedding-ada-002" )
documentos_similares = buscar_documentos_similares(index_name,response.data[0].embedding)

context = "\n".join([doc['_source']['texto'] for doc in documentos_similares])

print(f"-> Documentos similares: {len(documentos_similares)}")

prompt = f"""Use o contexto abaixo para responder à pergunta do usuário.
Fique atendo às numerações e denominações das unidades e apartamentos: Torre Bloco Quadra Lote Rua, muitas vezes as pessoa podem falar 42 A e 
isso pode significar 42 Bloco A, 42 Torre 1, então analise o contexto fornecido para fazer esse tipo de associação.

Se a resposta não puder ser encontrada no contexto, diga que não pode responder com base nas informações disponíveis.
Caso encontre a resposta, seja objetivo, cite a página e o trecho onde a resposta foi encontrada.

Contexto:
{context}

Pergunta: {query}

Resposta:
"""

message = anthropic.messages.create(        
    model=modelo,        
    max_tokens=500,
    temperature=0.3,
    system="Você é um assistente especialista em condomínios. Responda sempre em português e seja o mais claro possível e objetivo",
    messages=[
        {"role": "user", "content": prompt}
    ]
)

print('-> finalizado')

fim = time.time()
tempo_execucao = fim - inicio
print(f"-> execução: {tempo_execucao:.2f} s")

return {"msg":message.content[0].text,"tempo":f"{tempo_execucao:.2f}s"}

def resp_gemini(query,index_name,modelo):

print('-> gemini',query,index_name,modelo)
inicio = time.time()

response        = client.embeddings.create( input=query,model="text-embedding-ada-002" )
query_embedding = response.data[0].embedding

documentos_similares = buscar_documentos_similares(index_name,query_embedding)
documentos_similares:List[Dict]

context = "\n".join([doc['_source']['texto'] for doc in documentos_similares])

prompt = f"""Use o contexto abaixo para responder à pergunta do usuário.
Fique atendo às numerações e denominações das unidades e apartamentos: Torre Bloco Quadra Lote Rua, muitas vezes as pessoa podem falar 42 A e 
isso pode significar 42 Bloco A, 42 Torre 1, então analise o contexto fornecido para fazer esse tipo de associação.

Se a resposta não puder ser encontrada no contexto, diga que não pode responder com base nas informações disponíveis.
Caso encontre a resposta, seja objetivo, cite a página e o trecho onde a resposta foi encontrada.

Contexto:
{context}

Pergunta: {query}

Resposta:"""

gclient = genai.Client(api_key=GEMINI_API_KEY)
model   = modelo

contents = [
    types.Content(
        role="user",
        parts=[
            types.Part.from_text(text=prompt)
        ]
    )
]

generate_content_config = types.GenerateContentConfig(
    temperature = 0.3,
    top_p = 0.95,
    max_output_tokens = 1200,
    response_modalities = ["TEXT"]        
)

response = gclient.models.generate_content_stream(model = model, contents = contents, config = generate_content_config)

resposta_completa = ""
for item in response:
    resposta_completa += item.text

print('-> finalizado')
fim = time.time()
tempo_execucao = fim - inicio
print(f"-> execução: {tempo_execucao:.2f} s")

return {"msg":resposta_completa,"tempo":f"{tempo_execucao:.2f}s"}

def resp_openai(query,index_name,modelo):

print('-> openai',query,index_name,modelo)

response        = client.embeddings.create( input=query,model="text-embedding-ada-002" )
query_embedding = response.data[0].embedding

documentos_similares = buscar_documentos_similares(index_name,query_embedding)
documentos_similares:List[Dict]

context = "\n".join([doc['_source']['texto'] for doc in documentos_similares])

system_prompt = f"""Use o contexto abaixo para responder à pergunta do usuário.
Fique atendo às numerações e denominações das unidades e apartamentos: Torre Bloco Quadra Lote Rua, muitas vezes as pessoa podem falar 42 A e 
isso pode significar 42 Bloco A, 42 Torre 1, então analise o contexto fornecido para fazer esse tipo de associação.

Se a resposta não puder ser encontrada no contexto, diga que não pode responder com base nas informações disponíveis.
Caso encontre a resposta, seja objetivo, cite a página e o trecho onde a resposta foi encontrada.

Context:
{context}
"""
messages = ({"role": "user", "content": query})
messages_with_context = [{"role": "system", "content": system_prompt}, messages]

response = client.chat.completions.create(
    model=modelo,
    messages=messages_with_context,
    temperature=0.3
)

return response.choices[0].message.content

def resp_openai_r(query,index_name,modelo):

print('-> openai r:',query,index_name,modelo)
inicio = time.time()

response             = client.embeddings.create( input=query,model="text-embedding-ada-002" )
documentos_similares = buscar_documentos_similares(index_name,response.data[0].embedding)

context = "\n".join([doc['_source']['texto'] for doc in documentos_similares])

print(f"-> Documentos similares: {len(documentos_similares)}")
 
system_prompt = f"""Você é um assistente especialista em condomínios.
Use o contexto abaixo para responder à pergunta do usuário.
Fique atendo às numerações e denominações das unidades e apartamentos: Torre Bloco Quadra Lote Rua, muitas vezes as pessoa podem falar 42 A e 
isso pode significar 42 Bloco A, 42 Torre 1, então analise o contexto fornecido para fazer esse tipo de associação.

Se a resposta não puder ser encontrada no contexto, diga que não pode responder com base nas informações disponíveis.
Caso encontre a resposta, seja objetivo, cite a página e o trecho onde a resposta foi encontrada.

Context:
{context}

Pergunta: 
{query}
"""

response = client.chat.completions.create(
    model=modelo,
    messages=[{"role": "user", "content": system_prompt}],
)

print('-> finalizado')
fim = time.time()
tempo_execucao = fim - inicio
print(f"-> execução: {tempo_execucao:.2f} s")

return {"msg":response.choices[0].message.content,"tempo":f"{tempo_execucao:.2f}s"}

def resp_deepseek(query,index_name,modelo):

print('-> deepseek',query,index_name,modelo)

context = "\n".join([doc['_source']['texto'] for doc in documentos_similares])

system_prompt = f"""Você é um assistente especialista em condomínios.
Use o contexto abaixo para responder à pergunta do usuário.
Se a resposta não puder ser encontrada no contexto, diga que não pode responder com base nas informações disponíveis.
Caso encontre a resposta, cite a página e o trecho onde a resposta foi encontrada.

Context:
{context}

Pergunta: {query}
"""
messages_with_context = [{"role": "user", "content": system_prompt}]

client = Together(api_key=os.getenv("TOGETHER_API_KEY"))

response = client.chat.completions.create(
    model=modelo,
    messages=messages_with_context,
    max_tokens=500,
    temperature=0.3,
    top_p=0.7,
    top_k=50,
    repetition_penalty=1,
    stop=["< | end▁of▁sentence | >"],
    stream=False
)

return response.choices[0].message.content

def resp_Llama(query,index_name,modelo):

print('-> llama',query,index_name,modelo)

context = "\n".join([doc['_source']['texto'] for doc in documentos_similares])

system_prompt = f"""Você é um assistente especialista em condomínios.
Use o contexto abaixo para responder à pergunta do usuário.
Se a resposta não puder ser encontrada no contexto, diga que não pode responder com base nas informações disponíveis.
Caso encontre a resposta, cite a página e o trecho onde a resposta foi encontrada.

Context:
{context}

Pergunta: {query}
"""
messages_with_context = [{"role": "user", "content": system_prompt}]

client = Together(api_key=os.getenv("TOGETHER_API_KEY"))

response = client.chat.completions.create(
    model=modelo,
    messages=messages_with_context,
    max_tokens=500,
    temperature=0.3,
    top_p=0.7,
    top_k=50,
    repetition_penalty=1,
    stop=["<|eot_id|>","<|eom_id|>"],
    stream=False
)

return response.choices[0].message.content

claude-3-7-sonnet-20250219 claude-3-5-sonnet-20241022

gemini-1.5-flash-001 gemini-2.0-flash gemini-2.0-flash-thinking-exp-01-21 gemini-2.0-pro-exp-02-05

gpt-4o-mini

o1 o1-mini o3-mini

#print( f'Claude:\n{resp_claude(query,docs_similares,"claude-3-5-sonnet-20241022")}\n----------------------------------------------' )

#print( f'Gemini:\n{resp_gemini(query,docs_similares,"gemini-2.0-flash")}\n----------------------------------------------' )

#print( f'ChatGPT:\n{resp_openai(query,docs_similares,"gpt-4o-mini")}\n----------------------------------------------' )

#print( f'ChatGPT o:\n{resp_openai_r(query,docs_similares,"o3-mini")}\n----------------------------------------------' )

#print( f'DeepSeek:\n{resp_deepseek(query,docs_similares,"deepseek-ai/DeepSeek-R1-Distill-Llama-70B-free")}\n----------------------------------------------' )
#print( f'resp_Llama:\n{resp_Llama(query,docs_similares,"meta-llama/Llama-3.3-70B-Instruct-Turbo-Free")}\n' )`

[edudesouza]

3. Tool
   `def buscar_docs():

   '''
   rag.query = request.args.get('query')
   rag.modelo = "o3-mini"
   rag.index_name = "a1_convencao"
   result = rag.resp_openai_r(rag.query,rag.index_name,rag.modelo)
   '''

   rag.query = request.args.get('query')
   rag.modelo = "claude-3-5-sonnet-20241022"
   rag.index_name = "a1_convencao"
   result = rag.resp_claude(rag.query,rag.index_name,rag.modelo)

   '''
   rag.query = request.args.get('query')
   rag.modelo = "gemini-1.5-flash-8b"
   rag.index_name = "a1_convencao"
   result = rag.resp_gemini(rag.query,rag.index_name,rag.modelo)
   '''

   #return jsonify({"resposta":result['msg'],"query":rag.query,"modelo":rag.modelo,"tempo":result['tempo']})
   return json.dumps(result['msg'])`