AI agento sistemos su „LangGraph“ kūrimas: patvarumo ir srauto pridėjimas (žingsnis po žingsnio vadovas)

Ankstesniame vadove mes sukūrėme AI agentą, galintį atsakyti į klausimus naršydami internete. Tačiau kuriant agentus ilgiau vykdomoms užduotims, atsiranda dvi kritinės sąvokos: atkaklumas ir srautas. Atkaklumas leidžia bet kuriuo metu išsaugoti agento būseną, leidžiančią atnaujinti iš šios būsenos būsimoje sąveikoje. Tai labai svarbu ilgalaikėms programoms. Kita vertus, srautas leidžia skleisti realiojo laiko signalus apie tai, ką agentas daro bet kurią akimirką, užtikrinant skaidrumą ir kontroliuojant jo veiksmus. Šiame vadove mes patobulinsime savo agentą, pridėdami šias galingas funkcijas.

Agento nustatymas

Pradėkime atkurti savo agentą. Mes įkelsime reikiamus aplinkos kintamuosius, įdiegsime ir importuosime reikiamas bibliotekas, nustatysime „Tavily“ paieškos įrankį, apibrėžsime agento būseną ir galiausiai sukursime agentą.

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}

Pridedant atkaklumo

Norėdami pridėti atkaklumą, naudosime „Langgraph“ „CheckPointer“ savybė. „Checkpointer“ išsaugo agento būseną po kiekvieno mazgo ir tarp jų. Šiam vadovėliui naudosime „SQLitesaver“paprastas „Checkpointer“, kuris panaudoja „SQLite“, įmontuotą duomenų bazę. Nors paprastumui naudosime atmintyje esančią duomenų bazę, galite lengvai ją prijungti prie išorinės duomenų bazės arba naudoti kitus kontrolės punktus, tokius kaip „Redis“ ar „Postgres“, kad būtų tvirtesnis atkaklumas.

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

Tada mes modifikuosime savo agentą, kad priimtume „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

Dabar galime sukurti savo agentą, įgalintą atkaklumu:

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)

Pridedant srautą

Srautinis perdavimas yra būtinas realiojo laiko atnaujinimams. Yra dviejų tipų srautai, į kuriuos sutelksime dėmesį:

1. Srautinis pranešimai: Tarpinių pranešimų skleidimas, pavyzdžiui, AI sprendimai ir įrankių rezultatai.

2. Srautiniai žetonai: Individualių žetonų srautai nuo LLM atsakymo.
Pradėkime nuo srautinių pranešimų. Mes sukursime žmogaus žinią ir naudosime srautas Agento veiksmų stebėjimo būdas realiuoju laiku.

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'))

Galutinė išvestis: Dabartinis oras Teksase yra saulėtas, kurio temperatūra yra 19,4 ° C (66,9 ° F), o vėjo greitis – 4,3 mylios per valandą (6,8 km / h) …

Kai paleisite tai, pamatysite rezultatų srautą. Pirmiausia, AI pranešimas, nurodantis agentui pašalinti pašėlusius, po to pateiktas įrankio pranešimas su paieškos rezultatais ir galiausiai AI pranešimas, atsakantis į klausimą.

Supratimas apie gijos ID

Thread_id yra esminė sriegio konfigūracijos dalis. Tai leidžia agentui palaikyti atskirus pokalbius su skirtingais vartotojais ar kontekstais. Kiekvienam pokalbiui priskirdamas unikalų siūlą_ID, agentas gali sekti kelių sąveikų vienu metu, nesumaišydamas jų.

Pvz., Tęskime pokalbį paklausdami: „O kaip LA?“ Naudojant tą patį siūlą_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)

Galutinė išvestis: Dabartinis oras Los Andžele yra saulėtas, jo temperatūra yra 17,2 ° C (63,0 ° F), o vėjo greitis – 2,2 mylios per valandą (3,6 km / h)….

Agentas daro išvadą, kad dėl atkaklumo mes klausiame apie orą. Patikrinkite, paklauskime: „Kuris iš jų yra šiltesnis?“:

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)

Galutinė produkcija: Teksasas yra šiltesnis nei Los Andželas. Dabartinė Teksaso temperatūra yra 19,4 ° C (66,9 ° F), o dabartinė temperatūra Los Andžele yra 17,2 ° C (63,0 ° F)

Agentas teisingai lygina orą Teksase ir LA. Norėdami patikrinti, ar atkaklumas laiko pokalbius atskirai, užduokite tą patį klausimą kitaip 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)

Išvestis: man reikia daugiau informacijos, kad atsakyčiau į šį klausimą. Ar galite pateikti daugiau konteksto arba nurodyti, kuriuos du dalykus lyginate?

Šį kartą agentas susipainioja, nes neturi prieigos prie ankstesnio pokalbio istorijos.

Srautiniai žetonai

Norėdami transliuoti žetonus, naudosime astream_events metodas, kuris yra asinchroninis. Mes taip pat pereisime prie „Async Checkpointer“.

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

Tai transliuos žetonus realiuoju laiku, suteikdamas tiesioginį vaizdą apie agento minties procesą.

Išvada

Pridėję atkaklumą ir srautą, mes žymiai patobulinome savo AI agento galimybes. Atkaklumas leidžia agentui išlaikyti sąveikos kontekstą, o srautas suteikia realiojo laiko įžvalgos apie savo veiksmus. Šios savybės yra būtinos kuriant gamybai paruoštas programas, ypač susijusias su keliais vartotojais ar sąveika su žmonėmis.

Kitame vadove mes pasinersime Žmogaus sąveikakur atkaklumas vaidina lemiamą vaidmenį įgalinant sklandų žmonių ir AI agentų bendradarbiavimą. Sekite!

Nuorodos:

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

Be to, nepamirškite sekti mūsų „Twitter“ ir prisijunkite prie mūsų „Telegram“ kanalas ir „LinkedIn GrOUP. Nepamirškite prisijungti prie mūsų 75K+ ml subreddit.

🚨 Susipažinkite su „Intellagent“: atvirojo kodo daugialypės terpės sistema, skirta įvertinti sudėtingą pokalbio AI sistemą ^{(Paaukštintas)}