LangGraph vs Locus StateGraph — same workflow, same model (gpt-4o-mini), side by side. LangGraph→real OpenAI; Locus→OCI GenAI via BOAT-OC1.

A developer built and ran an identical multi-step research workflow using both LangGraph (with OpenAI's real API) and Locus StateGraph (with OCI GenAI via BOAT-OC1), both calling gpt-4o-mini. The side-by-side test verified API equivalence claims against actual Locus source code, surfacing four documentation and API gaps that were collected into a follow-up issue list.

| /usr/bin/env python3 | | | """LangGraph vs Locus StateGraph — same workflow, same model, side by side. | | | Workflow in both frameworks | | | ============================= | | | START → research → write → review →┐ | | | ▲ │ | | | └── if confidence < 0.85 ──┘ max 3 iterations | | | │ | | | END | | | State in both | | | =============== | | | topic: str — user input | | | notes: list str — accumulates across research iterations REDUCER | | | draft: str | | | confidence: float — review node's score | | | iter: int — loop counter | | | Model | | | ===== | | | Both sides call gpt-4o-mini : | | | - LangGraph → real OpenAI API OPENAI API KEY | | | - Locus → OCI GenAI openai.gpt-4o-mini BOAT-OC1 session token, | | | saasobservai compartment, us-chicago-1 | | | Plus, this script verifies the API equivalence claims from the prose | | | comparison I wrote — every claim is exercised and any failures are | | | collected into ISSUES at the end. | | | Install | | | ======= | | | uv pip install "locus-sdk oci ==0.2.0b19" langgraph langchain-openai | | | Env | | | === | | | OPENAI API KEY=sk-proj-… | | | OCI PROFILE=BOAT-OC1 | | | OCI COMPARTMENT ID=ocid1.compartment.oc1..… saasobservai prod | | | OCI REGION=us-chicago-1 | | | """ | | | from future import annotations | | | import asyncio | | | import operator | | | import os | | | import time | | | from typing import Annotated, Any, TypedDict | | | from pydantic import BaseModel, Field | | | ---------- LangGraph ------------------------------------------------------ | | | from langgraph.graph import StateGraph as LGStateGraph, START as LG START, END as LG END | | | from langgraph.checkpoint.memory import MemorySaver as LGMemorySaver | | | from langchain openai import ChatOpenAI | | | ---------- Locus ---------------------------------------------------------- | | | from locus.multiagent import StateGraph as LocusStateGraph | | | from locus.multiagent.graph import START as LOCUS START, END as LOCUS END | | | from locus.models import get model as locus get model | | | from locus.core.reducers import Reducer noqa: F401 — exists check | | | TOPIC = "Should a backend team standardising on Python adopt LangGraph in production?" | | | TARGET CONFIDENCE = 0.85 | | | MAX ITER = 3 | | | RESEARCH PROMPT = | | | "You are a tight research analyst. Topic: {topic}\n" | | | "Existing notes so far {n existing} : {notes so far}\n" | | | "Produce TWO new, non-overlapping bullet-point findings ≤ 25 words each . " | | | "Numeric or factual claims preferred. Output the two bullets only." | | | | | | WRITE PROMPT = | | | "You are a technical writer. Topic: {topic}\n" | | | "Research notes:\n{notes}\n\n" | | | "Write a 3-sentence verdict paragraph. Be concrete and decisive." | | | | | | REVIEW PROMPT = | | | "You are a sceptical editor. Topic: {topic}\nDraft:\n{draft}\n\n" | | | "Score the draft from 0.0 to 1.0 on factual confidence + decisiveness + " | | | "coverage . Reply with ONLY the number, e.g. '0.82'. No explanation." | | | | | | =========================================================================== | | | Model adapters — give each framework a sync complete prompt - str lambda | | | =========================================================================== | | | def make openai complete : | | | llm = ChatOpenAI model="gpt-4o-mini", api key=os.environ "OPENAI API KEY" , temperature=0.2 | | | def complete prompt: str - str: | | | return llm.invoke prompt .content.strip | | | return complete | | | def make oci complete : | | | model = locus get model | | | "oci:openai.gpt-4o-mini", | | | profile=os.environ.get "OCI PROFILE", "BOAT-OC1" , | | | compartment id=os.environ "OCI COMPARTMENT ID" , | | | region=os.environ.get "OCI REGION", "us-chicago-1" , | | | max tokens=400, | | | temperature=0.2, | | | | | | locus.ModelProtocol exposes async complete messages, tools=None . | | | from locus.core.messages import Message | | | def complete prompt: str - str: | | | msgs = Message role="user", content=prompt | | | resp = asyncio.run model.complete msgs | | | ModelResponse → text | | | return resp.content or "" .strip | | | return complete | | | def parse score text: str - float: | | | for tok in text.replace ",", " " .split : | | | try: | | | v = float tok | | | if 0.0 <= v <= 1.0: | | | return v | | | except ValueError: | | | continue | | | return 0.5 fallback | | | =========================================================================== | | | IMPLEMENTATION A — LangGraph | | | =========================================================================== | | | class LGState TypedDict, total=False : | | | topic: str | | | notes: Annotated list str , operator.add reducer = list append | | | draft: str | | | confidence: float | | | iter: int | | | def build langgraph : | | | complete = make openai complete | | | def research state: LGState - dict: | | | notes so far = "\n".join state.get "notes", or " none " | | | out = complete RESEARCH PROMPT.format | | | topic=state "topic" , n existing=len state.get "notes", , | | | notes so far=notes so far, | | | | | | new notes = l.strip "-• " .strip for l in out.splitlines if l.strip | | | return {"notes": new notes :2 , "iter": state.get "iter", 0 + 1} | | | def write state: LGState - dict: | | | notes = "\n".join f"- {n}" for n in state "notes" | | | return {"draft": complete WRITE PROMPT.format topic=state "topic" , notes=notes } | | | def review state: LGState - dict: | | | score text = complete REVIEW PROMPT.format topic=state "topic" , draft=state "draft" | | | return {"confidence": parse score score text } | | | def decide state: LGState - str: | | | if state "confidence" = TARGET CONFIDENCE or state.get "iter", 0 = MAX ITER: | | | return LG END | | | return "research" | | | g = LGStateGraph LGState | | | g.add node "research", research | | | g.add node "write", write | | | g.add node "review", review | | | g.add edge LG START, "research" | | | g.add edge "research", "write" | | | g.add edge "write", "review" | | | g.add conditional edges "review", decide, {"research": "research", LG END: LG END} | | | return g.compile checkpointer=LGMemorySaver | | | =========================================================================== | | | IMPLEMENTATION B — Locus StateGraph | | | =========================================================================== | | | class LocusState BaseModel : | | | topic: str = "" | | | notes: list str = Field default factory=list | | | draft: str = "" | | | confidence: float = 0.0 | | | iter: int = 0 | | | def build locus graph : | | | complete = make oci complete | | | def research state: dict - dict: | | | notes so far = "\n".join state.get "notes", or or " none " | | | out = complete RESEARCH PROMPT.format | | | topic=state "topic" , n existing=len state.get "notes", or , | | | notes so far=notes so far, | | | | | | new notes = l.strip "-• " .strip for l in out.splitlines if l.strip :2 | | | Manual append — Locus' reducer behavior is verified separately below. | | | merged = list state.get "notes", or + new notes | | | return {"notes": merged, "iter": state.get "iter", 0 or 0 + 1} | | | def write state: dict - dict: | | | notes = "\n".join f"- {n}" for n in state.get "notes" or | | | return {"draft": complete WRITE PROMPT.format topic=state "topic" , notes=notes } | | | def review state: dict - dict: | | | score text = complete REVIEW PROMPT.format topic=state "topic" , draft=state "draft" | | | return {"confidence": parse score score text } | | | def decide state: dict - str: | | | if state.get "confidence" or 0.0 = TARGET CONFIDENCE or state.get "iter" or 0 = MAX ITER: | | | return LOCUS END | | | return "research" | | | g = LocusStateGraph state schema=LocusState | | | g.add node "research", research | | | g.add node "write", write | | | g.add node "review", review | | | g.add edge LOCUS START, "research" | | | g.add edge "research", "write" | | | g.add edge "write", "review" | | | g.add conditional edges "review", decide, {"research": "research", LOCUS END: LOCUS END} | | | return g.compile | | | =========================================================================== | | | Driver | | | =========================================================================== | | | def run langgraph graph - dict: | | | t0 = time.monotonic | | | state = graph.invoke | | | {"topic": TOPIC, "notes": , "iter": 0}, | | | config={"configurable": {"thread id": "t1"}, "recursion limit": 50}, | | | | | | return { state, " elapsed": time.monotonic - t0} | | | def run locus graph - dict: | | | NOTE: docs example uses graph.compile .run sync ... but only async | | | entry points exist ainvoke , astream , execute , stream . Adapter: | | | t0 = time.monotonic | | | result = asyncio.run graph.ainvoke {"topic": TOPIC, "notes": , "iter": 0} | | | state = result.final state if hasattr result, "final state" else result | | | if hasattr state, "model dump" : | | | state = state.model dump | | | return { state, " elapsed": time.monotonic - t0} | | | =========================================================================== | | | Equivalence verifier — runs my prose claims past reality | | | =========================================================================== | | | def verify equivalence claims - list str : | | | """Returns a list of failure strings — empty means every claim checks out.""" | | | fails: list str = | | | add = fails.append | | | Claim: builder name + signature parity | | | try: | | | from locus.multiagent import StateGraph as LG locus noqa: N814 | | | LG locus state schema=LocusState | | | except Exception as e: noqa: BLE001 | | | add f"StateGraph state schema=… builder shape failed in Locus: {e r}" | | | Claim: Send / Command / interrupt importable from claimed paths | | | try: | | | from locus.core.send import Send noqa: F401 | | | from locus.core.command import Command noqa: F401 | | | from locus.core.interrupt import interrupt noqa: F401 | | | except Exception as e: noqa: BLE001 | | | add f"Send/Command/interrupt import paths failed: {e r}" | | | Claim: Functional API @entrypoint / @task | | | try: | | | from locus.multiagent.functional import entrypoint, task noqa: F401 | | | except Exception as e: noqa: BLE001 | | | add f"@entrypoint / @task not importable from locus.multiagent.functional: {e r}" | | | Claim: Per-node RetryPolicy + CachePolicy | | | try: | | | from locus.multiagent.graph import RetryPolicy, CachePolicy noqa: F401 | | | except Exception as e: noqa: BLE001 | | | add f"RetryPolicy/CachePolicy not at locus.multiagent.graph: {e r}" | | | Claim: draw mermaid + draw ascii both ship | | | try: | | | from locus.multiagent.visualize import draw mermaid, draw ascii noqa: F401 | | | except Exception as e: noqa: BLE001 | | | add f"draw mermaid/draw ascii not importable: {e r}" | | | Claim: GraphConfig holds interrupt before/interrupt after/checkpointer | | | try: | | | from locus.multiagent.graph import GraphConfig | | | cfg = GraphConfig interrupt before= "a" , interrupt after= "b" , checkpointer=None | | | assert cfg.interrupt before == "a" and cfg.interrupt after == "b" | | | except Exception as e: noqa: BLE001 | | | add f"GraphConfig interrupt before/after wiring failed: {e r}" | | | Claim: StreamMode enum has values / updates / nodes / custom | | | try: | | | from locus.multiagent.graph import StreamMode | | | expected = {"values", "updates", "nodes", "custom"} | | | present = {m.value for m in StreamMode} | | | missing = expected - present | | | if missing: | | | add f"StreamMode missing modes: {missing} got {present} " | | | except Exception as e: noqa: BLE001 | | | add f"StreamMode not importable: {e r}" | | | Claim: Reducer-from-Pydantic extraction extract reducers from model | | | try: | | | from locus.core.reducers import extract reducers from model noqa: F401 | | | except Exception as e: noqa: BLE001 | | | add f"extract reducers from model missing from locus.core.reducers: {e r}" | | | Build a compiled graph once for the next several checks. | | | from locus.multiagent.graph import START as S, END as E | | | g0 = LocusStateGraph state schema=LocusState | | | g0.add node "n", lambda s: {} | | | g0.add edge S, "n" ; g0.add edge "n", E | | | compiled = g0.compile | | | Claim from docs/concepts/multi-agent/graph.md line ~73 : | | | result = graph.compile .run sync {...} | | | if not hasattr compiled, "run sync" : | | | add "docs/concepts/multi-agent/graph.md shows compiled.run sync ... " | | | "but no such method exists on the compiled graph " | | | " only async: ainvoke/astream/execute/stream ." | | | Claim from same doc: graph.compile .get mermaid | | | if not hasattr compiled, "get mermaid" : | | | add "docs/concepts/multi-agent/graph.md shows compiled.get mermaid " | | | "but the actual API is from locus.multiagent.visualize import " | | | "draw mermaid; draw mermaid compiled ." | | | Claim from prose: sync invoke parity with LangGraph's CompiledGraph | | | if not hasattr compiled, "invoke" : | | | add "No sync compiled.invoke ... — LangGraph users expect this " | | | "as the standard sync entry. Locus exposes only async ainvoke ." | | | Verify get graph returns something useful for rendering. | | | if hasattr compiled, "get graph" : | | | sub = compiled.get graph | | | if not hasattr sub, "draw mermaid" : | | | add " compiled.get graph returns a StateGraph rather than a " | | | "render-capable object — LangGraph's CompiledGraph.get graph " | | | "returns a Graph with .draw mermaid / .draw ascii / " | | | " .draw png . Convenience gap." | | | return fails | | | =========================================================================== | | | Main | | | =========================================================================== | | | def main - None: | | | for var in "OPENAI API KEY", "OCI COMPARTMENT ID" : | | | if not os.environ.get var : | | | raise SystemExit f"{var} not set" | | | print "=" 76 | | | print " LANGGRAPH vs LOCUS — same workflow, same model family gpt-4o-mini " | | | print f" topic : {TOPIC}" | | | print f" target conf : {TARGET CONFIDENCE} max iter: {MAX ITER}" | | | print "=" 76 | | | ----- Equivalence claim check ---------------------------------------- | | | print "\n— API equivalence claims verified against current Locus source —" | | | fails = verify equivalence claims | | | if not fails: | | | print " ✓ Every claim from the prose comparison resolves cleanly." | | | else: | | | print f" ✗ {len fails } claim s didn't hold up:" | | | for f in fails: | | | print f" • {f}" | | | ----- Run both graphs ------------------------------------------------- | | | print "\n— Building + running LangGraph real OpenAI gpt-4o-mini —" | | | lg graph = build langgraph | | | lg state = run langgraph lg graph | | | print f" iters={lg state 'iter' } notes={len lg state 'notes' } " | | | f"conf={lg state 'confidence' :.2f} elapsed={lg state ' elapsed' :.1f}s" | | | print "\n— Building + running Locus StateGraph OCI gpt-4o-mini via BOAT-OC1 —" | | | locus graph = build locus graph | | | locus state = run locus locus graph | | | print f" iters={locus state 'iter' } notes={len locus state 'notes' } " | | | f"conf={locus state 'confidence' :.2f} elapsed={locus state ' elapsed' :.1f}s" | | | ----- Side-by-side ---------------------------------------------------- | | | print "\n" + "=" 76 | | | print " SIDE-BY-SIDE FINAL STATE" | | | print "=" 76 | | | rows: list tuple str, Any, Any = | | | "iters", lg state "iter" , locus state "iter" , | | | "notes", len lg state "notes" , len locus state "notes" , | | | "confidence", f"{lg state 'confidence' :.2f}", | | | f"{locus state 'confidence' :.2f}" , | | | "elapsed s", f"{lg state ' elapsed' :.2f}", | | | f"{locus state ' elapsed' :.2f}" , | | | | | | print f" {'metric':<14}{'LangGraph':<20}{'Locus':<20}" | | | for name, a, b in rows: | | | print f" {name:<14}{str a :<20}{str b :<20}" | | | print "\n— LangGraph draft —" | | | print " " + lg state "draft" .replace "\n", "\n " | | | print "\n— Locus draft —" | | | print " " + locus state "draft" .replace "\n", "\n " | | | ----- Mermaid from each ---------------------------------------------- | | | print "\n— Mermaid LangGraph —" | | | try: | | | print lg graph.get graph .draw mermaid | | | except Exception as e: noqa: BLE001 | | | print f" LangGraph Mermaid raised: {e r} " | | | print "— Mermaid Locus —" | | | try: | | | from locus.multiagent.visualize import draw mermaid as locus draw mermaid | | | print locus draw mermaid locus graph | | | except Exception as e: noqa: BLE001 | | | print f" Locus Mermaid raised: {e r} " | | | ----- Verdict --------------------------------------------------------- | | | print "\n" + "=" 76 | | | if fails: | | | print f" RESULT: workflow ran in both; {len fails } doc/API gap s for Locus follow-up:" | | | print "=" 76 | | | for i, f in enumerate fails, 1 : | | | print f"\n Issue {i}\n {f}" | | | else: | | | print " RESULT: workflow ran in both; every doc claim verified." | | | print "=" 76 | | | if name == " main ": | | | main |