Creación de un sistema basado en agentes de IA con Langgraph: Agregar persistencia y transmisión (guía paso a paso)

En nuestro tutorial anterior, construimos un agente de IA capaz de responder consultas navegando por la web. Sin embargo, cuando construyen agentes para tareas de mayor duración, entran en juego dos conceptos críticos: persistencia y transmisión. La persistencia le permite salvar el estado de un agente en cualquier punto dado, lo que le permite reanudar de ese estado en futuras interacciones. Esto es crucial para aplicaciones de larga duración. Por otro lado, la transmisión le permite emitir señales en tiempo real sobre lo que el agente está haciendo en cualquier momento, proporcionando transparencia y control sobre sus acciones. En este tutorial, mejoraremos a nuestro agente agregando estas potentes características.

Configuración del agente

Comencemos por recrear a nuestro agente. Cargaremos las variables de entorno necesarias, instalaremos e importaremos las bibliotecas requeridas, configuraremos la herramienta de búsqueda tavily, definiremos el estado del agente y, finalmente, construiremos el agente.

pip install langgraph==0.2.53 langgraph-checkpoint==2.0.6 langgraph-sdk==0.1.36 langchain-groq langchain-community langgraph-checkpoint-sqlite==2.0.1

import os
os.environ['TAVILY_API_KEY'] = "<TAVILY_API_KEY>"
os.environ['GROQ_API_KEY'] = "<GROQ_API_KEY>"

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_groq import ChatGroq
from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=2)

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

class Agent:
    def __init__(self, model, tools, system=""):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        self.graph = graph.compile()
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    def call_openai(self, state: AgentState):
        messages = state['messages']
        if self.system:
            messages = [SystemMessage(content=self.system)] + messages
        message = self.model.invoke(messages)
        return {'messages': [message]}

    def exists_action(self, state: AgentState):
        result = state['messages'][-1]
        return len(result.tool_calls) > 0

    def take_action(self, state: AgentState):
        tool_calls = state['messages'][-1].tool_calls
        results = []
        for t in tool_calls:
            print(f"Calling: {t}")
            result = self.tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print("Back to the model!")
        return {'messages': results}

Agregar persistencia

Para agregar persistencia, usaremos Langgraph’s checkpointer característica. Un checkpointer ahorra el estado del agente después y entre cada nodo. Para este tutorial, usaremos Sqlitesaverun simple checkpointer que aprovecha SQLite, una base de datos incorporada. Si bien usaremos una base de datos en memoria por simplicidad, puede conectarla fácilmente a una base de datos externa o usar otros puntos de control como Redis o Postgres para una persistencia más robusta.

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
sqlite_conn = sqlite3.connect("checkpoints.sqlite",check_same_thread=False)
memory = SqliteSaver(sqlite_conn)

A continuación, modificaremos a nuestro agente para aceptar un checkpointer:

class Agent:
    def __init__(self, model, tools, checkpointer, system=""):
        # Everything else remains the same as before
        self.graph = graph.compile(checkpointer=checkpointer)
    # Everything else after this remains the same

Ahora, podemos crear nuestro agente con persistencia habilitada:

prompt = """You are a smart research assistant. Use the search engine to look up information. \
You are allowed to make multiple calls (either together or in sequence). \
Only look up information when you are sure of what you want. \
If you need to look up some information before asking a follow-up question, you are allowed to do that!
"""
model = ChatGroq(model="Llama-3.3-70b-Specdec")
bot = Agent(model, [tool], system=prompt, checkpointer=memory)

Agregar transmisión

La transmisión es esencial para las actualizaciones en tiempo real. Hay dos tipos de transmisión en los que nos centraremos:

1. Transmisión de mensajes: Emitir mensajes intermedios como decisiones de IA y resultados de la herramienta.

2. Tokens de transmisión: Transmitiendo tokens individuales de la respuesta de la LLM.
Comencemos por transmitir mensajes. Crearemos un mensaje humano y usaremos el arroyo Método para observar las acciones del agente en tiempo real.

messages = [HumanMessage(content="What is the weather in Texas?")]
thread = {"configurable": {"thread_id": "1"}}
for event in bot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v['messages'])

