¿Tu Customer Service Tarda 45 Minutos en Responder?

from typing import TypedDict, Annotated, Sequence from langchain_core.messages import BaseMessage import operator class CustomerServiceState(TypedDict): """ State definition para sistema multi-agent customer service. Atributos: messages: Historial completo de conversación (user + agent). current_intent: Intent clasificado por Router (faq, technical, action, escalate). retrieved_docs: Documentos relevantes del knowledge base. confidence_score: Confidence del LLM en su respuesta (0-1). actions_taken: Lista de acciones ejecutadas (para audit trail). customer_context: Info del customer (ID, tier, history). needs_escalation: Flag si requiere human intervention. next_agent: Qué agente debe ejecutar siguiente. """ # Messages se appendean (no overwrite) usando operator.add messages: Annotated[Sequence[BaseMessage], operator.add] # Intent clasificado por router current_intent: str # Retrieved docs del knowledge base retrieved_docs: list[dict] # Confidence score del LLM (0-1) confidence_score: float # Acciones ejecutadas (audit trail) actions_taken: list[str] # Customer context (CRM data) customer_context: dict # Flag escalation needs_escalation: bool # Next agent a ejecutar next_agent: str 

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate import logging logger = logging.getLogger(__name__) # Modelo ligero para routing (rápido + barato) router_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) def router_agent(state: CustomerServiceState) -> CustomerServiceState: """ Router Agent: Clasifica intent del usuario y decide next path. Paths posibles: - 'faq': Query simple, fast path - 'technical': Problema complejo, retrieval profundo - 'action': Requiere ejecutar acción (refund, update) - 'escalate': Fuera de scope, necesita humano """ try: # Get último mensaje del usuario user_message = state["messages"][-1].content # Prompt para clasificación classification_prompt = ChatPromptTemplate.from_messages([ ("system", """Eres un clasificador de intents para customer service. Analiza el mensaje del usuario y clasifica en UNA de estas categorías: 1. FAQ: Preguntas simples sobre productos, pricing, shipping (respuesta directa en knowledge base) 2. TECHNICAL: Problemas técnicos, troubleshooting, errores (requiere investigación) 3. ACTION: Requiere ejecutar acción concreta (refund, cambiar plan, update info) 4. ESCALATE: Legal, complaints formales, sensitive topics (requiere humano) Responde SOLO con: FAQ, TECHNICAL, ACTION o ESCALATE"""), ("user", "{query}") ]) # Invoke LLM chain = classification_prompt | router_llm response = chain.invoke({"query": user_message}) intent = response.content.strip().upper() # Validate intent valid_intents = ["FAQ", "TECHNICAL", "ACTION", "ESCALATE"] if intent not in valid_intents: logger.warning(f"Invalid intent {intent}, defaulting to TECHNICAL") intent = "TECHNICAL" # Determine next agent next_agent_map = { "FAQ": "retrieval_fast", "TECHNICAL": "retrieval_deep", "ACTION": "action_executor", "ESCALATE": "human_escalation" } logger.info(f"Classified intent: {intent} → next: {next_agent_map[intent]}") # Update state return { "current_intent": intent.lower(), "next_agent": next_agent_map[intent], "confidence_score": 1.0 # Router siempre confident (es classification) } except Exception as e: logger.error(f"Router failed: {str(e)}") # Fallback: escalate a humano si router falla return { "current_intent": "escalate", "next_agent": "human_escalation", "confidence_score": 0.0, "needs_escalation": True } 

