Von dem Prototypen zur Serienproduktion: Was ich beim Aufbau echter agenter KI-Systeme gelernt habe

Dieser Leitfaden zeigt den Weg von Rohstoffen bis zu einem funktionsfähigen System für das Buch „Von Prototyp zu Produktion: Was ich beim Aufbau echter agenter KI-Systeme mit LangChain und LangGraph gelernt habe“. Der Schwerpunkt liegt auf umsetzbaren Schritten, expliziten Überprüfungen sowie Code, den Sie ohne Rätseln über die Absicht direkt in ein Repository einfügen können.

Hören Sie auf, KI-Prototypen zu entwickeln. Beginnen Sie stattdessen mit der Bereitstellung von KI-Systemen, die in der realen Welt bestehen können.

In der Phase „Aufhören mit dem Bau von AI-Prototypen“ sollten 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. Zur Vorzugsbehandlung kommen kleine, testbare Einheiten vor umfangreichen Skripten. Wenn ein Schritt fehlschlägt, sollte der Fehler auf eine einzige Verantwortung verweisen und nicht auf ein verworrenes Ablaufverfahren. Menschliche Genehmigung ist bei Schritten erforderlich, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

Warum LangChain und LangGraph?

Zur Phase „Warum LangChain und LangGraph“ sollten Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien definieren, bevor Sie Code ändern. 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. Setzen Sie menschliche Freigabe für Kanten voraus, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch Geschäftsabschluss.

Die Architektur: Von Dokumenten zu Entscheidungen

In der Phase „The Architecture From Documents“ sollten Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden, bevor Code geändert wird. 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 Zeiten sowie Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen fest. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo-Umgebung in gemeinsam genutzte Umgebungen übergeht. Setzen Sie menschliche Freigabe für Schritte voraus, die Geld kosten oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

Document Upload
      ↓
Chunking Pipeline (LangChain TextSplitter)
      ↓
Embedding Generation (Azure OpenAI ada-002)
      ↓
Vector Store (Azure AI Search — Hybrid Search)
      ↓
LangGraph Agent Orchestration
    ├── Classifier Agent (GPT-4)
    ├── Extractor Agent (GPT-4)
    ├── Validator Agent (GPT-3.5)
    └── HITL Node (Human-in-the-Loop)
      ↓
Structured Output → Backend → Database

Ebene 1: Der RAG-Pipeline

Für die RAG-Ebene der Schicht 1 sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien 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. Die Konfiguration sollte außerhalb des Anwendungscode gespeichert werden. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Ablauf durchzulesen. 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.

Chunking-Strategie

Zur Phase der Chunking-Strategie sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Setzen Sie menschliche Freigabe für Schritte voraus, die Geld kosten oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet noch nicht vollständige Geschäftsabdeckung. Zur Phase der Chunking-Strategie sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene 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, teilweise abgeschlossene Vorgänge ab.

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=102,  # 20% overlap
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
)chunks = splitter.split_documents(documents)

Einbetten und Vektorlagern

Beim Arbeiten an der Phase des Einbettens und Vektorlagern sollten Sie zunächst den Vertrag festhalten: erforderliche Eingaben, Erfolgsignal 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 Einsatzbereich von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Messen Sie die Trefferquote anhand einer festgelegten Fragestellung, bevor Sie die Anfragen anpassen. Häufige Änderungen der Anfragen beheben selten ein schwaches Suchverhalten.

from langchain_openai import AzureOpenAIEmbeddings
from langchain_community.vectorstores import AzureSearch

embeddings = AzureOpenAIEmbeddings(
    azure_deployment="text-embedding-ada-002",
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
    api_key=os.getenv("AZURE_OPENAI_KEY"),
    api_version="2024-02-01",
)vector_store = AzureSearch(
    azure_search_endpoint=os.getenv("AZURE_SEARCH_ENDPOINT"),
    azure_search_key=os.getenv("AZURE_SEARCH_KEY"),
    index_name="tax-documents",
    embedding_function=embeddings.embed_query,
)

Hybridsuche – Die wichtigste Optimierung, die Sie vorgenommen haben

Wenn Sie die Phase „Hybrid Search The Most“ durcharbeiten, notieren Sie zunächst den Vertrag: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Operator ohne das Durchlesen des gesamten Systems prüfen können. Erstellen Sie nach aufwändigen Schritten einen Checkpoint. Das Wiederaufnehmen des Vorgangs sollte keine doppelte Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt.