Salida final: el clima actual en Texas está soleado con una temperatura de 19.4 ° C (66.9 ° F) y una velocidad del viento de 4.3 mph (6.8 kph) …

Cuando ejecute esto, verá un flujo de resultados. Primero, un mensaje de IA instruye al agente que llame a Tavily, seguido de un mensaje de herramienta con los resultados de búsqueda, y finalmente, un mensaje de IA que responde la pregunta.

Comprender las identificaciones de hilo

El Thread_id es una parte crucial de la configuración del hilo. Permite al agente mantener conversaciones separadas con diferentes usuarios o contextos. Al asignar un hilo único a cada conversación, el agente puede realizar un seguimiento de múltiples interacciones simultáneamente sin mezclarlas.

Por ejemplo, continuemos la conversación preguntando: “¿Qué pasa en Los Ángeles?” Usando el mismo thread_id:

messages = [HumanMessage(content="What about in LA?")]
thread = {"configurable": {"thread_id": "1"}}
for event in bot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

Salida final: el clima actual en Los Ángeles está soleado con una temperatura de 17.2 ° C (63.0 ° F) y una velocidad del viento de 2.2 mph (3.6 kph) …

El agente infiere que estamos preguntando sobre el clima, gracias a la persistencia. Para verificar, preguntemos: “¿Cuál es más cálido?”:

messages = [HumanMessage(content="Which one is warmer?")]
thread = {"configurable": {"thread_id": "1"}}
for event in bot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

Salida final: Texas es más cálido que Los Ángeles. La temperatura actual en Texas es de 19.4 ° C (66.9 ° F), mientras que la temperatura actual en Los Ángeles es de 17.2 ° C (63.0 ° F)

El agente compara correctamente el clima en Texas y LA. Para probar si la persistencia mantiene las conversaciones separadas, hagamos la misma pregunta con un Thread_id:

messages = [HumanMessage(content="Which one is warmer?")]
thread = {"configurable": {"thread_id": "2"}}
for event in bot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

Salida: Necesito más información para responder esa pregunta. ¿Puede proporcionar más contexto o especificar qué dos cosas está comparando?

Esta vez, el agente se confunde porque no tiene acceso a la historia de la conversación anterior.

Tokens de transmisión

Para transmitir tokens, usaremos el Astream_events Método, que es asincrónico. También cambiaremos a un checkpointer Async.

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
    abot = Agent(model, [tool], system=prompt, checkpointer=checkpointer)
    messages = [HumanMessage(content="What is the weather in SF?")]
    thread = {"configurable": {"thread_id": "4"}}
    async for event in abot.graph.astream_events({"messages": messages}, thread, version="v1"):
        kind = event["event"]
        if kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                # Empty content in the context of OpenAI means
                # that the model is asking for a tool to be invoked.
                # So we only print non-empty content
                print(content, end="|")

Esto transmitirá tokens en tiempo real, dándole una visión en vivo del proceso de pensamiento del agente.

Conclusión

Al agregar persistencia y transmisión, hemos mejorado significativamente las capacidades de nuestro agente de IA. La persistencia permite al agente mantener el contexto a través de las interacciones, mientras que la transmisión proporciona información en tiempo real sobre sus acciones. Estas características son esenciales para construir aplicaciones listas para la producción, especialmente aquellas que involucran múltiples usuarios o interacciones humanas en el circuito.

En el próximo tutorial, nos sumergiremos en interacciones humanas en el bucledonde la persistencia juega un papel crucial en permitir una colaboración perfecta entre humanos y agentes de IA. ¡Manténganse al tanto!

Referencias:

1. (Deeplearning.ai) https://learn.deeplearning.ai/courses/ai-agents-in-langgraph

Además, no olvides seguirnos Gorjeo y únete a nuestro Canal de telegrama y LinkedIn GRsalpicar. No olvides unirte a nuestro 75k+ ml de subreddit.

🚨 Conocer Intellagent: Un marco de múltiples agentes de código abierto para evaluar un sistema de IA conversacional complejo ^{(Promocionado)}