LangChain Agenten-Workflow

Im vorherigen Abschnitt haben wir die Parameter voncreate_agent()kennengelernt.

Dieser Abschnitt taucht tief in die Interna des Agenten ein, um zu verstehen, wie er funktioniert – wie das Modell und die Tools zusammenarbeiten und wie der Agent stoppt.

Agent-Ausführungsschleife

Der Kern eines Agenten ist einfach:Modell aufrufen → prüfen, ob ein Tool benötigt wird → Tool ausführen → wiederholen. Sobald das Modell keine Tool-Aufrufe mehr anfordert, stoppt der Agent und gibt das Endergebnis zurück.

Im Folgenden verfolgen wir jeden Schritt des Agenten, um diesen Prozess zu verstehen.

Beispiel

from dotenv import load_dotenv
load_dotenv()

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

@tool
def get_weather(city: str) -> str:
"""Fragt das Wetter für eine angegebene Stadt ab.

Argumente:
city: Stadtname
"""
weather_data = {
"Hangzhou": "Sonnig, 25°C",
"Beijing": "Bewölkt, 18°C",
}
return weather_data.get(city, f"Wetterdaten für {city} nicht gefunden")

@tool
def get_time(city: str) -> str:
"""Fragt die aktuelle Zeit für eine angegebene Stadt ab.

Argumente:
city: Stadtname
"""
time_data = {
"Hangzhou": "14:30",
"Beijing": "14:30",
"New York": "02:30",
}
return time_data.get(city, f"Zeitdaten für {city} nicht gefunden")

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[get_weather, get_time],
system_prompt="Du bist ein hilfreicher Assistent.",
)

