# Coming from LangGraph

> Keep your LangGraph graph, swap the chat model for a durable one and run the graph inside a Pyrula agent: a concept map, the scaffold and an executed example.

Source: https://docs.pyrula.io/getting-started/coming-from/langgraph/

Build the graph in the body of a Pyrula `@agent` (or `@workflow`), wrap its chat model in
`durable_chat_model`, and run any tool's outside effect through `ctx.step`. The run's journal
replaces the checkpointer: after a crash the body runs again from the top, the graph walks the same
path, and finished model calls are answered from the journal.

To keep LangGraph in charge instead, make one node a durable Pyrula run with `as_langgraph_node`.
Both shapes are in [LangGraph](https://docs.pyrula.io/agents/integrations/langgraph.md).

## Scaffold

```bash
pip install "pyrula[agents,langgraph,langchain]"
pyrula init --langgraph
python app.py
```

See [Start from a scaffold](https://docs.pyrula.io/getting-started/existing-app.md#start-from-a-scaffold).

## Concept map

| LangGraph | Pyrula | Change |
|---|---|---|
| `StateGraph`, nodes, edges | Unchanged, built inside an `@agent` or `@workflow` body | Direct swap |
| Chat model (`ChatOpenAI`, ...) | `durable_chat_model(model)`; pass `model_id=` if it has no `model_name` | Direct swap |
| `model.invoke()`, `graph.invoke()` | `ainvoke`; sync `invoke` raises `UnjournaledSurface` in a run | Small rewrite |
| `model.astream()` | Supported; journaled as one call | Operational change |
| Checkpointer | None; the run's journal is the checkpoint | Conceptual redesign |
| `thread_id` | A Pyrula thread; read earlier turns with `ctx.thread.history(...)` ([HTTP](https://docs.pyrula.io/agents/serving/http.md)) | Conceptual redesign |
| `interrupt()` | `await ctx.interrupt(...)`, optional `timeout=` | Small rewrite |
| `Command(resume=...)` | `WorkflowClient.resume(name, run_id, action, value)`, `/resume`, or Pyrula Cloud | Small rewrite |
| `ToolNode` tool with an effect | Route the effect through `ctx.step`, or it runs again on resume | Small rewrite |
| LangChain tool in a native `@agent` | `from_langchain_tool(tool)` | Direct swap |
| `recursion_limit` | [`AgentLimits`](https://docs.pyrula.io/reference/agents.md#agentlimits); a cycling graph ends with `budget_exceeded` | Small rewrite |
| Parallel model branches | Supported; give each wrapped model its own `name=` | Direct swap |
| Compiled subgraph as a node | Supported | Direct swap |
| `create_react_agent` | Supported | Direct swap |
| `astream(stream_mode="updates")` | Supported | Direct swap |
| `astream(stream_mode="messages")` | Not supported through `durable_chat_model` | Not supported |
| Node `RetryPolicy` | Use it on the node that calls the model | Direct swap |
| `with_fallbacks`, `with_retry` on the model | Not supported; a fallback would call live during replay | Not supported |
| Routing on model output | Must follow the recorded responses, not a clock or random value | Small rewrite |
| Time travel | A [what-if run](https://docs.pyrula.io/workflows/guides/what-if.md) started from Pyrula Cloud | Conceptual redesign |
| Cancelling a graph run | Cancel or pause the run; each model call is a stop point | Direct swap |
| LangGraph as orchestrator | `as_langgraph_node(...)` on a checkpointed graph | Conceptual redesign |
| LangSmith tracing | `pyrula dev`, Pyrula Cloud, or OpenTelemetry ([Observability](https://docs.pyrula.io/agents/observability.md)) | Operational change |
| Where the graph runs | A worker process against a store | Operational change |

## Example

The graph is a plain LangGraph agent. The model keeps asking for the tool, and `max_llm_calls` ends
the run after three calls with `budget_exceeded`.

```python
import asyncio
import operator
from typing import Annotated, Any, TypedDict

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool as lc_tool
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode

from pyrula.agents import AgentContext, AgentLimits, agent
from pyrula.agents.integrations import durable_chat_model
from pyrula.agents.testing import AgentRunner

provider_calls: list[int] = []


class LoopingChat(BaseChatModel):
    """Stands in for a provider that keeps asking for the same tool."""

    model_name: str = "looping-model"

    @property
    def _llm_type(self) -> str:
        return "looping"

    def bind_tools(self, tools, **kwargs):
        return self.bind(tools=[{"name": t.name} for t in tools], **kwargs)

    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
        provider_calls.append(len(messages))
        call = {"id": f"c{len(messages)}", "name": "lookup", "args": {"order": "A-1"}}
        usage = {"input_tokens": 100, "output_tokens": 20, "total_tokens": 120}
        reply = AIMessage(content="", tool_calls=[call], usage_metadata=usage)
        return ChatResult(generations=[ChatGeneration(message=reply)])

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        raise NotImplementedError("durable_chat_model journals the async path only")


class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]


@lc_tool
async def lookup(order: str) -> str:
    """Look up an order."""
    return f"{order}: shipped"


@agent(name="docs_from_langgraph_support", timeout=60, limits=AgentLimits(max_llm_calls=3))
async def support(ctx: AgentContext, message: str) -> str:
    model = durable_chat_model(LoopingChat()).bind_tools([lookup])

    async def call_model(state: State) -> dict[str, Any]:
        return {"messages": [await model.ainvoke(state["messages"])]}

    def route(state: State) -> str:
        return "tools" if state["messages"][-1].tool_calls else END

    graph = StateGraph(State)
    graph.add_node("model", call_model)
    graph.add_node("tools", ToolNode([lookup]))
    graph.add_edge(START, "model")
    graph.add_conditional_edges("model", route, ["tools", END])
    graph.add_edge("tools", "model")

    # No checkpointer: the run's journal is the checkpoint.
    out = await graph.compile().ainvoke({"messages": [HumanMessage(content=message)]})
    return out["messages"][-1].content


async def main() -> None:
    events = await AgentRunner(support).run(message="where is A-1?")
    assert events[-1].kind == "run:error"
    assert events[-1].payload["code"] == "budget_exceeded"
    assert len(provider_calls) == 3


asyncio.run(main())
```

A crash and resume on this graph shape is in
[LangGraph](https://docs.pyrula.io/agents/integrations/langgraph.md#a-graph-inside-a-pyrula-run).