Criando um programa de agente de interface do usuário com Langgraph: Adicionando persistência e distribuição (guia passo a passo)

Em nosso estudo anterior, criamos um agente de IA que pode responder a perguntas através da web. No entanto, quando os agentes são construídos em agentes de longa data, dois conceitos críticos começam a reproduzir: irritação incluindo assinatura. A substituição permite manter um status de agente em qualquer ponto, permitindo reiniciar o estado por contato futuro. Isso é importante para aplicações longas. Por outro lado, a transmissão permite extrair sinais em tempo real sobre o que um agente é feito a qualquer momento, para fornecer e controlar suas ações. Nesta lição, desenvolveremos nosso agente adicionando esses recursos poderosos.

Configurando um agente

Vamos primeiro devolver nosso agel. Enviaremos as variáveis ​​ambientais, inseriremos e enviaremos as bibliotecas necessárias, definiremos a ferramenta de pesquisa a quente, explicará o estado do agente e, eventualmente, construir um 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'] = ""
os.environ['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}

Adicionando persistência

Instalando, usaremos o Langgraph's falido . A oposição de verificação salva o status do agente depois de todos os nós. Nesta lição, vamos usar SqlicaserPontos de verificação simples recebidos pelo SQLite, um banco de dados embutido. Embora utilizemos um banco de dados de memória Simpland, você pode se conectar facilmente ao banco de dados externo ou usar outras áreas de teste, como Redis ou Postgres de perseverança mais persistente.

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

Em seguida, mudaremos nossa agência para aceitar o check -se:

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

Agora, podemos construir nosso agente com perseverança capacitada:

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)

Adicionando transmissão

A transmissão é importante para a revisão do tempo real. Existem dois tipos de transmissões para se concentrar:

1. Para espalhar as mensagens: Mensagens intermediárias como opções de IA e resultados da barra de ferramentas.

2. Espalhar fichas: Espalhar tokens individuais da resposta LLM.
Vamos começar espalhando mensagens. Vamos criar a mensagem de uma pessoa e usarmos fluxo Como verificar as ações do agente em tempo 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'])

Última liberação: clima atual no Texas anualmente com a temperatura de 19,4 ° C (66,9 ° F) Espaço de velocidade do ar (6,8 km / h) … .. ..

Se você executar isso, verá a propagação dos resultados. Primeiro, a IA educou que o agente exige rastreamento, seguido pela mensagem da ferramenta com os resultados da pesquisa e, finalmente, a mensagem da IA ​​está respondendo à pergunta.

Entendendo IDs

Esta página cabo_id É uma parte importante da configuração da string. Permite que o agente salve conversas exclusivas com usuários ou sites. Ao alocar uma corda única para cada conversa, o agente pode economizar muito interagindo muito ao mesmo tempo sem se misturar.

Por exemplo, vamos continuar a conversa perguntando: “E LA?” Usando a mesma linha:

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)

Última liberação: As condições climáticas atuais em Los Angeles são pelo menos 17,2 ° C (63,0 ° F) velocidade apressiva de 2,2 mph) ….

Agente Fres pedindo o tempo, por insistência. Para confirmar, perguntemos: “O que é quente?” ::

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)

Último lançamento: o Texas é mais quente que Los Angeles. A temperatura atual no Texas é de 19,4 ° C (66,9 ° F), enquanto a temperatura atual em Los Angeles é de 17,2 ° C (63,0 ° F)

O agente compara o clima ao Texas e LA. Verificar se a persistência mantém as conversas separadas, vamos fazer a mesma pergunta de maneira diferente cabo_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)

Extração: preciso de mais detalhes para responder a essa pergunta. Você pode dar mais contexto ou especificar quais duas coisas você compara?

Enquanto isso, o agente está confuso porque você não tem acesso ao histórico do bate -papo anterior.

Espalhar fichas

Tokens de streaming, vamos usar Abstam_events maneira, assíncrona. Vamos mudar e no checkpointer assíncrono.

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="|")

Isso transmitirá os tokens em tempo real e fornecerá uma idéia ao vivo do processo de pensamento do agente.

Loja

Ao adicionar persistência e distribuição, melhoramos nosso poder ao agente. A perseverança permite que o agente salve o contexto nessa parceria, enquanto a distribuição fornece uma verdadeira compreensão de suas ações. Esses recursos são importantes para criar aplicativos eficientes, especialmente aqueles que envolvem muitos usuários ou parcerias – em loop.

No próximo estudo, entraremos Cooperação pessoal – em loopQuando a perseverança desempenha um papel muito importante na criação de interações ameaçadoras entre pessoas e agentes. Fique atento!

Referências:

1. (Deeplecceaning.ai)

Além disso, não se esqueça de seguir Sane e junte -se ao nosso Estação de telégrafo incluindo LinkedIn grtópico. Não se esqueça de se juntar ao nosso 75k + ml subreddit.

🚨 Conheça o trabalho: um código aberto aberto com várias fontes para verificar o programa difícil AI ^(Atualizado)