AI Agent Data Ingestion: Connecting to Your Data Sources

AI Agent Data Ingestion: Connecting to Your Data Sources

💡 What You Will Learn

AI Agent Data Ingestion: Connecting to Your Data Sources

from langchain_community.document_loaders import (
    TextLoader, PyPDFLoader, CSVLoader, WebBaseLoader
)

def load_documents(source_type: str, source_path: str):
    """"""
    loaders = {
        'txt': TextLoader(source_path, encoding='utf-8'),
        'pdf': PyPDFLoader(source_path),
        'csv': CSVLoader(source_path),
        'web': WebBaseLoader(source_path),
    }
    loader = loaders.get(source_type)
    if not loader:
        raise ValueError(f"Data source: {source_type}")
    return loader.load()

|:----|:-----|:----|:--------|:-----| || 512 tokens | 64 || 78% | || 1000 chars | 200 || 85% | ||| 0 || 88% |

from langchain_chroma import Chroma
from langchain_huggingface import HuggingFaceEmbeddings

# 
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 
vector_store = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# RetrieveTesting
results = vector_store.similarity_search_with_score("", k=3)
for doc, score in results:
    print(f": {score:.4f} | : {doc.page_content[:100]}")

def rag_query(query: str, k: int = 3) -> str:
    """RAGRetrieve+Generate"""
    docs = vector_store.similarity_search(query, k=k)
    context = "\n---\n".join([d.page_content for d in docs])

    prompt = f"""""


{context}

{query}
"""
    return llm.invoke(prompt)