retriever = vector_store.as_retriever(
    search_type="hybrid",
    search_kwargs={
        "k": 5,
        "score_threshold": 0.75,
    }
)

Redis Embedding Cache – 40 % Kostensenkung

Beim Arbeiten an der Redis Embedding Cache 40-Stufe sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Wiederherstellungsprozess gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlnachrichten gehören zum Produkt selbst und nicht zu späteren Optimierungen. Messen Sie die Trefferquote anhand eines festgelegten Fragebogens, bevor Sie die Anfragen anpassen. Eine häufige Änderung der Anfragen behebt selten ein schwaches Suchverhalten. Beim Arbeiten an der Redis Embedding Cache 40-Stufe sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgssignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Betrachten Sie diese Stufe als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Ergebnisse ab.

from langchain.cache import RedisSemanticCache
from langchain.globals import set_llm_cache
import redis

redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST"),
    port=6380,
    password=os.getenv("REDIS_PASSWORD"),
    ssl=True,
)set_llm_cache(
    RedisSemanticCache(
        redis_=redis_client,
        embedding=embeddings,
        score_threshold=0.99,  # Very high — only cache exact or near-exact matches
    )
)

Schicht 2: Orchestrierung von LangGraph-Agenten

Die Schicht-2-Stufe der LangGraph-Agenten funktioniert am besten, wenn sie als messbare Ebene betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein optimales Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. 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 Einsatzbereich von einer Demo in gemeinsame Umgebungen wechselt. Halten Sie den Graphenzustand flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Wiederaufnehmen der Arbeit.

Warum LangGraph statt einfacher Ketten?

Der Ansatz „Warum LangGraph gegenüber einfachen Lösungen vorteilhaft ist“ funktioniert am besten, wenn er als messbare Struktur betrachtet wird. Erfassen Sie ein gelungenes Beispiel, einen Fehlerfall sowie die Notizen zur Rücksetzung, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne das Durchlesen des gesamten Graphen überprüfen können. Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Datenstrukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen bei der Fortsetzung der Verarbeitung.

Definieren des Zustands

Die Phase „Zustand definieren“ funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Dokumentieren Sie gleichzeitig den erfolgreichen Ablauf sowie den Wiederherstellungsprozess. Wiederholversuche, menschliche Kontrollen sowie die Handhabung von Fehlern gehören zum Produkt selbst und nicht zu späteren Optimierungen. Halten Sie den Zustand des Graphen flach und typisiert – verschachtelte Strukturen verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Verarbeitung. Die Phase „Zustand definieren“ funktioniert am besten, wenn sie als messbarer Bereich betrachtet wird. Erfassen Sie ein „goldenes Transkript“, einen Fehlerfall sowie eine Notiz zur Rücksetzung, bevor Sie den Umfang erweitern. Betrachten Sie diese Phase als Vertrag zwischen Eingaben und validierten Ausgaben. Benennen Sie die relevanten Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.

from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, END
import operator

class DocumentProcessingState(TypedDict):
    # Input
    document_id: str
    document_text: str
    document_chunks: List[str]

    # Processing
    document_type: Optional[str]
    extracted_fields: Optional[dict]
    validation_errors: List[str]
    retry_count: int

    # Output
    confidence_score: float
    requires_human_review: bool
    processing_notes: str
    final_result: Optional[dict]

Der Graph aufbauen

Zur Phase des Aufbaus des Graphen sollten vor der Codeänderung die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien 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. Erfassen Sie neben den funktionalen Ergebnissen auch die Laufzeiten sowie die Kosten für Tokens oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Setzen Sie menschliche Freigabe für Kanten voraus, die Geld kosten oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(DocumentProcessingState)# Add nodes
workflow.add_node("classifier", classify_document)
workflow.add_node("extractor", extract_fields)
workflow.add_node("validator", validate_extraction)
workflow.add_node("human_review", flag_for_human_review)
workflow.add_node("finalizer", finalize_result)# Define edges
workflow.set_entry_point("classifier")workflow.add_conditional_edges(
    "classifier",
    route_after_classification,
    {
        "extract": "extractor",
        "human_review": "human_review",
        "unknown": END,
    }
)workflow.add_conditional_edges(
    "extractor",
    route_after_extraction,
    {
        "validate": "validator",
        "retry": "extractor",
        "human_review": "human_review",
    }
)workflow.add_conditional_edges(
    "validator",
    route_after_validation,
    {
        "finalize": "finalizer",
        "retry": "extractor",
        "human_review": "human_review",
    }
)workflow.add_edge("finalizer", END)
workflow.add_edge("human_review", END)# Compile with checkpointing for resumability
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

