Адзін прыклад да серыйявання: што я выучыў, ствараючы рэальныя сістэмы AI з агентамі
Пераход агентнага AI з пратотыпа да прыметнай эксплуатацыі за дапамою LangChain і LangGraph: чаму пратотыпы застаюцца на месцы, архітэктура практыку працы з дакументамі і што стварая працяванне.
У гэтым карыце параграфе перакладзена схема пераходу ад сыр'ёў да рабочай сістэмы для кнігі «Ад пратотыпа да вырабоцтва: Чаго я навучыўся, ствараючы рэальныя агентныя сістэмы AI з LangChain і LangGraph». Акцэнт ставіцца на практычныя крокі, чысткія перакананні і код, які можна проста дадаць у рэпазітарый, не прабуючы здогадвацца пра мету.
Хавайце ствараць пратотыпы AI. Пачніце выкладаць сістэмы AI, якія зможуць функцыонаваць у рэальным свеце.
Для этапы «Паўстаноўкі протатыпаў AI» неабходна перад змянай коду вызначыць даны, якія будуць вводзіцца, адпаведальнага за даныя крок і крэтырыі завершэння. Аперацыяныя працавнікі должны магчымае перадзваначыць крок з вядомага пункта контролю, не прымушаныя здагадваліцца пра схованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выйшоў, прычына неудачы должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны працэс. Неабходна людская апраўда для тых крокаў, якія выкарыстоўваюць грошы або зменяюць даныя для працы. Працэс кампілявання не ёсць гарантыяй полнай адпрацоўкі бізнес-процэса.
Чаму 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», спачатку запісайце умовы працы: неабяжлівыя даныя, сігнал успеху і тое, што выканаецца у разе частковага нявыпалення. Такі спіс контроля дапамагае заліцварыць змяны ў кодзе пазнейша.
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
Этап слоя 2 з агентамі LangGraph працуе наяўней, калі яго спрыяваць як мерыемую структуру. Запісайце адну ідэальную транскрыпцыю, адзін прыклад неудачы і прыметкі па поверненню да пачатковага стану пры расшырэнні масштаба. Запісвайце часы выканання і кост токенав або запытаў разам з функцыйнальнымі рэзультатамі. Відразлівае паказанне костаў з’являецца перашкоду неспакою, калі працэс пераходзіць з дамовай асяроддзя ў спакульную. Зберагаюце стан графа ў простам і тыпаваным фармаце. Вкладзеныя блокі маскуюць інфармацыю пра тое, які вузел запісаў які поле, і спаказваюць працэс пасля перарываў.
Чаму LangGraph, а не простыя ланцюгі?
Метод «Why LangGraph Over Simple stage» працюе найэфектывнейша, калі яго розглядаць як вімерную паверхню. Зафіксавце адны ідэальны прыклад, адну справу з бягам і прыметку па вярнэнні да пачатковага стану пры розшырэнні масштаба. Храніце настройкі пазырочна ад коду прыемлена. Файлы са станамі серавэра, хранальнікі секрэтных даных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куды аператары можаць адрабоўваць аўдыт без падчытання всіх даных графа. Храніце стан графа у простам і тыпаваным фармаце. Вярнутыя структуры даных маскуюць інфармацыю пра тое, які вузел запісаў які поле, і спакойваюць працэс пасля перарываў.
Апішанне стану
Этап «Апраначы стану» працюе наяўней, калі яго спрыяваць як мерымае паверхне. Зберажыце адну ідэальную транскрыпцію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расшырваць масштаб. Дакументавайце як шлях успеху, так і шлях вярнэння. Перапрыбуткі, людзкія контралі і обработка некорэктных паведамленняў ёсць часткай продукту, а не пасляднім дапрацоўкам. Храніце стан графа ў простым і тыпаваным фармаце. Вярнутыя блокі маскуюць, які вузел запісаў канкрэтны поле, і спакшуюць продовжэнне працы пасля перарываў. Этап «Апраначы стану» працюе наяўней, калі яго спрыяваць як мерымае паверхне. Зберажыце адну ідэальную транскрыпцію, адзін прыклад неудачы і запіс пра вярнэнне да пачатковага стану перш чым расшырваць масштаб. Спрыявайце гэты этап як кантракт межаў вхідных дадзеных і перакананых выходных рэзультатаў. Даць назвы артыфактам, задаць крытэрыя успеху і адмовіцца ад беззвучнага частковага завершэння.
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)
Узел класыфікатора
Для этапы «The Classifier Node» неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і крэтыяры завершэння пры перадзеіснаванні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы схованы стан. Конфігурацыю трэба захаваць пазначыльна ад коду прыкладнення. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функцый должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаяўшы весь граф. Прызначыць людскія аправарэння для рэшэнак, якія выдвайуць грошы або зменяюць даны ў працэсе. Падключэння пад час компілявання не ўзначае повнасці бізнес-функцый.
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
Для этапы «The Heart» функцыяў маршрутацыі неабходна падчас змены коду адзначыць вхідныя даны, адпаведальнага за крок і крэтыяры выходу. Аператары должны магчымае перзапускаць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабходна аддзекаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія пераказы і обработка некорэктных паведамленняў ёсць часткай продукту, а не дадатковым элементам да пасляднейшай дапрацоўкі. Неабходна людзкая аправарэнне для тых крокоў, якія выкарыстоўваюць грошы або зміняюць даны праўдзівай працы. Працэс кампіляцыі не ёсць гарантыяй полнай адпаведнасці продукту выклікам бізнесу. Для этапы «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# backend)
Калі працуеце з фазай 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)
Калі працуеце над стадзіяй «Concurrency Throttling Python», спачатку запісайце угоду: неабяжлівыя даныя, сігнал успеху і тое, што выходзіць у разе частковага нявыпання. Такі спіс пераконвае ў тым, што пазнейшыя змены коду будуць чыстымі. Документавайце як шлях успеху, так і шлях вяснавання сітуацыі. Перапрыбуткі, людзкія контралі і обработка некорэктных паведамленняў ёсць часткай продукту, а не пазнейшым дапрацоўкам. Зявляйце логі з ідэнтыфікаторам запіту, ідэнтыфікаторам мадэлі і часам адпаведзення за кожны вызов. Без такога следу перыодычныя памылкі прадастаўця выглядаюць як багі ў прыемніку. Калі працуеце над стадзіяй «Concurrency Throttling 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 Tracing у кожнам АІ-званку
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",
}