Agentenbasierte KI mit LangChain Teil 1

Dieser Leitfaden zeigt Schritt für Schritt den Weg von Rohstoffen bis zu einem funktionsfähigen System für: Agentic AI mit LangChain – Teil 1: Aufbau einer semantischen RAG-Pipeline. Der Schwerpunkt liegt auf ausführbaren Schritten, expliziten Überprüfungen sowie Code, den Sie ohne Rätseln über die Absicht direkt in ein Repository einfügen können. Zur Übersicht sollten Sie vor dem Ändern des Codes die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

Technologiestack

Beim Erarbeiten des Tech Stacks sollte man zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie was bei einem teilweisen Versagen geschieht. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Man sollte kleine, testbare Einheiten vor umfangreichen Skripten bevorzugen. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Prompts anpassen. Eine häufige Änderung der Prompts behebt selten ein schwaches Suchverhalten.

1. Systemkonfigurationen

Beim Arbeiten an „1. Systemkonfigurationen“ sollten Sie zunächst einen Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Abläufe ab. Messen Sie die Genauigkeit anhand eines festgelegten Fragebogens, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Suchverhalten.

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. Projektstruktur

Beim Arbeiten an Punkt 2 „Projektstruktur“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Notieren Sie die Laufzeiten sowie die Kosten für Token oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo in gemeinsame Umgebungen wechselt. Messen Sie die Trefferquote anhand eines festgelegten Fragekatalogs, bevor Sie die Anfragenanweisungen anpassen. Häufige Änderungen der Anfragenanweisungen beheben selten ein schwaches Suchverhalten.

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── chains.py           # llm chains
├── nodes.py            # python functions utlizing llm chains
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

Beim Arbeiten an Punkt 2 „Projektstruktur“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgszeichen sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf und den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

2.1. LLM-Ketten

2.1. LLM-Ketten funktionieren am besten, wenn sie als messbare Struktur betrachtet werden. Erfassen Sie ein gelungenes Beispiel, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Ziehen Sie kleine, testbare Einheiten vor großen, komplexen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf einen verworrenen Ablauf. Legen Sie Budgets für Tokens pro Turnus und pro Sitzung fest. Agentenbasierte Tools erweitern den Kontext stark; feste Obergrenzen verhindern, dass Demonstrationen zu unerwarteten Rechnungen werden.

2.2. Symantic RAG

2.2. Symantic RAG funktioniert am besten, wenn es als messbare Oberfläche betrachtet wird. Erfassen Sie ein gelungenes Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Ergebnisse ab. Trennen Sie die Chunking-Strategie von der Abrufstrategie. Ein Änderung in einer sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern.

2.3. FastAPI Backend

2.3. Das FastAPI-Backend funktioniert am besten, wenn es als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Erfassen Sie außerdem die Laufzeiten sowie die Kosten für Tokens oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn sich der Einsatzbereich von einer Demo-Umgebung in gemeinsam genutzte Umgebungen verschiebt. Trennen Sie die Strategie zur Aufteilung in Blöcke von der Strategie zum Abrufen von Daten. Ein Änderungsbedarf bei einer dieser Strategien sollte nicht dazu führen, dass die andere neu geschrieben werden muss, wenn sich die Qualitätsmetriken ändern. 2.3. Das FastAPI-Backend funktioniert am besten, wenn es als messbarer Bereich betrachtet wird. Erfassen Sie vor der Erweiterung des Umfangs ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Dokumentieren Sie den erfolgreichen Ablauf sowie den Wiederherstellungsprozess gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen.

2.4. Streamlit-Klient

Für den Streamlit-Klienten 2.4 sollten Eingabefelder, der Verantwortliche für einen Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zuständen schließen zu müssen. Vorzuziehen sind kleine, testbare Einheiten statt umfangreicher Skripte. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortungsbereich hinweisen und nicht auf ein verworrenes Pipeline-System. Zitieren Sie die Passagen, die tatsächlich die Antwort untermauern. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden.

3. LLM-Ketten mit LCEL

Für 3. LLM-Ketten mit LCEL sollten die Eingaben, der Verantwortliche für den jeweiligen Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab. Wählen Sie bei dem nächsten Schritt Code oder einen Toolaufruf strukturierte Ausgaben mit Schema-Validierung vor freiformiger Prosa.

3.1. LangChain Expression Language (LCEL)

Für 3.1. LangChain Expression Language (LCEL) sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erhalten Sie Aufzeichnungen der Laufzeiten sowie der Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung wechselt. Zitieren Sie die Passagen, die tatsächlich die Grundlage für die Antwort bilden. Ohne Zitate können die Operator nicht zwischen Halluzinationen und Lücken in der Indizierung unterscheiden. Für 3.1. LangChain Expression Language (LCEL) sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Notfallweg gemeinsam. Wiederholungsversuche, menschliche Überprüfungen sowie die Handhabung von Fehlern gehören zum Produkt und nicht zu späten Korrekturen.

r Polnisch.

class GradeDocuments(BaseModel):
    """Binary score for relevance check on retrieved documents."""

    binary_score: bool = Field(
        description="Documents are relevant to the question, 'yes' or 'no'"
    )


