Leitfaden zum Caching auf Knotenebene in LangGraph

Wenn Sie LangGraph lernen oder mehr darüber erfahren, ist es hilfreich, etwas über das vorgefertigte Caching auf Knotenebene in LangGraph zu wissen. Caching eliminiert nicht nur unnötige Berechnungen, sondern verkürzt auch die Latenz. Wir werden uns in diesem Artikel mit der Umsetzung befassen. Es wird davon ausgegangen, dass Sie eine Vorstellung von Agenten und Knoten in LangGraph haben, da wir uns nicht auf diesen Teil der Geschichte konzentrieren werden. Lassen Sie uns additionally ohne weitere Umschweife auf die Konzepte und die Implementierung eingehen.

Was ist Caching?

Beim Caching werden Daten im temporären Speicher gespeichert, damit das System sie schnell abrufen kann. Im Rahmen von LLMs Und KI-Agentenspeichert es frühere Anfragen und verwendet sie wieder, wenn dieselben Eingabeaufforderungen an das Modell oder den Agenten gesendet werden. Da es sich nicht um eine neue Anfrage handelt, erhebt das System keine Gebühren dafür und die Antwort kommt aufgrund des temporären Speichers schneller an. Wenn ein Teil einer Eingabeaufforderung gleich bleibt, verwendet das System die vorherige Antwort wieder und generiert nur für den zusätzlichen Teil eine neue, was die Kosten auch für neue Anfragen erheblich senkt.

Caching-Parameter und Speicher

Es ist wichtig, den TTL-Parameter (Time to Dwell) zu kennen, der verwendet wird, um die Zeitspanne (in Sekunden) zu definieren, die der Cache im Speicher verbleibt. Wenn wir einstellen ttl=Keine oder lassen Sie es so wie es ist, dann wird der Cache den Speicher nie verlassen.

Beim Kompilieren eines Diagramms müssen wir einen Cache angeben. Wir verwenden InMemoryCache, um die Eingaben und Ausgaben des Knotens zu speichern, die später zum Abrufen der vorherigen Antwort des Knotens in diesem Artikel verwendet werden können. Alternativ können Sie je nach Bedarf auch SqliteCache, redisCache oder benutzerdefinierten Cache implementieren.

Caching in Aktion

Lassen Sie uns umsetzen Caching auf Knotenebene für eine Funktion, die dabei hilft, Celsius in Fahrenheit umzurechnen.

Installationen

!pip set up langgraph 

Definieren des Diagramms

Wir definieren zunächst die Diagrammstruktur und eine einfache Funktion, die eine langsame Berechnung simuliert time.sleep() um den Caching-Effekt sichtbar zu machen.

import time 

from typing_extensions import TypedDict 

from langgraph.graph import StateGraph 

from langgraph.cache.reminiscence import InMemoryCache 

from langgraph.varieties import CachePolicy 

class State(TypedDict): 

   celsius: float 

   fahrenheit: float 

builder = StateGraph(State) 

def convert_temperature(state: State) -> dict(str, float): 

   time.sleep(2) 

   fahrenheit = (state('celsius') * 9/5) + 32 

   return {"fahrenheit": fahrenheit} 

builder.add_node("convert_temperature", convert_temperature, cache_policy=CachePolicy(ttl=None)) 

builder.set_entry_point("convert_temperature") 

builder.set_finish_point("convert_temperature") 

cache=InMemoryCache() 

graph = builder.compile(cache=cache)

Aufrufen des Diagramms

Rufen wir nun das Diagramm mehrmals auf und beobachten, wie sich der Cache verhält.

print(graph.invoke({"celsius": 25})) 

print(graph.invoke({"celsius": 25}, stream_mode="updates"))  # Cached 

print(graph.invoke({"celsius": 36}, stream_mode="updates"))  

time.sleep(10) 

print(graph.invoke({"celsius": 36}, stream_mode="updates"))  

cache.clear() # clears the complete cache

Ausgabe:

{'celsius': 25, 'fahrenheit': 77.0} 
({'convert_temperature': {'fahrenheit': 77.0}, '__metadata__': {'cached': True}}) 
({'convert_temperature': {'fahrenheit': 96.8}}) 
({'convert_temperature': {'fahrenheit': 96.8}})

Das System ruft die Antwort bei der ersten wiederholten Anfrage aus dem Cache ab und die TTL ist auf 5 Sekunden eingestellt. Es behandelt die nächste wiederholte Anfrage als eine neue, wenn die Lücke die TTL überschreitet. Wir haben verwendet cache.clear() Um den gesamten Cache zu leeren, ist dies beim Festlegen nützlich ttl=Keine.