from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_pinecone import PineconeVectorStore from langchain_core.prompts import ChatPromptTemplate from langchain.chains import create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain import os # Setup Pinecone vector store embeddings = OpenAIEmbeddings(model="text-embedding-ada-002") vectorstore = PineconeVectorStore( index_name=os.getenv("PINECONE_INDEX_NAME"), embedding=embeddings ) # LLM para generar respuesta answer_llm = ChatOpenAI(model="gpt-4", temperature=0.3) def retrieval_agent_deep(state: CustomerServiceState) -> CustomerServiceState: """ Retrieval Agent (Deep): RAG completo para queries técnicas complejas. Process: 1. Embed user query 2. Search top-k docs en vector DB 3. Rerank por relevancia 4. Generate respuesta con context 5. Calculate confidence score """ try: user_query = state["messages"][-1].content # Retrieval con hybrid search (semantic + keyword) retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={ "k": 5, # Top 5 docs "filter": {"type": "technical_docs"} # Solo docs técnicos } ) # Prompt para respuesta answer_prompt = ChatPromptTemplate.from_messages([ ("system", """Eres un agente de soporte técnico experto. Usa SOLO la información del contexto proporcionado para responder. Si la información no está en el contexto, di "No tengo información suficiente para responder esto con certeza. Te estoy conectando con un especialista." Formato de respuesta: - Clara y concisa - Paso a paso si es procedimiento - Incluye links relevantes si están en el contexto - Si no estás 100% seguro, escala a humano"""), ("user", "Context: {context}\ \ Question: {input}") ]) # Crear chain combine_docs_chain = create_stuff_documents_chain(answer_llm, answer_prompt) retrieval_chain = create_retrieval_chain(retriever, combine_docs_chain) # Execute response = retrieval_chain.invoke({"input": user_query}) # Calculate confidence (si respuesta contiene "no tengo información" → low confidence) answer_text = response["answer"].lower() confidence = 0.9 if "no tengo información" not in answer_text else 0.3 # Check si necesita escalation needs_escalation = confidence < 0.7 return { "retrieved_docs": [ {"content": doc.page_content, "metadata": doc.metadata} for doc in response["context"] ], "messages": [("assistant", response["answer"])], "confidence_score": confidence, "needs_escalation": needs_escalation, "next_agent": "human_escalation" if needs_escalation else "end" } except Exception as e: logger.error(f"Retrieval failed: {str(e)}") return { "messages": [("assistant", "Lo siento, tuve un problema técnico. Te conecto con un agente humano.")], "confidence_score": 0.0, "needs_escalation": True, "next_agent": "human_escalation" } 

from langgraph.graph import StateGraph, END from langgraph.checkpoint.postgres import PostgresSaver import os # Import nodes from nodes.router_agent import router_agent from nodes.retrieval_agent import retrieval_agent_deep, retrieval_agent_fast from nodes.action_agent import action_executor from nodes.escalation_agent import human_escalation def create_customer_service_graph(): """ Construye el graph multi-agent con conditional routing. """ # Inicializar graph con State schema workflow = StateGraph(CustomerServiceState) # Add nodes workflow.add_node("router", router_agent) workflow.add_node("retrieval_fast", retrieval_agent_fast) workflow.add_node("retrieval_deep", retrieval_agent_deep) workflow.add_node("action_executor", action_executor) workflow.add_node("human_escalation", human_escalation) # Set entry point (primer nodo a ejecutar) workflow.set_entry_point("router") # Conditional edges: Router decide next path def route_after_classification(state: CustomerServiceState) -> str: """Routing logic basado en intent.""" return state["next_agent"] workflow.add_conditional_edges( "router", route_after_classification, { "retrieval_fast": "retrieval_fast", "retrieval_deep": "retrieval_deep", "action_executor": "action_executor", "human_escalation": "human_escalation" } ) # Edges desde retrieval/action nodes # Si needs_escalation = True → human, sino → END def check_escalation(state: CustomerServiceState) -> str: return "human_escalation" if state.get("needs_escalation", False) else END workflow.add_conditional_edges("retrieval_fast", check_escalation) workflow.add_conditional_edges("retrieval_deep", check_escalation) workflow.add_conditional_edges("action_executor", check_escalation) # Human escalation siempre termina workflow (humano toma control) workflow.add_edge("human_escalation", END) # Compile graph con checkpointer para persistence checkpointer = PostgresSaver.from_conn_string( os.getenv("POSTGRES_CONNECTION_STRING") ) app = workflow.compile(checkpointer=checkpointer) return app # Create graph instance customer_service_graph = create_customer_service_graph()