langgraph agents langchain state orchestration python

LangGraph-Bausteine: Zustand, Knoten, Kanten und Aufruf

Das LangGraph-Wortschatz vor der Erstellung: eingegebener Zustand, kleine Knoten, bedingte Kanten, Aufruf von Tools sowie Nachrichten als Zustand, dargestellt anhand eines ersten einfachen Agenten.

Verlag: REACTAPP.TOPAuf dieser Website hinzugefügt 26. Sept. 20269 Min. Lesezeit2150 Wörter

Quellen: The Building Blocks of LangGraph (Part 0)

Sobald Modelle die Fähigkeit zum Werkzeuggebrauch und zum mehrschrittigen Denken erlangten, wurden ad-hoc-Schleifen schwierig zu testen und fortzusetzen. LangGraph macht den Kontrollfluss explizit: Zustand, Knoten, Kanten und Aufrufe – sodass Agenten als Graphen dargestellt werden, die man überprüfen kann, anstatt als unübersichtlicher Prompt-Strang.

Der einfachste Graph

Ein minimales Agentenflussdiagramm sieht so aus: Start → Denk-/Handlungs-Knoten(n) → Ende, wobei der gemeinsame Zustand zwischen den Schritten übertragen wird.

Einen ersten einfachen Agenten erstellen

Receive Order

Prepare Food

Deliver Food

Receive Order
      ↓
Prepare Food
      ↓
Deliver Food

Pizza
Address: 123 Main Street
Customer: John

Receive Order
      ↓
Prepare Food
      ↓
Deliver Food

from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: strNodes

def node_1(state):
  print("---Node 1---")
  return {"graph_state": state['graph_state'] + "I am"}

def node_2(state):
  print("---Node 2---")
  return {"graph_state": state['graph_state'] + "happy!"}

def node_3(state):
  print("---Node 3---")
  return {"graph_state": state['graph_state'] + "sad!"}

import random
from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:

  # Often, we will use state to decide on the next node to visit
  user_input = state['graph_state']


  # Here, let's just do a 50 / 50 split between nodes 2, 3
  if random.random() < 0.5

      # 50% of the time, we return Node 2
      return "node_2"

  # 50% of the time, we return Node 3
  return "node_3"

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

#Build Graph
builder = StateGraph(state)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

#Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

#Add
graph = builder.compile()

#View
display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"graph_state" : "Hi, this is Lance."})

###OUTPUT
---Node 1---
---Node 3---
{'graph_state': 'Hi, this is Lance. I am sad!'}

{'graph_state': 'Hi, this is Lance. I am sad!'}

%%capture --no-stderr
%pip install --quiet -U langchain_openai langchain_core langgraph

from pprint import pprint
from langchain_core.messages import AIMessage, HumanMessage

messages = [AIMessage(content=f"So you said you were researching ocean mammals?", name="Model")]
messages.append(HumanMessage(content=f"Yes, that's right.",name="Lance"))
messages.append(AIMessage(content=f"Great, what would you like to learn about.", name="Model"))
messages.append(HumanMessage(content=f"I want to learn about the best place to see Orcas in the US.", name="Lance"))

for m in messages:
    m.pretty_print()

#OUTPUT
================================== Ai Message ==================================
Name: Model

So you said you were researching ocean mammals?
================================ Human Message =================================
Name: Lance

Yes, that's right.
================================== Ai Message ==================================
Name: Model

Great, what would you like to learn about.
================================ Human Message =================================
Name: Lance

I want to learn about the best place to see Orcas in the US.

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
result = llm.invoke(messages)
type(result)
result

#OUTPUT