Jetzt implementieren wir das Caching für den Knoten mit einem Agenten.

Voraussetzungen

Wir benötigen einen Gemini-API-Schlüssel, um Gemini-Modelle im Agenten verwenden zu können Google AI Studio So erhalten Sie Ihren API-Schlüssel: https://aistudio.google.com/api-keys

Installationen

Der langchain_google_genai Das Modul wird uns bei der Integration helfen Zwillinge Modelle im Knoten.

!pip set up langgraph langchain_google_genai 

Agentendefinition

Definieren wir einen einfachen Mathematikagenten, der Zugriff auf das Taschenrechnertool hat, und legen den fest ttl=Keine zur Zeit.

from langgraph.prebuilt import create_react_agent 

from langchain_google_genai import ChatGoogleGenerativeAI 

def solve_math_problem(expression: str) -> str: 

   """Resolve a math drawback.""" 

   strive: 

       # Consider the mathematical expression 

       outcome = eval(expression, {"__builtins__": {}}) 

       return f"The reply is {outcome}." 

   besides Exception: 

       return "I could not remedy that expression." 

# Initialize the Gemini mannequin with API key 

mannequin = ChatGoogleGenerativeAI( 

   mannequin="gemini-2.5-flash", 

   google_api_key=GOOGLE_API_KEY 

) 

# Create the agent 

agent = create_react_agent( 

   mannequin=mannequin, 

   instruments=(solve_math_problem), 

   immediate=( 

       "You're a Math Tutor AI. " 

       "When a consumer asks a math query, cause via the steps clearly " 

       "and use the device `solve_math_problem` for numeric calculations. " 

       "All the time clarify your reasoning earlier than giving the ultimate reply." 

   ), 

)

Definieren des Knotens

Als Nächstes packen wir den Agenten in einen LangGraph-Knoten und fügen ihm Caching hinzu.

import time 

from typing_extensions import TypedDict 

from langgraph.graph import StateGraph 

from langgraph.cache.reminiscence import InMemoryCache 

from langgraph.varieties import CachePolicy 

class AgentState(TypedDict): 

   immediate: str 

   response: str 

builder = StateGraph(AgentState) 

def run_agent(state: AgentState) -> AgentState: 

   print("Working agent...")  # this line helps present caching habits 

   response = agent.invoke({"messages": ({"function": "consumer", "content material": state("immediate")})}) 

   return {"response": response} 

builder.add_node("run_agent", run_agent, cache_policy=CachePolicy(ttl=None)) 

builder.set_entry_point("run_agent") 

builder.set_finish_point("run_agent") 

graph = builder.compile(cache=InMemoryCache())

Aufruf des Agenten

Rufen wir abschließend den Agenten zweimal an, um das Caching in Aktion zu sehen.

# Invoke graph twice to see caching 

print("First name") 

result1 = graph.invoke({"immediate": "What's (12 + 8) * 3?"},stream_mode="updates") 

print(result1) 

print("Second name (needs to be cached)") 

result2 = graph.invoke({"immediate": "What's (12 + 8) * 3?"},stream_mode="updates") 

print(result2)

Ausgabe:

Beachten Sie, dass der zweite Aufruf nicht „Laufender Agent..‚, was eine Druckanweisung im Knoten ist. So ist es uns gelungen, die Antwort vom Agenten zu erhalten, ohne den Agenten über den Cache-Speicher auszuführen.

Abschluss

LangGraphDas integrierte Caching auf Knotenebene bietet eine einfache, aber leistungsstarke Möglichkeit, Latenz und Berechnungen durch Wiederverwendung früherer Ergebnisse zu reduzieren. Mit Parametern wie TTL zur Verwaltung der Cache-Lebensdauer und Optionen wie InMemoryCache, SqliteCache oder RedisCache bietet es Flexibilität je nach Anwendungsfall. Anhand von Beispielen wie der Temperaturkonvertierung zu agentenbasierten Knoten! Wir haben gesehen, wie Caching redundante Ausführung vermeidet und Kosten spart. Insgesamt verbessert das Caching in LangGraph die Effizienz erheblich und macht Arbeitsabläufe schneller und optimierter.

Häufig gestellte Fragen

Leidenschaftlich für Technologie und Innovation, Absolvent des Vellore Institute of Know-how. Derzeit arbeite ich als Information Science Trainee mit Schwerpunkt auf Information Science. Großes Interesse an Deep Studying und generativer KI, begierig darauf, modernste Techniken zu erforschen, um komplexe Probleme zu lösen und wirkungsvolle Lösungen zu schaffen.