Від прототипу до виробництва: що я дізнався, створюючи справжні агентські системи ШІ
Переведення агентського ШІ від прототипу до продакшну за допомогою LangChain та LangGraph: чому прототипи застрягають, архітектура потоку документів та що заважає просуванню.
У цьому посібнику детально описано шлях від сировини до функціональної системи для книги «Від прототипу до виробництва: що я дізнався, створюючи справжні агентські системи ШІ за допомогою LangChain та LangGraph». Основна увага приділяється конкретним крокам виконання, чітким перевіркам та коду, який можна просто додати до репозиторію, не намагаючись здогадатися про його призначення.
Припиніть створювати прототипи ШІ. Почніть розробляти системи ШІ, які витримають випробування реальним світом.
На етапі «Припинити створення прототипів ШІ» необхідно визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори мають мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Краще використовувати невеликі, тестовані одиниці замість об’ємних скриптів. Коли крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану систему обробки даних. Необхідно встановити людське схвалення для кроків, які призводять до витрат грошей чи змінюють продуктивні дані. Підключення елементів під час компіляції не є гарантією повноти бізнес-функціоналу.
Чому LangChain та LangGraph?
На етапі Why LangChain та LangGraph необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори мають мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте результати роботи, визначте критерії успіху та не допускайте безповідомного часткового завершення. Вимагайте людського схвалення для операцій, які спричиняють витрати грошей чи змінюють дані у продакшені. Підключення елементів під час компіляції не є гарантією повноти бізнес-процесу.
Архітектура: від документів до рішень
На етапі «Архітектура з документів» необхідно визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Відображення витрат на ранньому етапі запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Встановлюйте людське схвалення для кроків, які вимагають фінансових витрат чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повноти бізнес-функціоналу.
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
Рівень 1: Канал RAG
Для етапу RAG рівня 1 необхідно визначити вхідні дані, відповідального за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Конфігурацію слід тримати окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Необхідно наводити конкретні уривки тексту, на яких ґрунтується відповідь. Без посилань оператори не зможуть відрізнити галюцинації від проблем з індексуванням.
Стратегія чанкінгу
На етапі стратегії чанкінгу необхідно визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Необхідно документувати як шлях успішного виконання, так і шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Встановлюйте людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повності функціоналу продукту. На етапі стратегії чанкінгу необхідно визначити вхідні дані, відповідальну особу за крок та критерії завершення перед зміною коду. Оператори повинні мати можливість знову виконати крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Називайте створювані елементи, визначайте критерії успіху та не допускайте безповідомного часткового завершення роботи.
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)
Вбудовування та векторне зберігання
Під час роботи над етапом вбудовування та векторного зберігання спочатку складіть опис вимог: необхідні вхідні дані, сигнал про успіх та наслідки часткової невдачі. Такий перелік допоможе уникнути нечесних змін у коді пізніше. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-середовищ до спільних. Вимірюйте рівень відтворення інформації на фіксованому наборі запитань перед налаштуванням підказок. Часта зміна підказок рідко допомагає покращити ефективність пошуку.
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,
)
Гібридний пошук — найважливіша оптимізація, яку ви здійснили
Під час роботи над етапом Hybrid Search The Most спочатку запишіть умови використання: необхідні параметри вхідних даних, сигнал про успішне виконання та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Створюйте контрольні точки після дорогих операцій. Функція відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор перезапускає пізнішу операцію.
retriever = vector_store.as_retriever(
search_type="hybrid",
search_kwargs={
"k": 5,
"score_threshold": 0.75,
}
)
Redis Embedding Cache — зниження витрат на 40%
Під час роботи над етапом Redis Embedding Cache 40 спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність у подальших змінах коду. Одночасно задокументуйте шлях успішного виконання та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Перед налаштуванням запитів вимірюйте рівень точності пошуку на фіксованому наборі запитань. Зміна запитів рідко допомагає покращити якість пошуку. Під час роботи над етапом Redis Embedding Cache 40 спочатку запишіть умови використання: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність у подальших змінах коду. Розглядайте цей етап як угоду між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте безповідомного часткового виконання завдань.
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
)
)
Рівень 2: Оркестрація агентів LangGraph
Етап роботи агентів LangGraph на Рівні 2 працює найкраще, якщо його розглядати як вимірювану поверхню. Запишіть один ідеальний запис діяльності, один випадок збою та примітки щодо скасування змін перед розширенням обсягу роботи. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам, коли процес переходить від демо-версії до спільних середовищ. Зберігайте стан графа у випрямленому та типованому форматі. Вкладені блоки приховують інформацію про те, який вузол заповнив певне поле, та ускладнюють продовження роботи після перерв.
Чому LangGraph кращий за прості ланцюги?
Підхід LangGraph Over Simple працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку про скасування змін перед розширенням обсягу роботи. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Тримайте стан графа простим та типованим. Вкладені блоки приховують інформацію про те, який вузол заповнив яке поле, і ускладнюють відновлення роботи після перерв.
Визначення стану
Етап «Визначення стану» працює найкраще, коли його розглядають як вимірювану поверхню. Зафіксуйте один ідеальний запис, один випадок збою та примітку щодо скасування змін перед розширенням обсягу роботи. Документуйте як успішний, так і відновлювальний сценарії роботи разом. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої доробки. Зберігайте стан графа у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив яке поле, що ускладнює продовження роботи після перерв. Етап «Визначення стану» працює найкраще, коли його розглядають як вимірювану поверхню. Зафіксуйте один ідеальний запис, один випадок збою та примітку щодо скасування змін перед розширенням обсягу роботи. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте мовчазного часткового завершення роботи.
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]
Створення графа
На етапі створення графа необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Записуйте час виконання та витрати на токени чи запити поруч із функціональними результатами. Візуалізація витрат заздалегідь запобігає несподіваним рахункам під час переходу з демо-середовища у спільні. Встановіть людське схвалення для ребер, які спричиняють витрати чи змінюють дані у продакшені. Підключення під час компіляції не є гарантією повноти бізнес-функціоналу.
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)
Вузол класифікатора
На етапі класифікаторних вузлів необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись визначити прихований стан. Конфігурацію слід зберігати окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Встановлюйте людське схвалення для дій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повноти бізнес-функціоналу.
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
Функції маршрутизації — серце LangGraph
Для етапу Routing Functions The Heart необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори мають мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Необхідно документувати як шлях успішного виконання, так і шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації. Встановлюйте людське схвалення для операцій, які призводять до витрат грошей чи змінюють дані у продакшені. Підключення на етапі компіляції не є гарантією повності функціоналу продукту. Для етапу Routing Functions The Heart необхідно визначити вхідні дані, власника кроку та критерії завершення перед зміною коду. Оператори мають мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте безповідомного часткового завершення роботи.
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"
Рівень 3: Шаблони надійності виробництва
Під час роботи над етапом надійності виробництва на рівні 3 спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається при частковій несправності. Цей перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із функціональними результатами. Відображення витрат на ранньому етапі запобігає несподіваним рахункам, коли процес переходить від демо-середовища до спільних середовищ. Створюйте контрольні точки після дорогих кроків. Функція відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор перезапускає пізнішу ланку.
Circuit Breaker з Polly (бекенд на C#)
Під час роботи з етапом Circuit Breaker у Polly спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається при частковій невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Зберігайте конфігурацію окремо від коду додатку. Файли середовища, сховища секретів та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь код. Робіть перевірки після дорогих операцій. Функція відновлення не повинна знову стягувати плату за той самий виклик LLM, коли оператор перезапускає пізнішу ланку.
// 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
)
);
Обмеження конкурентності (Python)
Під час роботи над етапом обмеження паралельних запитів у Python спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність у подальших змінах коду. Одночасно задокументуйте шлях успішного виконання та шлях відновлення. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої доробки. Фіксуйте ідентифікатор запиту, ідентифікатор моделі та час виконання кожного запиту. Без цих записів періодичні помилки постачальника виглядають як баги програми. Під час роботи над етапом обмеження паралельних запитів у Python спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність у подальших змінах коду. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Дайте назви всім елементам, визначте критерії успіху та не допускайте безповідомного часткового виконання.
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
Перехід від GPT-4 до GPT-3.5
Етапи переходу від GPT-4 до GPT-3.5 працюють найкраще, якщо їх розглядати як вимірювану структуру. Перед розширенням обсягу необхідно зафіксувати один ідеальний зразок роботи, один випадок збою та примітки щодо скасування змін. Записуйте час виконання та витрати на токени або запити поруч із функціональними результатами. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-версії до спільних середовищ. Зберігайте стан графа у простій та типованій формі. Вкладені структури приховують інформацію про те, який вузол заповнив певне поле, що ускладнює продовження роботи після перерв.
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")
Тестування на регресію для ідеального набору даних у процесах CI/CD
Етап CI CD для «золотого набору даних» працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку щодо скасування змін перед розширенням обсягу роботи. Тримайте конфігурацію окремо від коду додатку. Файли середовища, сховища конфіденційних даних та флаги функцій мають знаходитися в одному місці, де оператори можуть їх перевіряти, не читаючи весь граф. Тримайте стан графа простим та типованим. Вкладені блоки приховують інформацію про те, який вузол записав яке поле, і ускладнюють продовження роботи після перерв.
# 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}
Рівень 4: Спостережуваність
Етап спостережуваності на рівні 4 працює найкраще, якщо його розглядати як вимірювану поверхню. Збережіть один ідеальний запис, один випадок збою та примітку щодо скасування змін перед розширенням обсягу роботи. Одночасно задокументуйте шлях успішної роботи та шлях відновлення. Повторні спроби, людські перевірки та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Відстеження використання OpenTelemetry під час кожного запиту до ШІ
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
Ключові показники для відстеження
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",
}