async def retrieval_grader(document: Document, question: str):
    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini", temperature=0)
    structured_llm_grader = llm.with_structured_output(GradeDocuments)

    system = """You are a grader assessing relevance of a retrieved document to a user question. \n
    If the document contains keyword(s) or semantic meaning related to the question, grade it as relevant.\n
    Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question. \n
    if the document is relevant, give it a score 'yes', otherwise give it a score 'no'
    """

    grade_prompt = ChatPromptTemplate.from_messages(
        [
            ("system", system),
            ("human", "Retrieved document: \n\n {document} User question: {question}"),

        ]
    )

    retrieval_grader = grade_prompt | structured_llm_grader
    grade = await retrieval_grader.ainvoke({"document": document, "question": question})

    return grade.binary_score

4. Semantisches RAG mit PineCone

4.1. Einpflegeworkflow

4.2. Abrufworkflow

4.3. VectorService-Klasse

4.3.1. Initialisierung

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_pinecone import PineconeVectorStore

class VectorService:
    def __init__(self, index_name: str = "knowledgebase"):
        self.index_name = index_name

        # Embedding model
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small", api_key=api_key)

        # Pinecone vectorstore wrapper
        self.vectorstore = PineconeVectorStore(
            index_name=index_name,
            embedding=self.embeddings,
            pinecone_api_key=pinecone_key,
        )

        # Retriever
        self.retriever = self.vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 3}
        )

        # Text splitter (LangChain-native)
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=50,
            separators=["\n\n", "\n", ".", " ", ""]
        )

4.3.2. Speichern von Dateinhalten

    async def store_file_content_in_db(self, filepath: str) -> None:
        """
        Read the uploaded text file, split the file, embed chunks and save chunks to vector database
        :param filepath: filepath to load to vectordb
        :return: None
        """
        logger.debug(f"Loading file: {filepath}")

        # Load entire file (safe for most RAG use cases)
        async with aiofiles.open(filepath, "r", encoding="utf-8") as f:
            raw_text = await f.read()

        # Minimal cleaning (preserve structure)
        raw_text = raw_text.replace("\r", "")

        # Split into LangChain Documents
        docs = self.splitter.create_documents(
            [raw_text],
            metadatas=[{"source": os.path.basename(filepath)}]
        )

        logger.debug(f"Storing {len(docs)} chunks into Pinecone")

        # Ingest into Pinecone
        PineconeVectorStore.from_documents(
            documents=docs,
            embedding=self.embeddings,
            index_name=self.index_name,
            pinecone_api_key=pinecone_key,
        )

4.3.3. Dokumente suchen

    async def search_documents(self, query: str) -> list[Document]:
        logger.debug(f"Searching for: {query}")
        return self.retriever.invoke(query)

5. FastAPI-Backend

5.1. Datei-Uploader

async def save_file(file: UploadFile) -> str:
    """
    accept a UploadFile from FastAPI and save the file in specified location
    :param file: UploadFile specified by users from fastAPI interface
    :return: path of the saved file
    """
    await makedirs("uploads", exist_ok=True)
    filepath = os.path.join("uploads", file.filename)
    async with aiofiles.open(filepath, "wb") as f:
        while chunk := await file.read(DEFAULT_CHUNK_SIZE):
            await f.write(chunk)
    return filepath

5.2. Anfrage/Antwort-Modelle

class RAGRequest(BaseModel):
    prompt: str

class RAGResponse(BaseModel):
    answer: str
    num_original_documents: int

5.3. Endpoint für Antwortgenerierung

# dependencies.py
from fastapi import Body, HTTPException
from schemas import RAGRequest
from nodes import generate

async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await generate(body.prompt)

        return {
            "answer": generation.get("generation", ""),
            "num_original_documents": generation.get("num_original_documents", 0)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# main.py
@app.post("/generate_text", response_model=RAGResponse)
async def query_by_RAG_controller(generation: dict = Depends(get_generation)) -> RAGResponse:
    return RAGResponse(**generation)

5.4. Hintergrundverarbeitung

@app.post("/upload")
async def file_upload_controller(
        file: Annotated[UploadFile, File(description="Uploaded PDF documents")],
        bg_text_processor: BackgroundTasks,
):
    if file.content_type != "application/pdf":
        raise HTTPException(
            detail=f"Only uploading PDF documents are supported",
            status_code=status.HTTP_400_BAD_REQUEST,
        )
    try:
        filepath = await save_file(file)
        bg_text_processor.add_task(pdf_text_extractor, filepath)
        bg_text_processor.add_task(
            vector_service.store_file_content_in_db,
            filepath.replace("pdf", "txt")
        )
    except Exception as e:
        raise HTTPException(
            detail=f"An error occurred while saving file - Error: {e}",
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    return {"filename": file.filename, "message": "File uploaded successfully"}

6. Streamlit-Frontend

6.1. Datei hochladen

st.write("Upload a file to FastAPI")
file = st.file_uploader("Choose a file", type=["pdf"])

if st.button("Submit"):
    if file is not None:
        files = {"file": (file.name, file, file.type)}
        response = requests.post("http://localhost:8000/upload", files=files)
        st.write(response.text)
    else:
        st.write("No file uploaded.")

6.2. Gesprächshistorie

if "messages" not in st.session_state:
    st.session_state.messages = []

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

6.3. Chat-Workflow

if prompt := st.chat_input("Write your prompt in this input field"):
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.chat_message("user"):
        st.text(prompt)

    question = prompt

    response = requests.post(
        f"http://localhost:8000/generate_text",
        json={"prompt": question}
    )
    response.raise_for_status()

    response_json = response.json()
    answer = response_json["answer"]
    st.session_state.messages.append({"role": "assistant", "content": answer})

    with st.chat_message("assistant"):
        st.markdown(answer)

7. Zusammenfassung erstellen

Betriebskontrollliste