Bot de atención al cliente con LangChain
Este capítulo integra los conocimientos aprendidos anteriormente para construir un bot completo de atención al cliente. Puede consultar la base de conocimiento, gestionar pedidos y transferir la conversación a un agente humano cuando sea necesario.
Análisis de requisitos
| Funcionalidad | Método de implementación |
|---|---|
| Preguntas y respuestas de la base de conocimiento | Recuperación RAG + respuesta del modelo |
| Consulta de pedidos | Función de herramienta @tool |
| Memoria de conversación | SqliteSaver Checkpointer |
| Filtrado de contenido sensible | @before_model Middleware |
| Transferencia a agente humano | HITL interrupt() / Command(resume=...) |
Configuración del entorno
Antes de escribir el código, prepara el entorno de ejecución. Sigue los siguientes pasos y lo tendrás listo en unos minutos.
Paso 1: Verificar la versión de Python
Se recomienda usar Python 3.10 o superior. Abre la terminal y escribe:
python --version
Si la versión es inferior a 3.10, se recomienda actualizar Python antes de continuar con los siguientes pasos.
Paso 2: Crear y activar un entorno virtual (recomendado)
Para no contaminar el entorno Python del sistema, se recomienda crear un entorno virtual separado para este proyecto:
# 创建虚拟环境 python -m venv venv # 激活虚拟环境（Windows） venv\Scripts\activate # 激活虚拟环境（macOS / Linux） source venv/bin/activate
Una vez activado correctamente, aparecerá delante del prompt de la terminal(venv)la marca.
Paso 3: Instalar los paquetes de dependencias
Este proyecto utiliza 6 bibliotecas de terceros, cada una con una función distinta. Se recomienda instalarlas una por una en el siguiente orden, para que cuando surja un problema sea más fácil identificar qué biblioteca no se instaló correctamente.
| Comando de instalación | Función |
|---|---|
pip install langchain | Biblioteca principal de LangChain, proporciona capacidades básicas como create_agent, @tool, etc. |
pip install langchain-deepseek | Permite que init_chat_model reconozca el prefijo "deepseek:" e invoque el modelo DeepSeek. |
pip install langchain-openai | Proporciona OpenAIEmbeddings para convertir el texto de la base de conocimiento en vectores. |
pip install langchain-chroma chromadb | Base de datos vectorial local para almacenar y recuperar los vectores de la base de conocimiento. |
pip install langgraph-checkpoint-sqlite | SqliteSaver, persiste la memoria de conversación en un archivo SQLite. |
pip install python-dotenv | Lee variables de entorno como las claves de API desde el archivo .env. |
También se pueden instalar todos con un solo comando:
pip install langchain langchain-deepseek langchain-openai langchain-chroma chromadb langgraph-checkpoint-sqlite python-dotenv
Paso 4: Configurar las claves de API
En el directorio raíz del proyecto, crea un archivo llamado.env(ten en cuenta que el nombre del archivo comienza con un punto y no tiene extensión), y añade el siguiente contenido:
# DeepSeek 官网申请：https://platform.deepseek.com DEEPSEEK_API_KEY=sk-你的deepseek密钥 # OpenAI 官网申请：https://platform.openai.com # 这里只用来调用 embedding 接口，不涉及 Chat 模型 OPENAI_API_KEY=sk-你的openai密钥
Si no tienes una clave de OpenAI, podemos usar la de Alibaba Bailian,.envel código del archivo es el siguiente:
# DeepSeek 官网申请：https://platform.deepseek.com DEEPSEEK_API_KEY=sk-你的deepseek密钥 # 阿里云百炼控制台申请：https://bailian.console.aliyun.com # 这里用来调用通义千问的 Embedding 服务，给知识库文本做向量化 DASHSCOPE_API_KEY=sk-你的百炼密钥
El archivo .env contiene claves privadas. No lo subas a un repositorio Git ni lo compartas con nadie. En el proyecto, puedes crear un archivo
.gitignorey añadir una línea.envpara evitar un commit accidental.
Paso 5: Verificar la instalación
Crea un archivocheck_install.pyy ejecuta el siguiente script para comprobar si las dependencias y las claves están configuradas correctamente:
Ejemplo
# Archivo: check_install.py
import os
from dotenv import load_dotenv
load_dotenv()
# Comprueba si los paquetes de dependencias se pueden importar correctamente
import langchain
import langchain_deepseek
import langchain_openai
import langchain_chroma
import chromadb
import langgraph
print(f"Versión de langchain: {langchain.__version__}")
# Comprueba si las claves están configuradas
assert os.getenv("DEEPSEEK_API_KEY"), "No se detectó DEEPSEEK_API_KEY, revisa el archivo .env"
# assert os.getenv("OPENAI_API_KEY"), "No se detectó OPENAI_API_KEY, revisa el archivo .env"
assert os.getenv("DASHSCOPE_API_KEY"), "No se detectó DASHSCOPE_API_KEY, revisa el archivo .env"
print("Configuración del entorno exitosa ~ ¡Ya puedes empezar a escribir el bot de atención al cliente!")
import os
from dotenv import load_dotenv
load_dotenv()
# Comprueba si los paquetes de dependencias se pueden importar correctamente
import langchain
import langchain_deepseek
import langchain_openai
import langchain_chroma
import chromadb
import langgraph
print(f"Versión de langchain: {langchain.__version__}")
# Comprueba si las claves están configuradas
assert os.getenv("DEEPSEEK_API_KEY"), "No se detectó DEEPSEEK_API_KEY, revisa el archivo .env"
# assert os.getenv("OPENAI_API_KEY"), "No se detectó OPENAI_API_KEY, revisa el archivo .env"
assert os.getenv("DASHSCOPE_API_KEY"), "No se detectó DASHSCOPE_API_KEY, revisa el archivo .env"
print("Configuración del entorno exitosa ~ ¡Ya puedes empezar a escribir el bot de atención al cliente!")
Ejecuta:
python check_install.py
Si las claves están configuradas correctamente, la salida es la siguiente:
langchain 版本: 1.3.0 环境配置成功~可以开始写客服机器人了！
Si aparece un error, normalmente se debe a que falta un paquete o a que las claves en el archivo .env no son correctas. Según el mensaje de error, vuelve al paso correspondiente para revisarlo.
Código completo
Una vez completada la configuración del entorno (instalación de dependencias y configuración de .env, ver la sección anterior "Configuración del entorno"), puedes escribir el código completo del bot de atención al cliente:
Ejemplo
# Archivo: customer_service_bot.py
# Instalación de dependencias y configuración de .env en la sección anterior "Configuración del entorno"
from dotenv import load_dotenv
load_dotenv()
import os
import sqlite3
from typing import Annotated
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, AIMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command
# ========== 1. Preparar la base de conocimiento ==========
knowledge_base = [
"El tutorial EJEMPLO fue fundado en 2013 y es una plataforma líder en China para el aprendizaje gratuito de programación.",
"La plataforma ofrece más de 300 tutoriales, que cubren Python, Java, HTML, CSS, JavaScript, entre otros.",
"El tutorial básico de Python3 tiene 30 capítulos y ha superado los 5 millones de visitas de aprendizaje. El curso es completamente gratuito.",
"La membresía VIP cuesta ¥99/mes, ¥799/año, e incluye cursos en video y servicio de preguntas y respuestas uno a uno.",
"Política de reembolso: se puede reembolsar el importe completo dentro de los 7 días posteriores a la compra y antes de completar 3 lecciones.",
"La plataforma admite un entorno de programación en línea, no es necesario instalar ningún software para escribir y ejecutar código.",
"Horario de atención al cliente: lunes a viernes de 9:00 a 18:00, fines de semana de 10:00 a 16:00.",
]
# Utiliza el servicio de embeddings Qwen de Alibaba Cloud Bailian (DashScope)
# La API de embeddings de Bailian es compatible con la especificación de OpenAI, por lo que se usa directamente langchain-openai
# OpenAIEmbeddings, simplemente apuntando base_url al endpoint compatible de Bailian,
# no es necesario instalar langchain-community / dashscope (ese paquete ya no recibe mantenimiento).
# text-embedding-v4 es el modelo vectorial general recomendado actualmente, con una salida predeterminada de vectores de 1024 dimensiones.
#
# Dos parámetros clave no pueden faltar:
# - check_embedding_ctx_length=False: OpenAIEmbeddings usa tiktoken por defecto
# para codificar el texto en una matriz de IDs de token antes de enviarlo (la API oficial de OpenAI acepta este formato),
# pero la interfaz compatible de Bailian solo acepta cadenas de texto sin procesar; si no se desactiva esta opción, se produce el error
# "contents is neither str nor list of str".
# - chunk_size=10: la API de embeddings de Bailian acepta como máximo 10 textos por solicitud,
# OpenAIEmbeddings agrupa 1000 por defecto; si la base de conocimiento es un poco grande, se superará el límite y se producirá un error.
embeddings = OpenAIEmbeddings(
model="text-embedding-v4",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
check_embedding_ctx_length=False,
chunk_size=10,
)
chunks = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30
).create_documents(knowledge_base)
vector_store = Chroma.from_documents(chunks, embeddings)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})
# ========== 2. Definir herramientas ==========
@tool
def search_kb(query: str) -> str:
"""Busca en la base de conocimiento del tutorial EJEMPLO para obtener información oficial sobre la plataforma, los cursos, las políticas, etc.
Args:
query: pregunta de búsqueda o palabra clave
"""
docs = retriever.invoke(query)
if not docs:
return "No se encontró información relevante; se recomienda transferir a un agente humano."
return "\n".join(f"- {doc.page_content}" for doc in docs)
# Base de datos de pedidos simulada
orders_db = {
"ORD-2024-001": {"user": "Xiaoming", "item": "Membresía anual VIP",
"amount": 799, "status": "Completado", "date": "2024-01-15"},
"ORD-2024-002": {"user": "Xiaoming", "item": "Curso práctico de Python",
"amount": 199, "status": "En envío", "date": "2024-03-20"},
}
@tool
def query_order(order_id: str) -> str:
"""Consulta el estado y los detalles de un pedido según su número.
Args:
order_id: número de pedido, por ejemplo, ORD-2024-001
"""
order = orders_db.get(order_id.upper())
if not order:
return f"No se encontró el pedido {order_id}. Por favor, verifica que el número sea correcto."
return (f"Pedido {order_id}: {order['item']} | "
f"Importe ¥{order['amount']} | "
f"Estado {order['status']} | "
f"Fecha {order['date']}")
@tool
def transfer_to_human(reason: str) -> str:
"""Transfiere al usuario a un agente de atención al cliente humano.
Args:
reason: motivo de la transferencia
"""
approval = interrupt({
"action": "transfer_to_human",
"reason": reason,
"message": f"El usuario solicita transferencia a un agente humano, motivo: {reason}. ¿Desea transferir?"
})
if approval.get("confirmed"):
return (f"Se le ha transferido a un agente humano, tiempo de espera estimado: {approval.get('wait_time', 3)} minutos."
f"Número de ticket: TK-{approval.get('ticket_id', 'N/A')}")
return "La transferencia ha sido cancelada, continúo sirviéndole."
# ========== 3. Definir Middleware ==========
@before_model
def content_guard(state, runtime):
"""Filtrar contenido inapropiado en la entrada del usuario"""
last_msg = state["messages"][-1] if state.get("messages") else None
if not last_msg:
return None
content = str(getattr(last_msg, 'content', ''))
blocked = ["porno X", "X apuestas", "ilegal"]
for word in blocked:
if word in content:
return {
"jump_to": "end",
"messages": [HumanMessage(content="Lo siento, no puedo procesar esta solicitud.")]
}
return None
@after_model
def auto_signature(state, runtime):
"""Agregar automáticamente la firma del agente de servicio al cliente"""
msgs = state.get("messages", [])
if not msgs:
return None
last = msgs[-1]
if last.type == "ai" and last.content and not (
hasattr(last, 'tool_calls') and last.tool_calls
):
# Clave: reutilizar last.id para que el reducer add_messages reemplace ese mensaje in situ,
# en lugar de tratarlo como un nuevo mensaje agregado al historial (de lo contrario, el historial crecerá cada vez más,
# cada ronda agregará una versión "sin firma" y una versión "con firma")
return {"messages": [AIMessage(
id=last.id,
content=last.content
+ "\n\n---\nCentro de atención al cliente de EJEMPLO (Tutorial para novatos) | Horario de trabajo 9:00-18:00"
)]}
return None
# ========== 4. Crear el Agent ==========
# SqliteSaver.from_conn_string() devuelve un administrador de contexto, solo adecuado para "cerrar después de usar"
# guiones de una sola vez. El robot de atención al cliente necesita mantener la misma conexión de base de datos entre múltiples llamadas a chat(),
# por lo que aquí se establece la conexión manualmente y se pasa al constructor de SqliteSaver.
# check_same_thread=False es porque los frameworks web suelen llamar a la misma conexión entre hilos.
conn = sqlite3.connect("customer_service.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_kb, query_order, transfer_to_human],
middleware=[content_guard, auto_signature],
checkpointer=checkpointer,
system_prompt="""Eres el asistente inteligente "Xiao Cai" de EJEMPLO (Tutorial para novatos).
## Tus responsabilidades
1. Atiende calurosamente a cada usuario, usa "usted" para dirigirte
2. Para preguntas sobre información de la plataforma, contenido de cursos, políticas, usa search_kb para consultar
3. Para consultas de pedidos, usa la herramienta query_order
4. Si encuentras problemas que no puedes resolver, usa transfer_to_human para transferir a un agente humano
## Código de conducta
- Responde de forma concisa, cada vez 2-3 frases
- Si no lo sabes, consulta la base de conocimiento; si no lo encuentras, dilo honestamente
- Mantén un tono amable y cordial""",
)
# ========== 5. Interfaz de conversación ==========
def chat(thread_id: str, message: str) -> str:
"""Procesa los mensajes del usuario y devuelve una respuesta"""
config = {"configurable": {"thread_id": thread_id}}
# Ejecutar Agent
result = agent.invoke(
{"messages": [HumanMessage(content=message)]},
config=config,
)
# Comprobar si se necesita transferencia (HITL)
state = agent.get_state(config)
if state.tasks and state.tasks[0].interrupts:
interrupt_info = state.tasks[0].interrupts[0].value
return f"[Requiere aprobación] {interrupt_info.get('message', '')}"
return result["messages"][-1].content
def resume_transfer(thread_id: str, confirmed: bool,
wait_time: int = 3, ticket_id: str = "0001") -> str:
"""Después de que el agente humano apruebe en el backend, se reanuda el flujo de transferencia interrumpido por interrupt().
Corresponde a los datos de approval que esperan en la herramienta transfer_to_human.
"""
config = {"configurable": {"thread_id": thread_id}}
result = agent.invoke(
Command(resume={
"confirmed": confirmed,
"wait_time": wait_time,
"ticket_id": ticket_id,
}),
config=config,
)
return result["messages"][-1].content
# ========== 6. Pruebas ==========
if __name__ == "__main__":
user_id = "user_xiaoming"
print("=== Prueba 1: Consulta a la base de conocimiento ===")
print(chat(user_id, "¿Cuántos capítulos tiene el tutorial de Python3?"))
print()
print("=== Prueba 2: Consulta de pedidos ===")
print(chat(user_id, "¿Cuál es el estado de mi pedido ORD-2024-001?"))
print()
print("=== Prueba 3: Consulta VIP ===")
print(chat(user_id, "¿Cuánto cuesta la membresía VIP?"))
print()
print("=== Prueba 4: Prueba de memoria ===")
print(chat(user_id, "¿Qué pregunta hice hace un momento?"))
print()
print("=== Prueba 5: Transferencia a humano (HITL) ===")
print(chat(user_id, "Quiero presentar una queja, transfiera a un agente humano")) # Activar interrupt, esperar aprobación
print(resume_transfer(user_id, confirmed=True,
wait_time=5, ticket_id="8823")) # Reanudar después de la confirmación en el backend
conn.close()
# Instalación de dependencias y configuración de .env en la sección anterior "Configuración del entorno"
from dotenv import load_dotenv
load_dotenv()
import os
import sqlite3
from typing import Annotated
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, AIMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command
# ========== 1. Preparar la base de conocimiento ==========
knowledge_base = [
"El tutorial EJEMPLO fue fundado en 2013 y es una plataforma líder en China para el aprendizaje gratuito de programación.",
"La plataforma ofrece más de 300 tutoriales, que cubren Python, Java, HTML, CSS, JavaScript, entre otros.",
"El tutorial básico de Python3 tiene 30 capítulos y ha superado los 5 millones de visitas de aprendizaje. El curso es completamente gratuito.",
"La membresía VIP cuesta ¥99/mes, ¥799/año, e incluye cursos en video y servicio de preguntas y respuestas uno a uno.",
"Política de reembolso: se puede reembolsar el importe completo dentro de los 7 días posteriores a la compra y antes de completar 3 lecciones.",
"La plataforma admite un entorno de programación en línea, no es necesario instalar ningún software para escribir y ejecutar código.",
"Horario de atención al cliente: lunes a viernes de 9:00 a 18:00, fines de semana de 10:00 a 16:00.",
]
# Utiliza el servicio de embeddings Qwen de Alibaba Cloud Bailian (DashScope)
# La API de embeddings de Bailian es compatible con la especificación de OpenAI, por lo que se usa directamente langchain-openai
# OpenAIEmbeddings, simplemente apuntando base_url al endpoint compatible de Bailian,
# no es necesario instalar langchain-community / dashscope (ese paquete ya no recibe mantenimiento).
# text-embedding-v4 es el modelo vectorial general recomendado actualmente, con una salida predeterminada de vectores de 1024 dimensiones.
#
# Dos parámetros clave no pueden faltar:
# - check_embedding_ctx_length=False: OpenAIEmbeddings usa tiktoken por defecto
# para codificar el texto en una matriz de IDs de token antes de enviarlo (la API oficial de OpenAI acepta este formato),
# pero la interfaz compatible de Bailian solo acepta cadenas de texto sin procesar; si no se desactiva esta opción, se produce el error
# "contents is neither str nor list of str".
# - chunk_size=10: la API de embeddings de Bailian acepta como máximo 10 textos por solicitud,
# OpenAIEmbeddings agrupa 1000 por defecto; si la base de conocimiento es un poco grande, se superará el límite y se producirá un error.
embeddings = OpenAIEmbeddings(
model="text-embedding-v4",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
check_embedding_ctx_length=False,
chunk_size=10,
)
chunks = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30
).create_documents(knowledge_base)
vector_store = Chroma.from_documents(chunks, embeddings)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})
# ========== 2. Definir herramientas ==========
@tool
def search_kb(query: str) -> str:
"""Busca en la base de conocimiento del tutorial EJEMPLO para obtener información oficial sobre la plataforma, los cursos, las políticas, etc.
Args:
query: pregunta de búsqueda o palabra clave
"""
docs = retriever.invoke(query)
if not docs:
return "No se encontró información relevante; se recomienda transferir a un agente humano."
return "\n".join(f"- {doc.page_content}" for doc in docs)
# Base de datos de pedidos simulada
orders_db = {
"ORD-2024-001": {"user": "Xiaoming", "item": "Membresía anual VIP",
"amount": 799, "status": "Completado", "date": "2024-01-15"},
"ORD-2024-002": {"user": "Xiaoming", "item": "Curso práctico de Python",
"amount": 199, "status": "En envío", "date": "2024-03-20"},
}
@tool
def query_order(order_id: str) -> str:
"""Consulta el estado y los detalles de un pedido según su número.
Args:
order_id: número de pedido, por ejemplo, ORD-2024-001
"""
order = orders_db.get(order_id.upper())
if not order:
return f"No se encontró el pedido {order_id}. Por favor, verifica que el número sea correcto."
return (f"Pedido {order_id}: {order['item']} | "
f"Importe ¥{order['amount']} | "
f"Estado {order['status']} | "
f"Fecha {order['date']}")
@tool
def transfer_to_human(reason: str) -> str:
"""Transfiere al usuario a un agente de atención al cliente humano.
Args:
reason: motivo de la transferencia
"""
approval = interrupt({
"action": "transfer_to_human",
"reason": reason,
"message": f"El usuario solicita transferencia a un agente humano, motivo: {reason}. ¿Desea transferir?"
})
if approval.get("confirmed"):
return (f"Se le ha transferido a un agente humano, tiempo de espera estimado: {approval.get('wait_time', 3)} minutos."
f"Número de ticket: TK-{approval.get('ticket_id', 'N/A')}")
return "La transferencia ha sido cancelada, continúo sirviéndole."
# ========== 3. Definir Middleware ==========
@before_model
def content_guard(state, runtime):
"""Filtrar contenido inapropiado en la entrada del usuario"""
last_msg = state["messages"][-1] if state.get("messages") else None
if not last_msg:
return None
content = str(getattr(last_msg, 'content', ''))
blocked = ["porno X", "X apuestas", "ilegal"]
for word in blocked:
if word in content:
return {
"jump_to": "end",
"messages": [HumanMessage(content="Lo siento, no puedo procesar esta solicitud.")]
}
return None
@after_model
def auto_signature(state, runtime):
"""Agregar automáticamente la firma del agente de servicio al cliente"""
msgs = state.get("messages", [])
if not msgs:
return None
last = msgs[-1]
if last.type == "ai" and last.content and not (
hasattr(last, 'tool_calls') and last.tool_calls
):
# Clave: reutilizar last.id para que el reducer add_messages reemplace ese mensaje in situ,
# en lugar de tratarlo como un nuevo mensaje agregado al historial (de lo contrario, el historial crecerá cada vez más,
# cada ronda agregará una versión "sin firma" y una versión "con firma")
return {"messages": [AIMessage(
id=last.id,
content=last.content
+ "\n\n---\nCentro de atención al cliente de EJEMPLO (Tutorial para novatos) | Horario de trabajo 9:00-18:00"
)]}
return None
# ========== 4. Crear el Agent ==========
# SqliteSaver.from_conn_string() devuelve un administrador de contexto, solo adecuado para "cerrar después de usar"
# guiones de una sola vez. El robot de atención al cliente necesita mantener la misma conexión de base de datos entre múltiples llamadas a chat(),
# por lo que aquí se establece la conexión manualmente y se pasa al constructor de SqliteSaver.
# check_same_thread=False es porque los frameworks web suelen llamar a la misma conexión entre hilos.
conn = sqlite3.connect("customer_service.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_kb, query_order, transfer_to_human],
middleware=[content_guard, auto_signature],
checkpointer=checkpointer,
system_prompt="""Eres el asistente inteligente "Xiao Cai" de EJEMPLO (Tutorial para novatos).
## Tus responsabilidades
1. Atiende calurosamente a cada usuario, usa "usted" para dirigirte
2. Para preguntas sobre información de la plataforma, contenido de cursos, políticas, usa search_kb para consultar
3. Para consultas de pedidos, usa la herramienta query_order
4. Si encuentras problemas que no puedes resolver, usa transfer_to_human para transferir a un agente humano
## Código de conducta
- Responde de forma concisa, cada vez 2-3 frases
- Si no lo sabes, consulta la base de conocimiento; si no lo encuentras, dilo honestamente
- Mantén un tono amable y cordial""",
)
# ========== 5. Interfaz de conversación ==========
def chat(thread_id: str, message: str) -> str:
"""Procesa los mensajes del usuario y devuelve una respuesta"""
config = {"configurable": {"thread_id": thread_id}}
# Ejecutar Agent
result = agent.invoke(
{"messages": [HumanMessage(content=message)]},
config=config,
)
# Comprobar si se necesita transferencia (HITL)
state = agent.get_state(config)
if state.tasks and state.tasks[0].interrupts:
interrupt_info = state.tasks[0].interrupts[0].value
return f"[Requiere aprobación] {interrupt_info.get('message', '')}"
return result["messages"][-1].content
def resume_transfer(thread_id: str, confirmed: bool,
wait_time: int = 3, ticket_id: str = "0001") -> str:
"""Después de que el agente humano apruebe en el backend, se reanuda el flujo de transferencia interrumpido por interrupt().
Corresponde a los datos de approval que esperan en la herramienta transfer_to_human.
"""
config = {"configurable": {"thread_id": thread_id}}
result = agent.invoke(
Command(resume={
"confirmed": confirmed,
"wait_time": wait_time,
"ticket_id": ticket_id,
}),
config=config,
)
return result["messages"][-1].content
# ========== 6. Pruebas ==========
if __name__ == "__main__":
user_id = "user_xiaoming"
print("=== Prueba 1: Consulta a la base de conocimiento ===")
print(chat(user_id, "¿Cuántos capítulos tiene el tutorial de Python3?"))
print()
print("=== Prueba 2: Consulta de pedidos ===")
print(chat(user_id, "¿Cuál es el estado de mi pedido ORD-2024-001?"))
print()
print("=== Prueba 3: Consulta VIP ===")
print(chat(user_id, "¿Cuánto cuesta la membresía VIP?"))
print()
print("=== Prueba 4: Prueba de memoria ===")
print(chat(user_id, "¿Qué pregunta hice hace un momento?"))
print()
print("=== Prueba 5: Transferencia a humano (HITL) ===")
print(chat(user_id, "Quiero presentar una queja, transfiera a un agente humano")) # Activar interrupt, esperar aprobación
print(resume_transfer(user_id, confirmed=True,
wait_time=5, ticket_id="8823")) # Reanudar después de la confirmación en el backend
conn.close()
Resultado de la ejecución:
=== 测试 1：知识库查询 === 您好！Ejemplo的 **Python3 基础教程共 30 章**，完全免费，累计学习人次已超 500 万哦！如需了解其他教程，也可以随时问我～ --- Ejemplo 客服中心 | 工作时间 9:00-18:00 === 测试 2：订单查询 === 您好！您查询的订单 **ORD-2024-001** 状态为 **已完成**。订单内容是 **VIP 年费会员**，金额 **¥799**，下单日期为 **2024-01-15**。请问还有什么可以帮您的吗？ --- Ejemplo 客服中心 | 工作时间 9:00-18:00 === 测试 3：VIP 咨询 === 您好！VIP 会员有 **¥99/月** 和 **¥799/年** 两种套餐，包含视频课程和一对一答疑服务哦～请问您需要办理哪种呢？ --- Ejemplo 客服中心 | 工作时间 9:00-18:00 --- Ejemplo 客服中心 | 工作时间 9:00-18:00 === 测试 4：测试记忆 === 您好！您刚才问了以下三个问题： 1. **Python3 教程有多少章？** —— 共 30 章，完全免费 2. **我的订单 ORD-2024-001 状态是什么？** —— 状态为"已完成" 3. **VIP 会员多少钱？** —— ¥99/月 o ¥799/年 还有什么需要我帮忙的吗？ --- Ejemplo 客服中心 | 工作时间 9:00-18:00 --- Ejemplo 客服中心 | 工作时间 9:00-18:00 === 测试 5：人工转接（HITL） === [需要审批] 用户请求转接人工客服，原因：用户要求投诉，需要转接人工客服处理。是否转接？ 您好！已经为您转接人工客服，预计等待约 5 分钟。您的工单号是 **TK-8823**，请稍候，客服会尽快为您处理～ --- Ejemplo 客服中心 | 工作时间 9:00-18:00 --- Ejemplo 客服中心 | 工作时间 9:00-18:00
Resumen del proyecto
Este robot de atención al cliente integra las siguientes características de LangChain:
| Característica | Uso en el proyecto |
|---|---|
| Búsqueda RAG | Herramienta search_kb + almacenamiento vectorial Chroma |
| Llamada a herramientas | query_order、transfer_to_human |
| Checkpointer | SqliteSaver persiste las conversaciones, logrando memoria de múltiples turnos |
| Middleware | before_model filtro de contenido + after_model agregar firma (reutilizar id de mensaje para reemplazar in situ) |
| HITL | interrupt() pausa la ejecución + Command(resume=...) reanuda después de la aprobación, logrando un ciclo completo de transferencia humana |