# Mit stream_mode="updates" kannst du jeden Schritt sehen
print("=== Agent-Ausführungsverlauf ==="\n")
Schrittweite = 0
for chunk in agent.stream(
{"messages": [HumanMessage(content="Wie ist das Wetter jetzt in Hangzhou? Wie spät ist es?")]},
stream_mode="updates",
):
step += 1
print(f"--- Schritt {step} ---")
for node_name, update in chunk.items():
print(f"Knoten: {node_name}")
if "messages" in update:
for msg in update["messages"]:
if hasattr(msg, 'tool_calls') and msg.tool_calls:
# KI-Nachricht enthält Tool-Aufrufe
for tc in msg.tool_calls:
print(f" → Tool-Aufruf wird angefordert: {tc['name']}({tc['args']})")
elif msg.type == "tool":
print(f" → Tool-Ergebnis [{msg.name}]: {msg.content}")
elif msg.type == "ai" and msg.content:
print(f" → KI-Antwort: {msg.content[:100]}")

Ausführungsergebnis:

=== Agent 执行过程追踪 ===

--- 步骤 1 ---
Node: model
  → 请求调用工具: get_weather({'city': '杭州'})
  → 请求调用工具: get_time({'city': '杭州'})

--- 步骤 2 ---
Node: tools
  → 工具结果 [get_weather]: 晴，25°C
  → 工具结果 [get_time]: 14:30

--- 步骤 3 ---
Node: model
  → AI 回复: 杭州现在天气晴朗，气温25°C，当前时间是14:30.

Aus dieser Ablaufverfolgung können Sie sehen, dass der Agent3 Schritte ausgeführt hat：

Schritt 1 (Modellknoten): Das Modell empfängt die Frage, stellt fest, dass es die beiden Tools get_weather und get_time aufrufen muss, und gibt zwei tool_calls zurück
Schritt 2 (Tools-Knoten): Führt die beiden Tools aus, um die Wetter- und Zeitergebnisse zu erhalten
Schritt 3 (Modellknoten): Das Modell empfängt die Tool-Ergebnisse, stellt fest, dass die Informationen ausreichen, und generiert die endgültige Antwort

stream_mode ausführliche Erklärung

stream() unterstützt mehrere stream_modes, die jeweils Informationen mit unterschiedlicher Granularität liefern:

Mode	Zurückgegebener Inhalt	Geeignete Szenarien
updates	Zustandsaktualisierungen nach Ausführung jedes Knotens	Agent-Ausführungsschritte verfolgen, Zwischenergebnisse anzeigen
values	Vollständiger Zustand nach Ausführung jedes Knotens	Vollständigen Nachrichtenverlauf bei jedem Schritt sehen müssen
messages	Token-für-Token-Nachrichtenstrom	Frontend-Streaming-Anzeige des KI-Tippeffekts
custom	Benutzerdefinierte Ereignisse	Middleware sendet benutzerdefinierte Ereignisse über stream_writer

stream_mode="values" — Vollständige Zustandsänderungen anzeigen

Beispiel

print("=== stream_mode='values' ===\n")
for i, chunk in enumerate(agent.stream(
{"messages": [HumanMessage(content="Wie ist das Wetter in Hangzhou?")]},
stream_mode="values",
)):
messages = chunk.get("messages", [])
print(f"Zustand {i}: {len(messages)} Nachrichten")
for msg in messages:
print(f" [ {msg.type}] {str(msg.content)[:80]}")
if i >= 3: # Nur die ersten paar Zustände ansehen
break

Ausführungsergebnis:

=== stream_mode='values' ===

State 0: 1 条消息
  [human] 杭州天气怎么样？

State 1: 2 条消息
  [human] 杭州天气怎么样？
  [ai]

State 2: 3 条消息
  [human] 杭州天气怎么样？
  [ai]
  [tool] 晴，25°C

State 3: 4 条消息
  [human] 杭州天气怎么样？
  [ai]
  [tool] 晴，25°C
  [ai] 杭州今天天气晴朗，气温25°C，适合出门活动。

stream_mode="messages" — Token-für-Token-Streaming-Ausgabe

Beispiel

print("=== stream_mode='messages' (Streaming-Tippeffekt) ==="\n")
for msg_chunk, metadata in agent.stream(
{"messages": [HumanMessage(content="Stelle Python in einem Satz vor")]},
stream_mode="messages",
):
# msg_chunk ist ein AIMessageChunk, der nur ein Token enthält
if hasattr(msg_chunk, 'content') and msg_chunk.content:
print(msg_chunk.content, end="", flush=True)
print()

Beendigungsbedingungen des Agenten

Wann stoppt der Agent? Hauptsächlich in den folgenden Fällen:

Beendigungsbedingung	Beschreibung	Beispiel
Keine Tool-Aufrufe	tool_calls ist in der vom Modell zurückgegebenen AIMessage leer	Das Modell betrachtet die Aufgabe als abgeschlossen und antwortet direkt
return_direct=True	Tool für direkte Rückgabe markiert, endet unmittelbar nach der Ausführung	Abfrage-Tool, das Ergebnis ist die endgültige Antwort
structured_response	Das Modell hat strukturierte Ausgabe erzeugt	Die durch response_format festgelegte strukturierte Ausgabe ist vollständig
jump_to="end"	Middleware beendet aktiv über Zustandssteuerung	Berechtigungsverletzung erkannt, vorzeitig beendet

Beispiel

from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# Fall 1: Modell antwortet direkt (kein Tool nötig)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[], # Keine Tools
)

result = agent.invoke({
"messages": [HumanMessage(content="Stelle Python in einem Satz vor")]
})
# Das Modell antwortet direkt, die Schleife wird nur einmal ausgeführt
print(f"Szenario ohne Tools, Nachrichtenanzahl: {len(result['messages'])}")
print(f"Antwort: {result['messages'][-1].content[:80]}...")

# Fall 2: Tool-Aufruf erforderlich (Schleife mit mehreren Runden)
@tool
def search_course(keyword: str) -> str:
"""Beispielkurse durchsuchen"""
return f"3 Kurse zu {keyword} gefunden"

agent_with_tools = create_agent(
model=model,
tools=[search_course],
)

result = agent_with_tools.invoke({
"messages": [HumanMessage(content="Python-Kurse suchen")]
})
print(f"\nTool-Szenario, Nachrichtenanzahl: {len(result['messages'])}")
# Normalerweise gibt es: human, ai(tool_call), tool(result), ai(final)
for msg in result["messages"]:
print(f" [{msg.type}]", end="")
if hasattr(msg, 'tool_calls') and msg.tool_calls:
print(f" tool_calls: {[tc['name'] for tc in msg.tool_calls]}")
else:
print(f" {str(msg.content)[:50]}")

Ausführungsergebnis:

无工具场景，消息数: 2
Reply: Example is a programming learning platform for beginners, offering rich tutorials and examples...

有工具场景，消息数: 4
  [human] 搜索 Python 课程
  [ai] tool_calls: ['search_course']
  [tool] 找到 python 相关课程 3 门
  [ai] Searching Example for "Python courses" found 3 related courses...

invoke vs stream Vergleich

Methode	Rückgabezeitpunkt	Geeignete Szenarien	Benutzererfahrung
invoke()	Gibt nach Abschluss alles auf einmal zurück	Skripte, API-Schnittstellen, Stapelverarbeitung	Vollständiges Ergebnis nach dem Warten sehen
stream()	Gibt Zwischenzustände Schritt für Schritt zurück	Chat-Oberflächen, wenn der Prozess angezeigt werden muss	Fortschritt in Echtzeit sehen
ainvoke()	Gibt nach asynchronem Abschluss zurück	Web-Services, asynchrone Frameworks	Blockiert die Event-Loop nicht
astream()	Asynchrone schrittweise Rückgabe	WebSocket, SSE-Push	Serverseitiger Echtzeit-Push

Thread-ID in with_config übergeben

Wenn Sie einen Checkpointer verwenden (in späteren Kapiteln ausführlich beschrieben), müssen Sie thread_id über config übergeben, um Konversations-Threads zu verwalten:

Beispiel

# config wird verwendet, um Laufzeitkonfiguration zu übergeben
# thread_id wird verwendet, um verschiedene Konversations-Threads zu unterscheiden
config = {"configurable": {"thread_id": "conversation-001"}}

# invoke-Methode
result = agent.invoke(
{"messages": [HumanMessage(content="Hello")]},
config=config,
)

# stream-Methode unterstützt ebenfalls config
for chunk in agent.stream(
{"messages": [HumanMessage(content="Hello")]},
config=config,
stream_mode="updates",
):
print(chunk)

Andere Erweiterungen

Agent Entwicklungskonfiguration

Referenzhandbuch

LangChain Agenten-Workflow

Agent-Ausführungsschleife

Beispiel

stream_mode ausführliche Erklärung

stream_mode="values" — Vollständige Zustandsänderungen anzeigen

Beispiel

stream_mode="messages" — Token-für-Token-Streaming-Ausgabe

Beispiel

Beendigungsbedingungen des Agenten

Beispiel

invoke vs stream Vergleich

Thread-ID in with_config übergeben

Beispiel