Ankstesniame vadove mes sukūrėme AI agentą, galintį atsakyti į klausimus naršydami internete ir pridėjome atkaklumą, kad išlaikytume būseną. Tačiau daugeliu scenarijų galbūt norėsite įkišti žmogų į kilpą, kad stebėtumėte ir patvirtintumėte agento veiksmus. Tai galima lengvai pasiekti naudojant „Langgraph“. Panagrinėkime, kaip tai veikia.
Agento nustatymas
Mes tęsime iš ten, kur pasitraukėme paskutinėje pamokoje. Pirmiausia nustatykite aplinkos kintamuosius, atlikite reikiamą importą ir sukonfigūruokite „CheckPointer“.
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.1import 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, AIMessage
from langchain_groq import ChatGroq
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
sqlite_conn = sqlite3.connect("checkpoints.sqlite",check_same_thread=False)
memory = SqliteSaver(sqlite_conn)
# Initialize the search tool
tool = TavilySearchResults(max_results=2)Agento apibrėžimas
class Agent:
def __init__(self, model, tools, checkpointer, 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(checkpointer=checkpointer)
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}Agento būsenos nustatymas
Dabar mes sukonfigūruojame agento būseną šiek tiek modifikuodami. Anksčiau pranešimų sąrašas buvo anotuotas su operatoriumi.add, pridedant naujus pranešimus į esamą masyvą. Sąveikai su žmogaus sąveika kartais taip pat norime pakeisti esamus pranešimus tuo pačiu ID, o ne pridėti juos.
from uuid import uuid4
def reduce_messages(left: list(AnyMessage), right: list(AnyMessage)) -> list(AnyMessage):
# Assign IDs to messages that don't have them
for message in right:
if not message.id:
message.id = str(uuid4())
# Merge the new messages with the existing ones
merged = left.copy()
for message in right:
for i, existing in enumerate(merged):
if existing.id == message.id:
merged(i) = message
break
else:
merged.append(message)
return merged
class AgentState(TypedDict):
messages: Annotated(list(AnyMessage), reduce_messages)Žmogaus pridėjimas į kilpą
Sudarant grafiką, mes pristatome papildomą modifikaciją. „Interrupt_before =“ („Veiksmas“) parametras prideda pertraukimą prieš skambinant į veiksmo mazgą, užtikrinant rankinį patvirtinimą prieš vykdant įrankius.
class Agent:
def __init__(self, model, tools, checkpointer, system=""):
# Everything else remains the same as before
self.graph = graph.compile(checkpointer=checkpointer, interrupt_before=("action"))
# Everything else remains unchangedVykdykite agentą
Dabar mes inicijuosime sistemą tuo pačiu raginimu, modeliu ir „CheckPointer“ kaip ir anksčiau. Kai skambiname agentui, mes perduodame gijos konfigūraciją su gijos ID.
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")
abot = Agent(model, (tool), system=prompt, checkpointer=memory)
messages = (HumanMessage(content="Whats the weather in SF?"))
thread = {"configurable": {"thread_id": "1"}}
for event in abot.graph.stream({"messages": messages}, thread):
for v in event.values():
print(v)Atsakymai perduodami atgal, o procesas sustoja po AI pranešimo, kuris nurodo įrankio skambutį. Tačiau „Interrupt_Be“ parametras neleidžia nedelsiant vykdyti. Mes taip pat galime gauti dabartinę šios gijos grafiko būseną ir pamatyti, ką joje yra, taip pat yra tas, koks yra kitas mazgas, kuris bus vadinamas („Veiksmas“ čia).
abot.graph.get_state(thread)
abot.graph.get_state(thread).nextNorėdami tęsti, mes vėl skambiname srautui su ta pačia gijos konfigūracija, neperduodami nė vieno kaip įvestis. Tai perduoda rezultatus, įskaitant įrankio pranešimą ir galutinį AI pranešimą. Kadangi tarp veiksmo mazgo ir LLM mazgo nebuvo pridėta jokio pertraukimo, vykdymas tęsiasi sklandžiai.
for event in abot.graph.stream(None, thread):
for v in event.values():
print(v)Interaktyvus žmogaus patvirtinimas
Prieš tęsdami vykdymą, galime įdiegti paprastą kilpą, raginančią vartotoją patvirtinti. Naujas gijos ID naudojamas naujam vykdymui. Jei vartotojas nusprendžia ne toliau tęsti, agentas sustoja.
messages = (HumanMessage("What's the weather in LA?"))
thread = {"configurable": {"thread_id": "2"}}
for event in abot.graph.stream({"messages": messages}, thread):
for v in event.values():
print(v)
while abot.graph.get_state(thread).next:
print("\n", abot.graph.get_state(thread), "\n")
_input = input("Proceed? (y/n): ")
if _input.lower() != "y":
print("Aborting")
break
for event in abot.graph.stream(None, thread):
for v in event.values():
print(v)Puiku! Dabar jūs žinote, kaip galite įtraukti žmogų į kilpą. Dabar pabandykite eksperimentuoti su skirtingais pertraukimais ir pažiūrėkite, kaip agentas elgiasi.
Nuorodos: Deeplearning.ai (https://learn.deeplearning.ai/courses/ai-agents-in-langgraph/lesson/6/human-in-the-loop)
„Vineet Kumar“ yra „MarkTechPost“ konsultavimo praktikantas. Šiuo metu jis siekia savo BS iš Indijos technologijos instituto (IIT), Kanpur. Jis yra mašinų mokymosi entuziastas. Jis aistringai vertina tyrimus ir naujausius giluminio mokymosi, kompiuterinės vizijos ir susijusių sričių pažangą.
✅ (rekomenduojama) Prisijunkite prie mūsų telegramos kanalo
