Asistente personal LangChain con múltiples herramientas
En este artículo se construye un agente asistente personal que integra consulta meteorológica, gestión de agenda y envío de correos, mostrando el uso completo de la colaboración de múltiples herramientas y la salida estructurada.
Diseño del sistema
- Tres herramientas: consulta meteorológica, gestión de agenda, envío de correos
- Salida estructurada: resumen de agenda formateado como Markdown
- Salida en streaming: muestra en tiempo real el proceso de pensamiento y procesamiento de la IA
- Memoria de conversación: recuerda las preferencias y el contexto del usuario
Código completo
Antes de ejecutar, instala las dependencias y en.envconfigúralo correctamenteDEEPSEEK_API_KEY：
pip install langchain langchain-deepseek langgraph-checkpoint-sqlite python-dotenv
Ejemplo
# Ruta del archivo: personal_assistant.py
from dotenv import load_dotenv
load_dotenv()
import sqlite3
from datetime import datetime
from pydantic import BaseModel, Field
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver
# ========== 1. Datos simulados ==========
calendar_events = [
{"id": 1, "title": "Aprender Python", "date": "2024-03-25",
"time": "14:00", "duration": "2 horas"},
{"id": 2, "title": "Reunión semanal del equipo", "date": "2024-03-25",
"time": "10:00", "duration": "1 hora"},
{"id": 3, "title": "Revisión de código", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 horas"},
]
weather_db = {
"Hangzhou": {"condition": "Despejado", "temp": 25, "humidity": 60},
"Beijing": {"condition": "Nublado", "temp": 18, "humidity": 45},
"Shanghai": {"condition": "Lluvia ligera", "temp": 22, "humidity": 80},
}
# ========== 2. Definir herramientas ==========
@tool
def get_weather(city: str) -> str:
"""Consulta el clima en tiempo real de una ciudad especificada.
Args:
city: nombre de la ciudad, por ejemplo Hangzhou, Beijing, Shanghai
"""
data = weather_db.get(city)
if not data:
return f"No se puede consultar el clima de {city} por el momento. Ciudades compatibles: {', '.join(weather_db.keys())}"
return (f"Clima en {city}: {data['condition']}, "
f"Temperatura {data['temp']}°C, humedad {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Consulta la agenda para una fecha especificada. Si no se especifica fecha, consulta la agenda de hoy.
Args:
date: fecha, formato YYYY-MM-DD, por ejemplo 2024-03-25. Si no se pasa, consulta hoy
"""
if date is None:
date = datetime.now().strftime("%Y-%m-%d")
events = [e for e in calendar_events if e["date"] == date]
if not events:
return f"{date} no tiene eventos programados."
events.sort(key=lambda e: e["time"])
# Aquí se debe escribir el carácter emoji directamente, no como entidad HTML como 📅 —
# Esta parte es una cadena de Python, se imprimirá tal cual y no será interpretada por el navegador como un gráfico
lines = [f"📅 Agenda de {date}:"]
for e in events:
lines.append(f" - {e['time']} {e['title']}（{e['duration']}）")
return "\n".join(lines)
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Enviar correo electrónico (simulado).
Args:
to: correo del destinatario
subject: asunto del correo
body: cuerpo del correo
"""
# Simulación de envío
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"¡Correo enviado! Destinatario: {to}, Asunto: {subject}, ID del correo: {email_id}"
# ========== 3. Modelo de salida estructurada ==========
class DailySummary(BaseModel):
"""Resumen diario"""
date: str = Field(description="Fecha")
weather_summary: str = Field(description="Resumen del clima")
event_count: int = Field(description="Número de eventos de agenda")
key_events: list[str] = Field(description="Lista de eventos importantes")
suggestion: str = Field(description="Sugerencia de hoy")
def to_markdown(summary: DailySummary) -> str:
"""Convierte el DailySummary estructurado en texto Markdown,
corresponde al elemento "Salida estructurada: resumen de agenda formateado como Markdown" del diseño del sistema."""
lines = [
f"## Resumen diario de {summary.date}",
"",
f"- **Clima**: {summary.weather_summary}",
f"- **Número de eventos**: {summary.event_count}",
"",
"**Asuntos importantes:**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- Ninguno")
lines += ["", f"**Sugerencia de hoy**: {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Definir Middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Agrega dinámicamente la información de la fecha actual después del prompt del sistema.
La forma anterior de usar @before_model para insertar el mensaje de fecha en la lista messages tiene dos problemas:
1. La actualización de mensajes devuelta por before_model pasa por el reducer add_messages; el reducer solo
procesa "añadir nuevos mensajes al final", no reconoce la posición en la lista devuelta; insert(-1, ...)
la intención de insertarlo antes del mensaje del usuario realmente no tiene efecto;
2. before_model se activa repetidamente en el bucle de llamadas a herramientas de varias rondas, y es fácil que inserte este mensaje
entre AIMessage(tool_calls=...) y su ToolMessage correspondiente,
rompiendo el estricto requisito de orden de los mensajes.
En su lugar, se usa @dynamic_prompt para reescribir directamente la cadena del prompt del sistema; cada llamada al modelo
vuelve a calcularla, sin tocar la lista messages, sin riesgo de acumulación o desorden.
"""
now = datetime.now()
weekday = ["Lunes", "Martes", "Miércoles", "Jueves", "Viernes", "Sábado", "Domingo"][now.weekday()]
date_hint = (f"\n\n[Mensaje del sistema] La fecha actual es {now.strftime('%Yaño%mmes%ddía')},
f"día de la semana: {weekday}. Si el usuario no especifica una fecha, por defecto se consulta hoy.")
return request.system_prompt + date_hint
# ========== 5. Crear el Agent ==========
# Usa SqliteSaver para persistir la conversación, logrando la "memoria de conversación" del diseño del sistema.
# Establece la conexión tú mismo y luego pásala al constructor de SqliteSaver, en lugar de usar
# SqliteSaver.from_conn_string() (que es un administrador de contexto que solo sirve para usar con with,
# se cierra al terminar; consulta el artículo «LangChain robot de atención al cliente inteligente»).
conn = sqlite3.connect("personal_assistant.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0.3)
agent = create_agent(
model=model,
tools=[get_weather, query_schedule, send_email],
middleware=[inject_date_context],
response_format=DailySummary,
checkpointer=checkpointer,
system_prompt="""Eres un asistente personal "Xiao Zhu". Puedes consultar el clima, gestionar la agenda y enviar correos electrónicos.
## Forma de trabajo
1. Cuando el usuario pregunte "¿qué tal hoy?" o algo similar:
- Primero consulta el clima de hoy (get_weather)
- Luego consulta la agenda de hoy (query_schedule)
- Después genera el resumen diario
2. Cuando el usuario pida enviar un correo, usa la herramienta send_email
3. Cuando el usuario solo pregunte por el clima o solo por la agenda, llama solo a la herramienta correspondiente
## Estilo
- Tono amable y natural
- Prioriza el uso de herramientas para obtener datos en tiempo real, no inventes.""",
)
# ========== 6. Función de interacción ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Conversa con el asistente:
- Usa stream_mode="values" para mostrar en streaming el proceso de pensamiento y procesamiento de la IA (corresponde a "salida en streaming")
- Pasa thread_id; las conversaciones bajo el mismo thread_id son recordadas por SqliteSaver (corresponde a "memoria de conversación")
- Si en esta ronda se genera un resumen estructurado, se renderiza como Markdown y se imprime junto con el resto (corresponde a "salida estructurada")
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"Tú: {message}")
print(f"{'='*60}")
seen = 0
final_state = {}
for state in agent.stream(
{"messages": [HumanMessage(content=message)]},
config=config,
stream_mode="values",
):
final_state = state
msgs = state.get("messages", [])
# Cada vez que aparecen varios mensajes nuevos, significa que el Agent ha avanzado un paso; imprímelos en tiempo real
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 Decidió llamar a la herramienta: {call['name']}, parámetros: {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Herramienta"
print(f"🔧 {tool_name} devolvió: {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Asistente: {msg.content}")
seen = len(msgs)
# Nota: response_format se configura a nivel del Agent; cada llamada (incluyendo envío de correos,
# preguntas de seguimiento y otras solicitudes no relacionadas con el "resumen de agenda") intentará forzosamente generar un DailySummary,
# esta es una limitación conocida del response_format global. Si realmente se quiere un asistente multi-intención, una mejor práctica es
# convertir "generar resumen diario" en una herramienta separada, para que el Agent decida si invocarla o no,
# en lugar de acoplar a todo el Agent un esquema de salida que siempre está activo.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Resumen estructurado (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Pruebas ==========
if __name__ == "__main__":
chat(¿Cómo está el clima hoy en Hangzhou? Revisa mi agenda y dame un resumen de hoy)
chat(Ayúdame a enviar un correo a [email protected], con el asunto 'Resumen de hoy' y el contenido de que la agenda de hoy está confirmada)
# La tercera ronda no incluye información nueva, solo pone a prueba si el Agente recuerda la conversación anterior—
# Como se pasa el mismo thread_id, el checkpointer traerá de vuelta el historial completo
chat(El correo que te pedí enviar hace un momento, ¿cuál era el asunto?)
from dotenv import load_dotenv
load_dotenv()
import sqlite3
from datetime import datetime
from pydantic import BaseModel, Field
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langgraph.checkpoint.sqlite import SqliteSaver
# ========== 1. Datos simulados ==========
calendar_events = [
{"id": 1, "title": "Aprender Python", "date": "2024-03-25",
"time": "14:00", "duration": "2 horas"},
{"id": 2, "title": "Reunión semanal del equipo", "date": "2024-03-25",
"time": "10:00", "duration": "1 hora"},
{"id": 3, "title": "Revisión de código", "date": "2024-03-26",
"time": "15:00", "duration": "1.5 horas"},
]
weather_db = {
"Hangzhou": {"condition": "Despejado", "temp": 25, "humidity": 60},
"Beijing": {"condition": "Nublado", "temp": 18, "humidity": 45},
"Shanghai": {"condition": "Lluvia ligera", "temp": 22, "humidity": 80},
}
# ========== 2. Definir herramientas ==========
@tool
def get_weather(city: str) -> str:
"""Consulta el clima en tiempo real de una ciudad especificada.
Args:
city: nombre de la ciudad, por ejemplo Hangzhou, Beijing, Shanghai
"""
data = weather_db.get(city)
if not data:
return f"No se puede consultar el clima de {city} por el momento. Ciudades compatibles: {', '.join(weather_db.keys())}"
return (f"Clima en {city}: {data['condition']}, "
f"Temperatura {data['temp']}°C, humedad {data['humidity']}%")
@tool
def query_schedule(date: str = None) -> str:
"""Consulta la agenda para una fecha especificada. Si no se especifica fecha, consulta la agenda de hoy.
Args:
date: fecha, formato YYYY-MM-DD, por ejemplo 2024-03-25. Si no se pasa, consulta hoy
"""
if date is None:
date = datetime.now().strftime("%Y-%m-%d")
events = [e for e in calendar_events if e["date"] == date]
if not events:
return f"{date} no tiene eventos programados."
events.sort(key=lambda e: e["time"])
# Aquí se debe escribir el carácter emoji directamente, no como entidad HTML como 📅 —
# Esta parte es una cadena de Python, se imprimirá tal cual y no será interpretada por el navegador como un gráfico
lines = [f"📅 Agenda de {date}:"]
for e in events:
lines.append(f" - {e['time']} {e['title']}（{e['duration']}）")
return "\n".join(lines)
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Enviar correo electrónico (simulado).
Args:
to: correo del destinatario
subject: asunto del correo
body: cuerpo del correo
"""
# Simulación de envío
email_id = f"MSG-{datetime.now().strftime('%Y%m%d%H%M%S')}"
return f"¡Correo enviado! Destinatario: {to}, Asunto: {subject}, ID del correo: {email_id}"
# ========== 3. Modelo de salida estructurada ==========
class DailySummary(BaseModel):
"""Resumen diario"""
date: str = Field(description="Fecha")
weather_summary: str = Field(description="Resumen del clima")
event_count: int = Field(description="Número de eventos de agenda")
key_events: list[str] = Field(description="Lista de eventos importantes")
suggestion: str = Field(description="Sugerencia de hoy")
def to_markdown(summary: DailySummary) -> str:
"""Convierte el DailySummary estructurado en texto Markdown,
corresponde al elemento "Salida estructurada: resumen de agenda formateado como Markdown" del diseño del sistema."""
lines = [
f"## Resumen diario de {summary.date}",
"",
f"- **Clima**: {summary.weather_summary}",
f"- **Número de eventos**: {summary.event_count}",
"",
"**Asuntos importantes:**",
]
if summary.key_events:
lines += [f"- {event}" for event in summary.key_events]
else:
lines.append("- Ninguno")
lines += ["", f"**Sugerencia de hoy**: {summary.suggestion}"]
return "\n".join(lines)
# ========== 4. Definir Middleware ==========
@dynamic_prompt
def inject_date_context(request) -> str:
"""Agrega dinámicamente la información de la fecha actual después del prompt del sistema.
La forma anterior de usar @before_model para insertar el mensaje de fecha en la lista messages tiene dos problemas:
1. La actualización de mensajes devuelta por before_model pasa por el reducer add_messages; el reducer solo
procesa "añadir nuevos mensajes al final", no reconoce la posición en la lista devuelta; insert(-1, ...)
la intención de insertarlo antes del mensaje del usuario realmente no tiene efecto;
2. before_model se activa repetidamente en el bucle de llamadas a herramientas de varias rondas, y es fácil que inserte este mensaje
entre AIMessage(tool_calls=...) y su ToolMessage correspondiente,
rompiendo el estricto requisito de orden de los mensajes.
En su lugar, se usa @dynamic_prompt para reescribir directamente la cadena del prompt del sistema; cada llamada al modelo
vuelve a calcularla, sin tocar la lista messages, sin riesgo de acumulación o desorden.
"""
now = datetime.now()
weekday = ["Lunes", "Martes", "Miércoles", "Jueves", "Viernes", "Sábado", "Domingo"][now.weekday()]
date_hint = (f"\n\n[Mensaje del sistema] La fecha actual es {now.strftime('%Yaño%mmes%ddía')},
f"día de la semana: {weekday}. Si el usuario no especifica una fecha, por defecto se consulta hoy.")
return request.system_prompt + date_hint
# ========== 5. Crear el Agent ==========
# Usa SqliteSaver para persistir la conversación, logrando la "memoria de conversación" del diseño del sistema.
# Establece la conexión tú mismo y luego pásala al constructor de SqliteSaver, en lugar de usar
# SqliteSaver.from_conn_string() (que es un administrador de contexto que solo sirve para usar con with,
# se cierra al terminar; consulta el artículo «LangChain robot de atención al cliente inteligente»).
conn = sqlite3.connect("personal_assistant.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0.3)
agent = create_agent(
model=model,
tools=[get_weather, query_schedule, send_email],
middleware=[inject_date_context],
response_format=DailySummary,
checkpointer=checkpointer,
system_prompt="""Eres un asistente personal "Xiao Zhu". Puedes consultar el clima, gestionar la agenda y enviar correos electrónicos.
## Forma de trabajo
1. Cuando el usuario pregunte "¿qué tal hoy?" o algo similar:
- Primero consulta el clima de hoy (get_weather)
- Luego consulta la agenda de hoy (query_schedule)
- Después genera el resumen diario
2. Cuando el usuario pida enviar un correo, usa la herramienta send_email
3. Cuando el usuario solo pregunte por el clima o solo por la agenda, llama solo a la herramienta correspondiente
## Estilo
- Tono amable y natural
- Prioriza el uso de herramientas para obtener datos en tiempo real, no inventes.""",
)
# ========== 6. Función de interacción ==========
def chat(message: str, thread_id: str = "xiaoming"):
"""Conversa con el asistente:
- Usa stream_mode="values" para mostrar en streaming el proceso de pensamiento y procesamiento de la IA (corresponde a "salida en streaming")
- Pasa thread_id; las conversaciones bajo el mismo thread_id son recordadas por SqliteSaver (corresponde a "memoria de conversación")
- Si en esta ronda se genera un resumen estructurado, se renderiza como Markdown y se imprime junto con el resto (corresponde a "salida estructurada")
"""
config = {"configurable": {"thread_id": thread_id}}
print(f"\n{'='*60}")
print(f"Tú: {message}")
print(f"{'='*60}")
seen = 0
final_state = {}
for state in agent.stream(
{"messages": [HumanMessage(content=message)]},
config=config,
stream_mode="values",
):
final_state = state
msgs = state.get("messages", [])
# Cada vez que aparecen varios mensajes nuevos, significa que el Agent ha avanzado un paso; imprímelos en tiempo real
for msg in msgs[seen:]:
if msg.type == "ai" and getattr(msg, "tool_calls", None):
for call in msg.tool_calls:
print(f"🤔 Decidió llamar a la herramienta: {call['name']}, parámetros: {call['args']}")
elif msg.type == "tool":
tool_name = getattr(msg, "name", "") or "Herramienta"
print(f"🔧 {tool_name} devolvió: {str(msg.content)[:80]}")
elif msg.type == "ai" and msg.content:
print(f"🤖 Asistente: {msg.content}")
seen = len(msgs)
# Nota: response_format se configura a nivel del Agent; cada llamada (incluyendo envío de correos,
# preguntas de seguimiento y otras solicitudes no relacionadas con el "resumen de agenda") intentará forzosamente generar un DailySummary,
# esta es una limitación conocida del response_format global. Si realmente se quiere un asistente multi-intención, una mejor práctica es
# convertir "generar resumen diario" en una herramienta separada, para que el Agent decida si invocarla o no,
# en lugar de acoplar a todo el Agent un esquema de salida que siempre está activo.
if "structured_response" in final_state:
summary = final_state["structured_response"]
print("\n--- Resumen estructurado (Markdown) ---)
print(to_markdown(summary))
return final_state
# ========== 7. Pruebas ==========
if __name__ == "__main__":
chat(¿Cómo está el clima hoy en Hangzhou? Revisa mi agenda y dame un resumen de hoy)
chat(Ayúdame a enviar un correo a [email protected], con el asunto 'Resumen de hoy' y el contenido de que la agenda de hoy está confirmada)
# La tercera ronda no incluye información nueva, solo pone a prueba si el Agente recuerda la conversación anterior—
# Como se pasa el mismo thread_id, el checkpointer traerá de vuelta el historial completo
chat(El correo que te pedí enviar hace un momento, ¿cuál era el asunto?)
Resultado de la ejecución:
============================================================
Tú: 杭州今天天气怎么样？看看我的日程，然后给我一个今日总结
============================================================
🤔 决定调用工具: get_weather，参数: {'city': '杭州'}
🔧 get_weather 返回: 杭州天气：晴，温度 25°C，湿度 60%
🤔 决定调用工具: query_schedule，参数: {}
🔧 query_schedule 返回: 📅 2024-03-25 日程安排： - 10:00 团队周会（1小时） - 14:00...
🤖 助手: 早上好！今天杭州是大晴天，气温 25°C，湿度 60%，很适合出门活动。
今天您有两项日程：上午 10:00 的团队周会和下午 14:00 的 Python 学习，祝您今天顺利！
--- 结构化摘要（Markdown） ---
## 2024-03-25 今日摘要
- **天气**：杭州晴，25°C，湿度 60%
- **日程数量**：2
**重要事项：**
- 10:00 团队周会
- 14:00 Python 学习
**今日建议**：上午先参加团队周会，下午集中精力学习 Python，注意劳逸结合
============================================================
Tú: 帮我发一封邮件给 [email protected]，主题是'今日总结'，内容是今天日程已确认
============================================================
🤔 决定调用工具: send_email，参数: {'to': '[email protected]', 'subject': '今日总结', 'body': '今天日程已确认'}
🔧 send_email 返回: 邮件已发送！收件人：[email protected]，主题：今日总结，邮件ID：MSG-20240325143000
🤖 助手: 邮件已经发送成功啦！收件人是 [email protected]，主题"今日总结"。
--- 结构化摘要（Markdown） ---
## 2024-03-25 今日摘要
...（这里的摘要和邮件内容其实没什么关系，是 response_format 强制生成的，
属于前面提到的已知局限）
============================================================
Tú: 我刚才让你发的那封邮件，主题是什么来着？
============================================================
🤖 助手: 你刚才让我发的那封邮件主题是"今日总结"，收件人是 [email protected].
La tercera ronda no activó ninguna llamada a herramientas; el asistente pudo responder simplemente porque
thread_ides el mismo, SqliteSaver trajo de vuelta el historial completo de mensajes de las dos primeras rondas—así es como la "memoria conversacional" realmente funciona. Si se cambiathread_idpor un valor nuevo, la tercera ronda se convertirá en una conversación aislada y el asistente no sabrá qué sucedió "hace un momento".
Resumen del proyecto
Este asistente personal demuestra:
| Característica | Método de implementación |
|---|---|
| Colaboración multiherramienta | Clima + agenda + correo, el Agente selecciona automáticamente el orden de las llamadas |
| Salida estructurada | Modelo DailySummary de Pydantic + to_markdown() renderizado como texto Markdown |
| Salida en streaming | agent.stream(stream_mode="values"), muestra paso a paso las llamadas a herramientas y el proceso de razonamiento |
| Memoria conversacional | SqliteSaver + thread_id, recuerda el contexto a través de múltiples rondas de llamadas |
| Inyección de fecha | @dynamic_prompt reescribe dinámicamente el mensaje del sistema, evitando contaminar el historial de mensajes |
| Interacción en lenguaje natural | El usuario describe sus necesidades en lenguaje natural y el Agente planifica de forma autónoma |