Der Klassifikationsknoten

Zur Phase des Klassifizierungs-Nodes sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien 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. Die Konfiguration sollte außerhalb des Anwendungscode gespeichert werden. Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Graphen durchlesen zu müssen. Menschliche Freigabe sollte für Kanten erforderlich sein, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet nicht automatisch vollständige Geschäftsabdeckung.

from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class ClassificationResult(BaseModel):
    document_type: str = Field(
        description="One of: TAX_RETURN, PAYSLIP, TRAVEL_EXPENSE, IMMIGRATION_DOC, INVOICE, UNKNOWN"
    )
    confidence_score: float = Field(
        description="Confidence between 0.0 and 1.0"
    )
    reasoning: str = Field(
        description="Brief reasoning for the classification"
    )llm = AzureChatOpenAI(
    azure_deployment="gpt-4",
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
    api_key=os.getenv("AZURE_OPENAI_KEY"),
    api_version="2024-02-15-preview",
    temperature=0,  # Always 0 for classification — determinism matters
)classification_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an expert tax document classifier.

Classify the document and return ONLY valid JSON matching the schema.
NEVER hallucinate values not present in the document.
If you cannot classify with confidence > 0.7, return UNKNOWN.{format_instructions}"""),
    ("human", "Document text:\n\n{document_text}"),
])parser = JsonOutputParser(pydantic_object=ClassificationResult)classifier_chain = classification_prompt | llm | parserdef classify_document(state: DocumentProcessingState) -> DocumentProcessingState:
    try:
        result = classifier_chain.invoke({
            "document_text": state["document_text"][:3000],  # Truncate for context window
            "format_instructions": parser.get_format_instructions(),
        })

        state["document_type"] = result["document_type"]
        state["confidence_score"] = result["confidence_score"]
        state["processing_notes"] += f"Classified as {result['document_type']} with confidence {result['confidence_score']:.2f}. "

    except Exception as e:
        state["document_type"] = "UNKNOWN"
        state["confidence_score"] = 0.0
        state["processing_notes"] += f"Classification failed: {str(e)}. "

    return state

Routenfunktionen – Das Herzstück von LangGraph

Für die Routing Functions The Heart-Phase sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlnachrichten gehören zum Produkt selbst, nicht zu späteren Optimierungen. Setzen Sie menschliche Freigabe für Schritte voraus, die Geld ausgeben oder Produktionsdaten ändern. Eine Verkabelung zur Kompilierzeit bedeutet noch nicht vollständige Geschäftsabdeckung. Für die Routing Functions The Heart-Phase sollten vor dem Ändern des Codes die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien definiert werden. Die Bediener sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.

def route_after_classification(state: DocumentProcessingState) -> str:
    if state["document_type"] == "UNKNOWN":
        return "unknown"
    elif state["confidence_score"] < 0.7:
        state["requires_human_review"] = True
        return "human_review"
    else:
        return "extract"

def route_after_extraction(state: DocumentProcessingState) -> str:
    if state["retry_count"] >= 3:
        state["requires_human_review"] = True
        return "human_review"
    elif not state.get("extracted_fields"):
        state["retry_count"] = state.get("retry_count", 0) + 1
        return "retry"
    else:
        return "validate"def route_after_validation(state: DocumentProcessingState) -> str:
    if state["validation_errors"]:
        if state["retry_count"] >= 3:
            state["requires_human_review"] = True
            return "human_review"
        state["retry_count"] = state.get("retry_count", 0) + 1
        return "retry"
    return "finalize"

Schicht 3: Muster zur Produktionszuverlässigkeit

Beim Arbeiten an der Schicht 3 zur Produktionszuverlässigkeit sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsindikator 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 Tokens oder Abfragen neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo-Umgebung in gemeinsam genutzte Umgebungen übergeht. Legen Sie nach aufwändigen Schritten einen Kontrollpunkt an. Das Wiederaufnehmen des Vorgangs sollte keine erneuten Gebühren für denselben LLM-Aufruf verursachen, wenn ein Operator einen späteren Knoten erneut ausführt.

Circuit Breaker mit Polly (C#-Backend)

Beim Arbeiten mit der Polly-Stufe „Circuit Breaker“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Erfolgsignal sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne das Durchlesen des gesamten Systems überprüfen können. Führen Sie nach aufwändigen Schritten einen Checkpoint durch. Das Wiederaufnehmen des Vorgangs sollte keine doppelte Abrechnung für denselben LLM-Aufruf verursachen, wenn ein Betreuer einen späteren Knoten erneut ausführt.

// In .NET Core — Polly circuit breaker
services.AddHttpClient<ILLMGateway, LLMGateway>()
    .AddPolicyHandler(
        Policy<HttpResponseMessage>
            .Handle<HttpRequestException>()
            .OrResult(r => (int)r.StatusCode >= 500)
            .CircuitBreakerAsync(
                handledEventsAllowedBeforeBreaking: 5,
                durationOfBreak: TimeSpan.FromSeconds(60),
                onBreak: (ex, duration) =>
                    logger.LogWarning("Circuit breaker opened for {Duration}s", duration),
                onReset: () =>
                    logger.LogInformation("Circuit breaker reset")
            )
    )
    .AddPolicyHandler(
        Policy<HttpResponseMessage>
            .Handle<HttpRequestException>()
            .WaitAndRetryAsync(
                retryCount: 3,
                sleepDurationProvider: attempt =>
                    TimeSpan.FromSeconds(Math.Pow(2, attempt)) + // Exponential
                    TimeSpan.FromMilliseconds(new Random().Next(0, 1000)) // Jitter
            )
    );

Konkurrenzbeschränkung (Python)

Beim Arbeiten an der Phase „Concurrency Throttling in Python“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Signal für Erfolg sowie das Vorgehen bei teilweisen Fehlern. Diese Checkliste sorgt dafür, dass spätere Codeänderungen transparent bleiben. Dokumentieren Sie sowohl den erfolgreichen Ablauf als auch den Wiederherstellungsprozess gemeinsam. Wiederholte Versuche, menschliche Überprüfungen sowie die Handhabung von Fehlnachrichten gehören zum Produkt selbst und nicht zu späteren Optimierungen. Protokollieren Sie bei jedem Aufruf die Anfrage-ID, die Modell-ID sowie die Latenzzeit. Ohne diese Aufzeichnungen wirken intermittierende Fehler des Anbieters wie Programmfehler. Beim Arbeiten an der Phase „Concurrency Throttling in Python“ sollten Sie zunächst den Vertrag aufschreiben: erforderliche Eingaben, Signal für Erfolg 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 Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Vorgänge ab.

import asyncio
from asyncio import Semaphore

class ThrottledDocumentProcessor:
    def __init__(self, max_concurrent: int = 50):
        self._semaphore = Semaphore(max_concurrent)

    async def process_document(
        self,
        document_id: str,
        document_text: str
    ) -> dict:
        async with self._semaphore:
            config = {"configurable": {"thread_id": document_id}}
            result = await app.ainvoke(
                {
                    "document_id": document_id,
                    "document_text": document_text,
                    "document_chunks": [],
                    "validation_errors": [],
                    "retry_count": 0,
                    "confidence_score": 0.0,
                    "requires_human_review": False,
                    "processing_notes": "",
                    "final_result": None,
                },
                config=config,
            )
            return result

Umstieg von GPT-4 auf GPT-3.5

Der fünfstufige Übergang von GPT-4 zu GPT-3.5 funktioniert am besten, wenn er als messbare Struktur betrachtet wird. Erfassen Sie vor Erweiterung des Umfangs ein optimales Transkript, einen Fehlerfall sowie eine Notiz zur Rücksetzung. Notieren Sie die Laufzeiten sowie die Kosten pro Token oder Abfrage neben den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von einer Demo in gemeinsame Umgebungen übergeht. Halten Sie den Graphenzustand strukturiert und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Fortsetzen der Arbeit.

async def classify_with_fallback(document_text: str) -> dict:
    models = [
        ("gpt-4", 0.95),       # Primary — higher confidence weight
        ("gpt-35-turbo", 0.80), # Fallback — lower confidence weight
    ]

    for deployment, confidence_multiplier in models:
        try:
            llm = AzureChatOpenAI(
                azure_deployment=deployment,
                temperature=0,
                request_timeout=30,
            )
            result = await classifier_chain.ainvoke({
                "document_text": document_text,
                "format_instructions": parser.get_format_instructions(),
            })
            result["confidence_score"] *= confidence_multiplier
            result["model_used"] = deployment
            return result
        except Exception as e:
            logger.warning(f"Model {deployment} failed: {e}. Trying next...")
            continue

    raise Exception("All models failed")

Regressionstests für das optimale Datensatz-Set in CI/CD

Die CI/CD-Phase für das „Golden Dataset“ funktioniert am besten, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zum Rollback, bevor Sie den Umfang erweitern. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den Betreuer ohne Durchsicht des gesamten Graphen prüfen können. Halten Sie den Zustand des Graphen flach und typisiert. Verschachtelte Datenblöcke verbergen, welcher Knoten welches Feld geschrieben hat, und führen zu Unterbrechungen beim Wiederaufnehmen der Verarbeitung.

# golden_dataset.py
GOLDEN_DOCUMENTS = [
    {
        "id": "GD001",
        "text": "Employee: John Smith\nDate: 15-Mar-2026\nGross Pay: $4,500\nNet Pay: $3,200...",
        "expected_type": "PAYSLIP",
        "expected_fields": {
            "employee_name": "John Smith",
            "date": "2026-03-15",
            "gross_amount": 4500.0,
        },
        "min_confidence": 0.85,
    },
    # ... 200 more golden examples across all document types
]

async def run_regression_suite():
    passed = 0
    failed = 0
    failures = []

    for golden_doc in GOLDEN_DOCUMENTS:
        result = await processor.process_document(
            golden_doc["id"],
            golden_doc["text"]
        )

        # Check type
        if result["document_type"] != golden_doc["expected_type"]:
            failures.append({
                "id": golden_doc["id"],
                "expected_type": golden_doc["expected_type"],
                "actual_type": result["document_type"],
                "failure": "wrong_classification",
            })
            failed += 1
            continue

        # Check confidence
        if result["confidence_score"] < golden_doc["min_confidence"]:
            failures.append({
                "id": golden_doc["id"],
                "expected_confidence": golden_doc["min_confidence"],
                "actual_confidence": result["confidence_score"],
                "failure": "low_confidence",
            })
            failed += 1
            continue

        passed += 1

    pass_rate = passed / len(GOLDEN_DOCUMENTS)

    if pass_rate < 0.95:  # 95% pass rate required
        raise Exception(
            f"Regression suite failed: {pass_rate:.1%} pass rate "
            f"({failed} failures). Blocking deployment."
        )

    return {"passed": passed, "failed": failed, "pass_rate": pass_rate}

Schicht 4: Beobachtbarkeit

Die Beobachtbarkeitsphase der Schicht 4 funktioniert am besten, wenn sie als messbarer Ansatz betrachtet wird. Erfassen Sie ein „goldenes“ Transkript, einen Fehlerfall sowie eine Notiz zum Rollback, bevor Sie den Umfang erweitern. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Wiederherstellungsprozess. Wiederholversuche, menschliche Überprüfungen sowie die Handhabung von Fehlnachrichten gehören zum Produkt selbst und nicht zu späteren Optimierungen.

OpenTelemetry Tracing bei jeder KI-Aufruf

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(
    BatchSpanProcessor(
        AzureMonitorTraceExporter(
            connection_string=os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
        )
    )
)
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)async def classify_document_traced(
    state: DocumentProcessingState,
) -> DocumentProcessingState:
    with tracer.start_as_current_span("classify_document") as span:
        span.set_attribute("document.id", state["document_id"])
        span.set_attribute("document.length", len(state["document_text"]))

        start_time = time.time()
        result = await classify_document(state)
        latency_ms = (time.time() - start_time) * 1000

        span.set_attribute("llm.document_type", result["document_type"])
        span.set_attribute("llm.confidence_score", result["confidence_score"])
        span.set_attribute("llm.latency_ms", latency_ms)
        span.set_attribute("llm.model", "gpt-4")
        span.set_attribute("llm.requires_human_review", result["requires_human_review"])

        return result

Wichtige Metriken zur Überwachung

METRICS_TO_TRACK = {
    # Reliability
    "classification_accuracy": "% correct vs golden dataset",
    "confidence_score_distribution": "histogram — drift detection",
    "human_review_rate": "% documents escalated — should stay < 5%",

    # Performance
    "end_to_end_latency_p50": "median processing time per document",
    "end_to_end_latency_p99": "tail latency — SLA violations",
    "llm_call_latency": "broken down by model and node",

    # Cost
    "tokens_per_document": "input + output tokens",
    "cache_hit_rate": "Redis embedding cache effectiveness",
    "monthly_api_spend": "trend by document type",

    # Business
    "documents_processed_per_hour": "throughput",
    "analyst_hours_saved_weekly": "business impact",
    "error_rate_by_document_type": "which types fail most",
}

Die Ergebnisse

Wichtige Erkenntnisse

Fazit

Operative Checkliste