langchain_core.messages.ai.AIMessage
result
AIMessage(content='One of the best places to see orcas in the United States is the Pacific Northwest, particularly around the San Juan Islands in Washington State. Here are some details:\n\n1. **San Juan Islands, Washington**: These islands are a renowned spot for whale watching, with orcas frequently spotted between late spring and early fall. The waters around the San Juan Islands are home to both resident and transient orca pods, making it an excellent location for sightings.\n\n2. **Puget Sound, Washington**: This area, including places like Seattle and the surrounding waters, offers additional opportunities to see orcas, particularly the Southern Resident killer whale population.\n\n3. **Olympic National Park, Washington**: The coastal areas of the park provide a stunning backdrop for spotting orcas, especially during their migration periods.\n\nWhen planning a trip for whale watching, consider peak seasons for orca activity and book tours with reputable operators who adhere to responsible wildlife viewing practices. Additionally, land-based spots like Lime Kiln Point State Park, also known as “Whale Watch Park,” on San Juan Island, offer great opportunities for orca watching from shore.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 228, 'prompt_tokens': 67, 'total_tokens': 295, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_50cad350e4', 'finish_reason': 'stop', 'logprobs': None}, id='run-57ed2891-c426-4452-b44b-15d0a5c3f225-0', usage_metadata={'input_tokens': 67, 'output_tokens': 228, 'total_tokens': 295, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

def multiply(a:int, b:int)-> int:
  return a*b

llm_with_tools = llm.bind_tools([multiply])

tool_call = llm_with_tools.invoke([HumanMessage(content=f"What is 2 multiplied by 3
", name="Lance")])

tool_call.tool_calls

#OUTPUT
[{'name': 'multiply',
  'args': {'a': 2, 'b': 3},
  'id': 'call_lBBBNo5oYpHGRqwxNaNRbsiT',
  'type': 'tool_call'}]

from typing_extentions import TypedDict
from langchain_core.messges import AnyMessage

class MessagesState(TypedDict):
  messages : list[AnyMessage]

from typing import Annotated
from langgraph.graph.message import add_messages

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

from langgraph.graph import MessagesState

class MessagesState(MessagesState):
    # Add any keys needed beyond messages, which is pre-built
    pass

# Initial state
initial_messages = [AIMessage(content="Hello! How can I assist you?", name="Model"),
                    HumanMessage(content="I'm looking for information on marine biology.", name="Lance")
                   ]

# New message to add
new_message = AIMessage(content="Sure, I can help with that. What specifically are you interested in?", name="Model")

# Test
add_messages(initial_messages , new_message)

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# Node
def tool_calling_llm(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# Build graph
builder = StateGraph(MessagesState)
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_edge(START, "tool_calling_llm")
builder.add_edge("tool_calling_llm", END)
graph = builder.compile()

# View
display(Image(graph.get_graph().draw_mermaid_png()))

messages = graph.invoke({"messages": HumanMessage(content="Hello!")})
for m in messages['messages']:
    m.pretty_print()

#OUTPUT
================================ Human Message =================================

Hello!
================================== Ai Message ==================================

Hi there! How can I assist you today?

messages = graph.invoke({"messages": HumanMessage(content="Multiply 2 and 3")})
for m in messages['messages']:
    m.pretty_print()

OUTPUT

================================ Human Message =================================

Multiply 2 and 3!
================================== Ai Message ==================================
Tool Calls:
  multiply (call_Er4gChFoSGzU7lsuaGzfSGTQ)
 Call ID: call_Er4gChFoSGzU7lsuaGzfSGTQ
  Args:
    a: 2
    b: 3

Zustand

Der typisierte Zustand (oft ein TypedDict oder Reducer) enthält Nachrichten, Flags und Zwischenergebnisse. Wählen Sie die Reducer bewusst aus – verwenden Sie „append“ für Nachrichtenlisten und „replace“ für Skalare.

Knoten

Knoten sind Funktionen: Sie lesen den Zustand, führen Arbeit aus (LLM, Tool, reiner Python) und geben eine teilweise Aktualisierung zurück. Halten Sie sie klein und testbar.

Kanten

Statische Kanten legen die Reihenfolge fest; bedingte Kanten teilen sich je nach Zustand auf. Graphen sind nützlich, wenn Verzweigungen und Schleifen auftreten.

Graphenkonstruktion

Registrieren Sie Knoten, verbinden Sie Kanten und kompilieren Sie mit einem optionalen Checkpointer zur Gewährleistung der Langlebigkeit.

Aufruf von Graphen

Rufen Sie den Graphen mit einem Eingabezustand und Konfiguration auf (Thread-ID bei der Erstellung eines Checkpoints). Streamen Sie Ereignisse für die Benutzeroberfläche.

Kette versus Graph

Linienförmige Ketten eignen sich weiterhin für eine einmalige Verarbeitung. Graphen ermöglichen Pausieren/Wiederaufnehmen, Verzweigungen sowie gemeinsam genutzte, benannte Zustände.

Nachrichten und Chat-Modelle

Nachrichtenlisten stellen den üblichen konversationellen Zustand dar. Binden Sie Chat-Modelle innerhalb der Knoten ein, anstatt I/O in der Logik der Kanten zu verbergen.

Werkzeuge

Stellen Sie Werkzeuge den Knoten zur Aufrufung von Modellen zur Verfügung; parsen Sie Werkzeugaufrufe und schreiben Sie Beobachtungen wieder in den Zustand zurück.

Nutzung von Nachrichten als Zustand

Wenn Nachrichten der Zustand sind, müssen Reducer sicher hinzufügen, und Trunkierung/Zusammenfassung wird zu einem expliziten Knoten – nicht zu einem Zufall.

Fazit

Teil 0 befasst sich mit dem Wortschatz: Zustand, Knoten, Kanten, Kompilieren, Aufrufen. Spätere Teile fügen auf derselben Grundlage dauerhafte Threads, HITL-Unterbrechungen sowie mehragentenbasierte Muster hinzu.

Namen Sie Knoten nach Geschäftsabläufen (classify, retrieve, draft) statt node1. Lesbare Graphen sind auch bedienbare Graphen.

Fügen Sie einen Unit-Test hinzu, der den kompilierten Graphen mit einem festgelegten Zustand aufruft und überprüft, welche Knoten ausgeführt wurden. Allein die Funktion compile() beweist nicht die Vollständigkeit des Geschäftsablaufs.

Namen Sie Knoten nach Geschäftsabläufen (classify, retrieve, draft) statt node1. Lesbare Graphen sind auch bedienbare Graphen.

Fügen Sie einen Unit-Test hinzu, der das kompilierte Graph mit einem festgelegten Zustand aufruft und überprüft, welche Knoten ausgeführt wurden. Allein die Funktion compile() beweist nicht die Vollständigkeit der Geschäftslogik.

Namen Sie die Knoten nach Geschäftsschritten (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Namen Sie die Knoten nach Geschäftsabläufen (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Fügen Sie einen Unit-Test hinzu, der den kompilierten Graphen mit einem festgelegten Zustand aufruft und überprüft, welche Knoten ausgeführt wurden. Allein compile() beweist nicht die Vollständigkeit des Geschäftsablaufs.

Namen Sie die Knoten nach Geschäftsabläufen (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Namen Sie die Knoten nach Geschäftsschritten (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Namen Sie die Knoten nach Geschäftsabläufen (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Namen Sie die Knoten nach Geschäftsschritten (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Namen Sie die Knoten nach Geschäftsabläufen (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Namen Sie die Knoten nach Geschäftsschritten (classify, retrieve, draft) anstelle von node1. Lesbare Graphen sind auch bedienbare Graphen.

Namen Sie die Knoten nach Geschäftsabläufen (classify, retrieve, draft) statt node1. Lesbare Diagramme sind auch bedienbare Diagramme.

Fügen Sie einen Unit-Test hinzu, der das kompilierte Diagramm mit einem festgelegten Zustand aufruft und überprüft, welche Knoten ausgeführt wurden. Allein die Funktion compile() beweist nicht die Vollständigkeit des Geschäftsablaufs.

Verwenden Sie für Routing-Tests lieber deterministische Knoten: Injectieren Sie eine künstliche LLM, die feste Toolaufrufe zurückgibt, damit der CI keinen Netzwerkzugriff benötigt.

Dokumentieren Sie die authz-Regel für thread_id frühzeitig: Das Finden eines Checkpoints bedeutet nicht automatisch die Berechtigung, ihn zu lesen.

Lassen Sie tools mit Nebeneffekten hinter Zustandsänderungs-Kanten – auch in Prototypen von Teil 0 – denn Gewohnheiten bilden sich schnell.

Verwenden Sie für Routing-Tests lieber deterministische Knoten: Injectieren Sie eine künstliche LLM, die feste Toolaufrufe zurückgibt, damit der CI keinen Netzwerkzugriff benötigt.

Dokumentieren Sie die thread_id-Authentifizierungsregel frühzeitig: Das Finden eines Checkpoints bedeutet nicht automatisch die Berechtigung, ihn zu lesen.

Lassen Sie tools mit Nebeneffekten hinter Zustimmungsgrenzen – auch in Prototypen von Teil 0 – denn Gewohnheiten bilden sich schnell.

Wählen Sie für Routing-Tests deterministische Knoten: Fügen Sie eine falsche LLM ein, die feste Toolaufrufe zurückgibt, damit der CI keinen Netzwerkzugriff benötigt.

Dokumentieren Sie die thread_id-Authentifizierungsregel frühzeitig: Das Finden eines Checkpoints bedeutet nicht automatisch die Berechtigung, ihn zu lesen.

Lassen Sie tools mit Nebeneffekten hinter Zustimmungsgrenzen – auch in Prototypen von Teil 0 – denn Gewohnheiten bilden sich schnell.

Wählen Sie für Routing-Tests deterministische Knoten: Fügen Sie eine falsche LLM ein, die feste Toolaufrufe zurückgibt, damit der CI keinen Netzwerkzugriff benötigt.

← Alle Artikel Brief senden