In diesem Tutorial implementieren wir mithilfe von LangGraph ein Agenten-KI-Muster, das Argumentation und Aktion als transaktionalen Workflow und nicht als einmalige Entscheidung behandelt. Wir modellieren ein zweiphasiges Commit-System, in dem ein Agent reversible Änderungen inszeniert, strikte Invarianten validiert, über Graph-Interrupts eine Pause zur menschlichen Genehmigung einlegt und erst dann ein Commit oder Rollback durchführt. Damit zeigen wir, wie Agentensysteme so gestaltet werden können, dass Sicherheit, Überprüfbarkeit und Kontrollierbarkeit im Mittelpunkt stehen. Dabei geht es über reaktive Chat-Agenten hinaus hin zu strukturierten, Governance-bewussten KI-Workflows, die mithilfe von OpenAI-Modellen zuverlässig in Google Colab ausgeführt werden. Schauen Sie sich das an Vollständige Codes hier.
!pip -q set up -U langgraph langchain-openai
import os, json, uuid, copy, math, re, operator
from typing import Any, Dict, Listing, Optionally available
from typing_extensions import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.reminiscence import InMemorySaver
from langgraph.varieties import interrupt, Command
def _set_env_openai():
if os.environ.get("OPENAI_API_KEY"):
return
attempt:
from google.colab import userdata
okay = userdata.get("OPENAI_API_KEY")
if okay:
os.environ("OPENAI_API_KEY") = okay
return
besides Exception:
cross
import getpass
os.environ("OPENAI_API_KEY") = getpass.getpass("Enter OPENAI_API_KEY: ")
_set_env_openai()
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")
llm = ChatOpenAI(mannequin=MODEL, temperature=0)Wir richten die Ausführungsumgebung ein, indem wir LangGraph installieren und das OpenAI-Modell initialisieren. Wir laden den API-Schlüssel sicher und konfigurieren ein deterministisches LLM, um sicherzustellen, dass das gesamte Verhalten der Downstream-Agenten reproduzierbar und kontrolliert bleibt. Schauen Sie sich das an Vollständige Codes hier.
SAMPLE_LEDGER = (
{"txn_id": "T001", "title": "Asha", "e-mail": "(e-mail protected)", "quantity": "1,250.50", "date": "12/01/2025", "word": "Membership renewal"},
{"txn_id": "T002", "title": "Ravi", "e-mail": "(e-mail protected)", "quantity": "-500", "date": "2025-12-02", "word": "Chargeback?"},
{"txn_id": "T003", "title": "Sara", "e-mail": "(e-mail protected)", "quantity": "700", "date": "02-12-2025", "word": "Late price waived"},
{"txn_id": "T003", "title": "Sara", "e-mail": "(e-mail protected)", "quantity": "700", "date": "02-12-2025", "word": "Duplicate row"},
{"txn_id": "T004", "title": "Lee", "e-mail": "(e-mail protected)", "quantity": "NaN", "date": "2025/12/03", "word": "Dangerous quantity"},
)
ALLOWED_OPS = {"exchange", "take away", "add"}
def _parse_amount(x):
if isinstance(x, (int, float)):
return float(x)
if isinstance(x, str):
attempt:
return float(x.exchange(",", ""))
besides:
return None
return None
def _iso_date(d):
if not isinstance(d, str):
return None
d = d.exchange("/", "-")
p = d.cut up("-")
if len(p) == 3 and len(p(0)) == 4:
return d
if len(p) == 3 and len(p(2)) == 4:
return f"{p(2)}-{p(1)}-{p(0)}"
return None
def profile_ledger(rows):
seen, anomalies = {}, ()
for i, r in enumerate(rows):
if _parse_amount(r.get("quantity")) is None:
anomalies.append(i)
if r.get("txn_id") in seen:
anomalies.append(i)
seen(r.get("txn_id")) = i
return {"rows": len(rows), "anomalies": anomalies}
def apply_patch(rows, patch):
out = copy.deepcopy(rows)
for op in sorted((p for p in patch if p("op") == "take away"), key=lambda x: x("idx"), reverse=True):
out.pop(op("idx"))
for op in patch:
if op("op") in {"add", "exchange"}:
out(op("idx"))(op("area")) = op("worth")
return out
def validate(rows):
points = ()
for i, r in enumerate(rows):
if _parse_amount(r.get("quantity")) is None:
points.append(i)
if _iso_date(r.get("date")) is None:
points.append(i)
return {"okay": len(points) == 0, "points": points}Wir definieren die Kernbuchabstraktion zusammen mit der Patch-, Normalisierungs- und Validierungslogik. Wir behandeln Datentransformationen als umkehrbare Vorgänge, sodass der Agent sicher über Änderungen nachdenken kann, bevor er sie festschreibt. Schauen Sie sich das an Vollständige Codes hier.
class TxnState(TypedDict):
messages: Annotated(Listing(AnyMessage), add_messages)
raw_rows: Listing(Dict(str, Any))
sandbox_rows: Listing(Dict(str, Any))
patch: Listing(Dict(str, Any))
validation: Dict(str, Any)
authorised: Optionally available(bool)
def node_profile(state):
p = profile_ledger(state("raw_rows"))
return {"messages": (AIMessage(content material=json.dumps(p)))}
def node_patch(state):
sys = SystemMessage(content material="Return a JSON patch listing fixing quantities, dates, emails, duplicates")
usr = HumanMessage(content material=json.dumps(state("raw_rows")))
r = llm.invoke((sys, usr))
patch = json.hundreds(re.search(r"(.*)", r.content material, re.S).group())
return {"patch": patch, "messages": (AIMessage(content material=json.dumps(patch)))}
def node_apply(state):
return {"sandbox_rows": apply_patch(state("raw_rows"), state("patch"))}
def node_validate(state):
v = validate(state("sandbox_rows"))
return {"validation": v, "messages": (AIMessage(content material=json.dumps(v)))}
def node_approve(state):
determination = interrupt({"validation": state("validation")})
return {"authorised": determination == "approve"}
def node_commit(state):
return {"messages": (AIMessage(content material="COMMITTED"))}
def node_rollback(state):
return {"messages": (AIMessage(content material="ROLLED BACK"))}Wir modellieren den internen Standing des Agenten und definieren jeden Knoten im LangGraph-Workflow. Wir drücken das Agentenverhalten als diskrete, überprüfbare Schritte aus, die den Zustand verändern und gleichzeitig den Nachrichtenverlauf bewahren. Schauen Sie sich das an Vollständige Codes hier.
builder = StateGraph(TxnState)
builder.add_node("profile", node_profile)
builder.add_node("patch", node_patch)
builder.add_node("apply", node_apply)
builder.add_node("validate", node_validate)
builder.add_node("approve", node_approve)
builder.add_node("commit", node_commit)
builder.add_node("rollback", node_rollback)
builder.add_edge(START, "profile")
builder.add_edge("profile", "patch")
builder.add_edge("patch", "apply")
builder.add_edge("apply", "validate")
builder.add_conditional_edges(
"validate",
lambda s: "approve" if s("validation")("okay") else "rollback",
{"approve": "approve", "rollback": "rollback"}
)
builder.add_conditional_edges(
"approve",
lambda s: "commit" if s("authorised") else "rollback",
{"commit": "commit", "rollback": "rollback"}
)
builder.add_edge("commit", END)
builder.add_edge("rollback", END)
app = builder.compile(checkpointer=InMemorySaver())Wir konstruieren die LangGraph-Zustandsmaschine und kodieren explizit den Kontrollfluss zwischen Profilerstellung, Patching, Validierung, Genehmigung und Finalisierung. Wir verwenden bedingte Kanten, um Governance-Regeln durchzusetzen, anstatt uns auf implizite Modellentscheidungen zu verlassen. Schauen Sie sich das an Vollständige Codes hier.
def run():
state = {
"messages": (),
"raw_rows": SAMPLE_LEDGER,
"sandbox_rows": (),
"patch": (),
"validation": {},
"authorised": None,
}
cfg = {"configurable": {"thread_id": "txn-demo"}}
out = app.invoke(state, config=cfg)
if "__interrupt__" in out:
print(json.dumps(out("__interrupt__"), indent=2))
determination = enter("approve / reject: ").strip()
out = app.invoke(Command(resume=determination), config=cfg)
print(out("messages")(-1).content material)
run()Wir führen den Transaktionsagenten aus und kümmern uns um die Human-in-the-Loop-Genehmigung durch Graph-Interrupts. Wir nehmen die Ausführung deterministisch wieder auf und zeigen, wie Agenten-Workflows pausieren, externe Eingaben akzeptieren und sicher mit einem Commit oder Rollback abschließen können.
Abschließend haben wir gezeigt, wie LangGraph es uns ermöglicht, Agenten zu entwickeln, die über Zustände nachdenken, Validierungs-Gates durchsetzen und an genau definierten Kontrollpunkten mit Menschen zusammenarbeiten. Wir haben den Agenten nicht wie ein Orakel behandelt, sondern als einen Transaktionskoordinator, der seine eigenen Aktionen inszenieren, prüfen und rückgängig machen und dabei einen vollständigen Prüfpfad aufrechterhalten kann. Dieser Ansatz verdeutlicht, wie agentische KI auf reale Systeme angewendet werden kann, die Vertrauen, Compliance und Wiederherstellbarkeit erfordern, und bietet eine praktische Grundlage für den Aufbau autonomer Arbeitsabläufe in Produktionsqualität, die sicher, clear und von Menschen überwacht bleiben.
Schauen Sie sich das an Vollständige Codes hier. Sie können uns auch gerne weiter folgen Twitter und vergessen Sie nicht, bei uns mitzumachen 100.000+ ML SubReddit und Abonnieren Unser E-newsletter. Warten! Bist du im Telegram? Jetzt können Sie uns auch per Telegram kontaktieren.
Asif Razzaq ist CEO von Marktechpost Media Inc.. Als visionärer Unternehmer und Ingenieur setzt sich Asif dafür ein, das Potenzial der künstlichen Intelligenz für das soziale Wohl zu nutzen. Sein jüngstes Unterfangen ist die Einführung einer Medienplattform für künstliche Intelligenz, Marktechpost, die sich durch eine ausführliche Berichterstattung über maschinelles Lernen und Deep-Studying-Nachrichten auszeichnet, die sowohl technisch fundiert als auch für ein breites Publikum leicht verständlich ist. Die Plattform verfügt über mehr als 2 Millionen monatliche Aufrufe, was ihre Beliebtheit beim Publikum verdeutlicht.