Od prototypu do produkcji: co nauczyłem się, budując prawdziwe systemy AI o zdolnościach autonomicznych
Przenoszenie sztucznej inteligencji o charakterze agentowym od prototypu do produkcji za pomocą LangChain i LangGraph: dlaczego prototypy utykają, architektura pipeline’u dokumentów oraz czynniki, które temu przeszkadzają.
To przewodnik odtwarza drogę od surowców do gotowego systemu dla: Od prototypu do produkcji: Co nauczyłem się budując prawdziwe systemy AI z LangChain i LangGraph. Skupia się na krokach realizowalnych w praktyce, wyraźnych sprawdzeniach oraz kodzie, który można bez problemu umieścić w repozytorium, nie musząc zgadywać intencji.
Przestań budować prototypy AI. Zacznij wdrażać systemy AI, które przetrwają w rzeczywistym świecie.
W fazie zatrzymania tworzenia prototypów AI należy określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydatków lub zmian w danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego.
Dlaczego LangChain i LangGraph?
W fazie Why LangChain i LangGraph należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij pliki generowane, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Wymagaj ludzkiej aprobaty dla operacji, które wiążą się z wydatkami lub zmianami w danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności biznesowej.
Architektura: Od dokumentów do decyzji
W fazie „Architektura z dokumentów” należy określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu systemu. Należy rejestrować czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Zatwierdzenie przez człowieka jest konieczne w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie gwarantują pełnej kompletności rozwiązania biznesowego.
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
Warstwa 1: Pipeline RAG
Dla etapu RAG w warstwie 1 należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury. Należy wskazać konkretne fragmenty, które stanowiły podstawę odpowiedzi. Bez tych odniesień operatorzy nie będą w stanie odróżnić halucynacji od braku danych w indeksie.
Strategia dzielenia na fragmenty
W fazie Strategii Dzielenia na Fragmenty należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Konieczne jest uzyskanie zatwierdzenia człowieka w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności funkcjonalności biznesowej. W fazie Strategii Dzielenia na Fragmenty należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Należy nadać nazwy poszczególnym elementom, zdefiniować kryteria sukcesu oraz odrzucić przypadki cichego, częściowego ukończenia zadania.
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)
Zaawansowane przechowywanie wektorowe i embeddingi
Podczas pracy nad etapem zaawansowanego przechowywania wektorowego i embeddingów najpierw zapisz specyfikację: wymagane dane wejściowe, sygnał o pomyślności oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Obok wyników funkcjonalnych zapisz czas wykonywania operacji oraz koszt tokenów lub zapytań. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy przechodzi się z środowiska demonstracyjnego do współdzielonych środowisk. Zmierz stopień odzyskiwania informacji na ustalonej grupie pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko poprawiają słabe wyniki wyszukiwania.
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,
)
Wyszukiwanie hybrydowe — najważniejsza optymalizacja, jaką wprowadziłeś
Gdy przechodzisz przez etap Hybrid Search The Most, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. System powinien unikać ponownego pobierania opłat za tę samą funkcję LLM, gdy operator ponawia próbę z późniejszym węzłem.
retriever = vector_store.as_retriever(
search_type="hybrid",
search_kwargs={
"k": 5,
"score_threshold": 0.75,
}
)
Redis Embedding Cache — 40% redukcja kosztów
Gdy pracujesz nad etapem Redis Embedding Cache 40, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno prawidłowy przebieg, jak i ścieżkę odzyskiwania. Próby ponowne, kontrola przez ludzi oraz obsługa wiadomości błędnych są częścią produktu, a nie elementem późniejszej dopracowywania. Zmierz stopień odzyskiwania informacji na ustalonej serii pytań przed dostosowywaniem promptów. Częste zmiany promptów rzadko naprawiają słabe możliwości wyszukiwania. Gdy pracujesz nad etapem Redis Embedding Cache 40, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań.
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
)
)
Szczegół 2: Orkiestracja agentów LangGraph
Etap agentów LangGraph w Szczególe 2 działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zapisuj czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Utrzymuj stan grafu w formie prostych, spójnych struktur. Zagłębione elementy ukrywają informację o tym, który węzeł zapisał dane w danym polu, i powodują przerwę w kontynuacji działania po zakłóceniach.
Dlaczego LangGraph zamiast prostych łańcuchów?
Metoda Why LangGraph Over Simple działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę o cofnięciu zmian, zanim rozszerzysz zakres. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całego grafu. Utrzymuj stan grafu w prostej formie i z określonym typem. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane w danym polu, co utrudnia kontynuację pracy po przerwach.
Definiowanie stanu
Faza Definiowania Stanu działa najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zdokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę przywracania do normalnego stanu. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieprzyjętych stanowią część produktu, a nie elementy dodawane później. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji po przerwach. Faza Definiowania Stanu działa najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi danymi wyjściowymi. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań.
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]
Budowanie grafu
W fazie budowania grafu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega nieoczekiwanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Zatwierdzenie przez człowieka powinno być wymagane dla krawędzi, które generują wydatki lub zmieniają dane produkcyjne. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności biznesowej.
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)
Węzeł klasyfikatora
W etapie węzła klasyfikatora należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Zastosuj zatwierdzenie przez człowieka do połączeń, które powodują wydatki lub zmieniają dane produkcyjne. Połączenia ustalane w czasie kompilacji nie równają się pełności funkcjonalności biznesowej.
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
Funkcje routingu — serce LangGraph
Dla etapu Routing Functions The Heart należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie wykonać ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później. Konieczne jest uzyskanie zatwierdzenia człowieka w przypadku operacji, które powodują wydatki lub zmieniają dane produkcyjne. Połączenia realizowane w czasie kompilacji nie równają się pełnej kompletności funkcjonalności biznesowej. Dla etapu Routing Functions The Heart należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie wykonać ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Należy nadać nazwy poszczególnym elementom, zdefiniować kryteria sukcesu oraz odrzucić przypadki cichego, częściowego ukończenia zadania.
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"
Szczegół 3: Wzorce niezawodności w produkcji
Gdy przechodzisz przez etap niezawodności w produkcji na poziomie Szczegółu 3, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego awarii. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Obok wyników funkcjonalnych zapisz czas wykonywania oraz koszt tokena lub zapytania. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą funkcję LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Circuit Breaker z Polly (backend w C#)
Gdy pracujesz z etapem Circuit Breaker w Polly, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego awarii. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Ustaw punkty kontrolne po kosztownych krokach. Funkcja wznowienia nie powinna ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
// 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
)
);
Ograniczanie równoczesności (Python)
Gdy pracujesz nad etapem Concurrency Throttling w Pythonie, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno prawidłowy przebieg, jak i ścieżkę naprawczą. Próby ponowne, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dopinane później. Zapisuj ID żądania, ID modelu oraz opóźnienie przy każdej próbie połączenia. Bez tych informacji przerywane błędy dostawcy wyglądają jak błędy aplikacji. Gdy pracujesz nad etapem Concurrency Throttling w Pythonie, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań.
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
Przejście z GPT-4 na GPT-3.5 jako rozwiązanie awaryjne
Proces przechodzenia z GPT-4 na GPT-3.5 działa najlepiej, gdy traktuje się go jako mierzalną powierzchnię do analizy. Zanim rozszerzy się zakres pracy, należy zarejestrować jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian. Obok wyników funkcjonalnych należy zapisywać czasy wykonywania oraz koszt tokenów lub zapytań. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się od środowiska demonstracyjnego do współdzielonych środowisk. Należy utrzymywać stan grafu w formie prostych, uporządkowanych danych. Zagłębione struktury ukrywają informacje o tym, który węzeł zapisał dane do którego pola, co powoduje przerwę w kontynuacji pracy po interwencjach.
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")
Testy regresji w ramach CI/CD dla idealnego zestawu danych
Etap CI CD dla Złotego Zestawu Danych funkcjonuje najlepiej, gdy traktowany jest jako mierzalna powierzchnia do analizy. Zapisz jeden wzorcowy zapis, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury. Utrzymuj stan struktury w formie prostych, typizowanych elementów. Wplecione dane ukrywają informację o tym, który węzeł zapisał dany pole, co utrudnia kontynuację pracy po przerwach.
# 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}
Szczegół 4: Obserwowalność
Etap obserwowalności Szczegółu 4 działa najlepiej, gdy traktowany jest jako mierzalna powierzchnia do analizy. Zapisz jeden wzorcowy zapis, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zdokumentuj zarówno prawidłowy przebieg działania, jak i ścieżkę przywracania do normalnego stanu. Próby ponownych działań, mechanizmy kontroli ludzkiej oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
OpenTelemetry Tracing przy każdej wywołaniu AI
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
Kluczowe wskaźniki do śledzenia
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",
}