De prototipo a producción: lo que aprendí al crear sistemas reales de IA agente
Llevar la IA agente del prototipo a la producción con LangChain y LangGraph: por qué se estancan los prototipos, la arquitectura del pipeline de documentos y qué impide su avance.
Esta guía reconstruye el proceso desde las materias primas hasta un sistema funcional para: De prototipo a producción: Lo que aprendí al crear sistemas reales de IA agente con LangChain y LangGraph. El enfoque está en pasos operativos, verificaciones explícitas y código que puedes incorporar directamente a un repositorio sin tener que adivinar su propósito.
Deja de crear prototipos de IA. Comienza a lanzar sistemas de IA que sobrevivan en el mundo real.
En la fase de “Detener la creación de prototipos de IA”, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado. Incluya la aprobación humana en aquellos casos en que se gastan fondos o se modifican datos de producción. La conexión durante la compilación no equivale a una solución completa para el negocio.
¿Por qué LangChain y LangGraph?
En la etapa de Why LangChain y LangGraph, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Incluya la aprobación humana en aquellos casos en que se gasten fondos o se modifiquen datos de producción. La conexión en tiempo de compilación no equivale a la completitud del proceso empresarial.
La arquitectura: de los documentos a las decisiones
En la fase de “La arquitectura a partir de documentos”, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Implemente la aprobación humana en aquellos casos en los que se gastan fondos o se modifican datos de producción. La conexión durante la compilación no equivale a una solución completa desde el punto de vista empresarial.
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
Capa 1: La tubería RAG
Para la etapa RAG de Capa 1, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el sistema. Mencione las secciones que realmente sirvieron como base para la respuesta. Sin citaciones, los operadores no podrán distinguir entre alucinaciones y lagunas en el indexado.
Estrategia de fragmentación
En la fase de Estrategia de Fragmentación, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Incluya la aprobación humana en aquellos casos que impliquen gastos o cambios en los datos de producción. La configuración en tiempo de compilación no equivale a la completitud del proceso empresarial. En la fase de Estrategia de Fragmentación, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Considere esta fase como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas.
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)
Incorporación y almacenamiento vectorial
Al trabajar en la etapa de incorporación y almacenamiento vectorial, anote primero el contrato: entradas requeridas, señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando se pasa de entornos de demostración a entornos compartidos. Mida el rendimiento en la recuperación de información con un conjunto fijo de preguntas antes de ajustar los prompts. Cambiar constantemente los prompts rara vez soluciona un sistema deficiente de recuperación.
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,
)
Búsqueda híbrida: la optimización más importante que realizó
Al trabajar en la etapa “Hybrid Search The Most”, anote primero el contrato: los datos de entrada requeridos, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestas las futuras modificaciones en el código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Haga un punto de control después de los pasos costosos. La función de reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intente nuevamente un nodo posterior.
retriever = vector_store.as_retriever(
search_type="hybrid",
search_kwargs={
"k": 5,
"score_threshold": 0.75,
}
)
Caché de incrustación Redis: reducción del 40% en costos
Al trabajar en la etapa 40 de Redis Embedding Cache, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Documente tanto el camino óptimo como el de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no de mejoras posteriores. Mida la capacidad de recuperación con un conjunto fijo de preguntas antes de ajustar los prompts. El cambio constante de prompts rara vez soluciona un sistema de recuperación deficiente. Al trabajar en la etapa 40 de Redis Embedding Cache, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Trate esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina las verificaciones de éxito y rechace las completaciones parciales silenciosas.
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
)
)
Nivel 2: Orquestación de agentes LangGraph
La etapa de los agentes LangGraph en el Nivel 2 funciona mejor cuando se trata como una superficie medible. Capture una transcripción ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la versión de demostración a entornos compartidos. Mantenga el estado del grafo plano y con tipos definidos. Los bloques anidados ocultan qué nodo escribió qué campo y provocan interrupciones en la continuación del proceso.
¿Por qué LangGraph en lugar de cadenas simples?
El enfoque de LangGraph sobre la etapa “Simple” funciona mejor cuando se trata como una superficie medible. Capture una transcripción ejemplar, un caso de fallo y la nota de reversión antes de ampliar el alcance. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el grafo. Mantenga el estado del grafo plano y tipado; los bloques anidados ocultan qué nodo escribió qué campo y provocan interrupciones en la continuación del proceso.
Definición del estado
La etapa de Definición del Estado funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documente tanto el camino óptimo como el camino de recuperación juntos. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Mantenga el estado del grafo plano y tipado; los bloques anidados ocultan qué nodo escribió qué campo y provocan interrupciones en la continuación del proceso. La etapa de Definición del Estado funciona mejor cuando se trata como una superficie medible. Capture un registro ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Considere esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.
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]
Construcción del grafo
En la fase de construcción del grafo, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Coloque la aprobación humana en las conexiones que generan gastos o modifican datos de producción. La configuración en tiempo de compilación no equivale a la completitud del proceso empresarial.
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)
El nodo clasificador
En la etapa del Nodo Clasificador, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el grafo. Coloque la aprobación humana en las conexiones que generan gastos o modifican datos de producción. La conexión en tiempo de compilación no equivale a la completitud del proceso empresarial.
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
Funciones de enrutamiento: el corazón de LangGraph
En la fase The Heart de las Funciones de Enrutamiento, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Incluya la aprobación humana en aquellos procesos que generen gastos o modifiquen datos de producción. La configuración en tiempo de compilación no equivale a la completitud del proceso empresarial. En la fase The Heart de las Funciones de Enrutamiento, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta fase como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.
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"
Nivel 3: Patrones de fiabilidad en producción
Al trabajar en la etapa de fiabilidad en producción del Nivel 3, anote primero el contrato: entradas requeridas, señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Registre los tiempos y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de la versión de demostración a entornos compartidos. Haga una verificación después de los pasos costosos. La reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intenta nuevamente un nodo posterior.
Disyuntor eléctrico con Polly (backend en C#)
Al trabajar con la etapa Circuit Breaker de Polly, primero anote el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación ayuda a mantener honestos los cambios posteriores en el código. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el sistema. Haga un punto de control después de los pasos costosos. La función de reanudación no debe volver a facturar la misma llamada al LLM cuando un operador intente nuevamente un nodo posterior.
// 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
)
);
Limitación de concurrencia (Python)
Al trabajar en la etapa de control de concurrencia en Python, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Documente tanto el camino óptimo como el de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no de mejoras posteriores. Registre el ID de la solicitud, el ID del modelo y la latencia en cada llamada. Sin ese registro, los errores intermitentes del proveedor parecen bugs de la aplicación. Al trabajar en la etapa de control de concurrencia en Python, anote primero el contrato: las entradas requeridas, la señal de éxito y qué ocurre en caso de fallo parcial. Esa lista de verificación mantiene honestas las futuras modificaciones del código. Trate esta etapa como un contrato entre las entradas y las salidas validadas. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas.
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
Alternativa de GPT-4 a GPT-3.5
La etapa de transición de GPT-4 a GPT-3.5 funciona mejor cuando se trata como una superficie medible. Capture un transcripte ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Registre los tiempos y el costo en tokens o consultas junto con los resultados funcionales. La visibilidad temprana de los costos evita facturas inesperadas cuando el proceso pasa de la versión de demostración a entornos compartidos. Mantenga el estado del grafo en formato plano y tipado. Los bloques anidados ocultan qué nodo escribió qué campo y provocan interrupciones en la continuación del proceso.
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")
Pruebas de regresión CI/CD para el conjunto de datos ideal
La etapa de CI CD del conjunto de datos dorado funciona mejor cuando se trata como una superficie medible. Capture una transcripción ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin tener que leer todo el grafo. Mantenga el estado del grafo plano y tipado. Los bloques anidados ocultan qué nodo escribió qué campo y provocan interrupciones en la continuación del proceso.
# 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}
Capa 4: Observabilidad
La etapa de observabilidad de la Capa 4 funciona mejor cuando se trata como una superficie medible. Capture una transcripción ideal, un caso de fallo y la nota de reversión antes de ampliar el alcance. Documente junto con ello el camino óptimo y el camino de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores.
Rastreo con OpenTelemetry en cada llamada de IA
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
Métricas clave a seguir
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",
}