От прототипа к производству: что я узнал, создавая настоящие агентные ИИ-системы
Переход агентного ИИ от прототипа к производственному использованию с 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 первого уровня необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Конфигурацию следует хранить отдельно от кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь код. Указывайте те участки текста, на которых основан ответ. Без цитат операторы не смогут отличить галлюцинации от пробелов в индексации.
Стратегия разбиения на чанки
На этапе стратегии разбиения на части необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Необходимо задокументировать как успешный, так и восстановительный сценарии работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не этапом последующей доработки. Внедрять утверждение человеком для операций, связанных с тратой денег или изменением производственных данных. Настройка на этапе компиляции не гарантирует полноты функционала продукта. На этапе стратегии разбиения на части необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успеха и не допускайте безответственного частичного завершения работы.
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 — снижение затрат на 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 сначала запишите условия работы: необходимые входные данные, сигнал успешного выполнения и действия при частичной неисправности. Такой список помогает сохранять честность при последующих изменениях кода. Рядом с функциональными результатами записывайте время выполнения и стоимость токенов или запросов. Отображение затрат с самого начала предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Выполняйте проверки после дорогостоящих операций. При повторной попытке обработки последующего узла система не должна снова взимать плату за один и тот же вызов большой языковой модели.
Оптовик с 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",
}