Implement LangGraph runtime adapter boundary

This commit is contained in:
Daniel Maddern 2026-08-15 17:08:12 +07:00
parent 8109d01992
commit 32b45ea4e6
4 changed files with 92 additions and 8 deletions

View file

@ -1,29 +1,95 @@
from __future__ import annotations
from typing import Any
from typing import Any, TypedDict
from uuid import UUID
from graph.models import GraphRun
from graph.native_runtime import NativeGraphRuntime
from graph.registry import NodeHandlerRegistry
from graph.runtime import GraphRuntime
from graph.spec import ExecutionGraphSpec
class LangGraphState(TypedDict, total=False):
graph_run_id: int
current_node: str
status: str
edge_result: str
class LangGraphRuntime(GraphRuntime):
"""Initial LangGraph adapter placeholder.
"""LangGraph adapter behind Artifex's runtime-neutral graph boundary."""
M1 keeps this deterministic. M2 will wire the autonomous task loop here while
preserving this boundary.
"""
def __init__(self, registry: NodeHandlerRegistry | None = None) -> None:
self.registry = registry
async def start(self, project_id: UUID) -> str:
async def start(self, project_id: UUID | None = None, **kwargs: Any) -> str:
graph_run = kwargs.get("graph_run")
if graph_run is not None:
self.run_until_terminal_or_paused(graph_run)
return str(graph_run.id)
return f"project-{project_id}"
async def pause(self, run_id: str) -> None:
return None
async def resume(self, run_id: str) -> None:
return None
if self.registry is None:
return None
graph_run = GraphRun.objects.get(id=run_id)
self.run_until_terminal_or_paused(graph_run)
async def cancel(self, run_id: str) -> None:
return None
async def signal(self, run_id: str, event: dict[str, Any]) -> None:
return None
def run_until_terminal_or_paused(self, graph_run: GraphRun) -> GraphRun:
if self.registry is None:
raise RuntimeError("LangGraphRuntime requires a NodeHandlerRegistry")
try:
from langgraph.graph import END, StateGraph
except ImportError as exc:
raise RuntimeError("LangGraphRuntime requires the langgraph package") from exc
spec = ExecutionGraphSpec.from_dict(graph_run.execution_graph_version.graph_spec)
native = NativeGraphRuntime(self.registry)
workflow = StateGraph(LangGraphState)
for node_id in spec.nodes:
workflow.add_node(node_id, self._node_runner(native, graph_run, node_id))
workflow.set_entry_point(graph_run.current_node or spec.entry)
for node_id in spec.nodes:
if node_id in spec.terminal_nodes:
workflow.add_edge(node_id, END)
continue
edges = [edge for edge in spec.edges if edge.source == node_id]
if not edges:
workflow.add_edge(node_id, END)
continue
workflow.add_conditional_edges(
node_id,
lambda state: str(state.get("edge_result", "success")),
{edge.condition or "success": edge.target for edge in edges},
)
compiled = workflow.compile()
compiled.invoke({"graph_run_id": graph_run.id, "current_node": graph_run.current_node or spec.entry})
graph_run.refresh_from_db()
return graph_run
def _node_runner(self, native: NativeGraphRuntime, graph_run: GraphRun, node_id: str):
def run_node(state: LangGraphState) -> LangGraphState:
if graph_run.current_node != node_id:
graph_run.current_node = node_id
graph_run.save(update_fields=["current_node", "updated_at"])
native.run_until_terminal_or_paused(graph_run, interrupt_after=node_id)
graph_run.refresh_from_db()
return {
"graph_run_id": graph_run.id,
"current_node": graph_run.current_node,
"status": graph_run.status,
"edge_result": str(graph_run.metadata.get("last_edge_result", "success")),
}
return run_node

View file

@ -9,7 +9,7 @@ class GraphRuntime(ABC):
"""Execution runtime boundary. LangGraph must stay behind this interface."""
@abstractmethod
async def start(self, project_id: UUID) -> str: ...
async def start(self, project_id: UUID | None = None, **kwargs: Any) -> str: ...
@abstractmethod
async def pause(self, run_id: str) -> None: ...

View file

@ -6,6 +6,7 @@ requires-python = ">=3.12"
dependencies = [
"django>=5.1,<6.0",
"psycopg[binary]>=3.2,<4.0",
"langgraph>=0.2,<0.3",
"structlog>=24.4,<25.0",
]

View file

@ -0,0 +1,17 @@
from __future__ import annotations
import importlib.util
import pytest
from graph.langgraph_runtime import LangGraphRuntime
def test_langgraph_runtime_reports_missing_dependency() -> None:
if importlib.util.find_spec("langgraph") is not None:
pytest.skip("langgraph is installed; adapter execution is covered by integration parity tests")
runtime = LangGraphRuntime()
with pytest.raises(RuntimeError, match="langgraph package|NodeHandlerRegistry"):
runtime.run_until_terminal_or_paused(None) # type: ignore[arg-type]