# pedantic-graph-interrupt
Run interruptible Pydantic Graphs that can be interrupted and resumed.
## Table of Contents
- [pedantic-graph-interrupt](#pedantic-graph-interrupt)
- [Table of Contents](#table-of-contents)
- [Use cases](#use-cases)
- [Installation](#installation)
- [Concepts](#concepts)
- [Usage](#usage)
- [Define graph with interrupt nodes](#define-graph-with-interrupt-nodes)
- [Initialize persistence](#initialize-persistence)
- [Start the graph](#start-the-graph)
## Use cases
* LLM AI chatbots implemented using [pydantic-ai] that need to obtain user
input outside of the graph.
* Document processing workflows where an external approval or human
involvement is required.
* Graphs that can run for a very long time, so their execution must be
broken into smaller continuous chunks. Most often each chunk is executed
in a background worker, when the graph is ready to resume its run.
* And more...
## Installation
```bash
pip install pydantic-graph-interrupt
```
## Concepts
* `InterruptNode` is a special node type that interrupts the graph execution.
* `proceed()` is an alternative to `graph.run()`. It can be used to both start
the graph run from scratch or resume it after an interruption.
## Usage
### Define graph with interrupt nodes
```python
"""graph.py"""
from dataclasses import dataclass, field
from pydantic_graph import BaseNode, End, GraphRunContext
from pydantic_graph_interrupt import InterruptibleGraph, InterruptNode, Required
@dataclass
class MyState:
user_name: str | None = None
messages: list[str] = field(default_factory=list)
@dataclass
class Greet(BaseNode[MyState]):
async def run(self, ctx: GraphRunContext[MyState]) -> "WaitForName":
ctx.state.messages.append("Hello, what is your name?")
return WaitForName()
@dataclass # ↓ This is an interrupt node
class WaitForName(InterruptNode[MyState]):
# ↓ This is a required field that must be provided to resume the graph
user_name: Annotated[str | None, Required] = None
async def run(self, ctx: GraphRunContext[MyState]) -> "Goodbye":
ctx.state.user_name = self.user_name
return Goodbye(user_name=self.user_name)
@dataclass
class Goodbye(BaseNode[MyState]):
user_name: str
async def run(self, ctx: GraphRunContext[MyState]) -> End:
ctx.state.messages.append(f"Goodbye, {self.user_name}!")
return End(ctx.state)
my_graph = InterruptibleGraph(nodes=[Greet, WaitForName, Goodbye])
```
### Initialize persistence
When you just start, you have to initialize the persistence. This defines
the starting node of the graph and sets correct types inside the persistence.
```python
"""steps.py"""
from pathlib import Path
from pydantic_graph.persistence.file import FileStatePersistence
from .graph import MyState, my_graph, Greet
async def start():
persistence = FileStatePersistence(Path("offline_state.json"))
state = MyState()
await my_graph.initialize(Greet(), persistence=persistence, state=state)
```
### Start the graph
```python
"""main.py"""
from .steps import start
await start()
```
[State Persistence]: https://ai.pydantic.dev/graph/#state-persistence
[pydantic-ai]: https://ai.pydantic.dev/
Raw data
{
"_id": null,
"home_page": null,
"name": "pydantic-graph-interrupt",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": "vduseev <vagiz@duseev.com>",
"keywords": "graph, interrupt, pydantic, pydantic-graph",
"author": null,
"author_email": "vduseev <vagiz@duseev.com>",
"download_url": "https://files.pythonhosted.org/packages/2e/67/e4c86337f0da3b02fba0c0a0f59343e95d58bde308cddd9fe21bb8e57c34/pydantic_graph_interrupt-0.1.0.tar.gz",
"platform": null,
"description": "# pedantic-graph-interrupt\n\nRun interruptible Pydantic Graphs that can be interrupted and resumed.\n\n## Table of Contents\n\n- [pedantic-graph-interrupt](#pedantic-graph-interrupt)\n - [Table of Contents](#table-of-contents)\n - [Use cases](#use-cases)\n - [Installation](#installation)\n - [Concepts](#concepts)\n - [Usage](#usage)\n - [Define graph with interrupt nodes](#define-graph-with-interrupt-nodes)\n - [Initialize persistence](#initialize-persistence)\n - [Start the graph](#start-the-graph)\n\n## Use cases\n\n* LLM AI chatbots implemented using [pydantic-ai] that need to obtain user\n input outside of the graph.\n* Document processing workflows where an external approval or human\n involvement is required.\n* Graphs that can run for a very long time, so their execution must be\n broken into smaller continuous chunks. Most often each chunk is executed\n in a background worker, when the graph is ready to resume its run.\n* And more...\n\n## Installation\n\n```bash\npip install pydantic-graph-interrupt\n```\n\n## Concepts\n\n* `InterruptNode` is a special node type that interrupts the graph execution.\n* `proceed()` is an alternative to `graph.run()`. It can be used to both start\n the graph run from scratch or resume it after an interruption.\n\n## Usage\n\n### Define graph with interrupt nodes\n\n```python\n\"\"\"graph.py\"\"\"\nfrom dataclasses import dataclass, field\nfrom pydantic_graph import BaseNode, End, GraphRunContext\nfrom pydantic_graph_interrupt import InterruptibleGraph, InterruptNode, Required\n\n@dataclass\nclass MyState:\n user_name: str | None = None\n messages: list[str] = field(default_factory=list)\n\n@dataclass\nclass Greet(BaseNode[MyState]):\n async def run(self, ctx: GraphRunContext[MyState]) -> \"WaitForName\":\n ctx.state.messages.append(\"Hello, what is your name?\")\n return WaitForName()\n\n@dataclass # \u2193 This is an interrupt node\nclass WaitForName(InterruptNode[MyState]):\n # \u2193 This is a required field that must be provided to resume the graph\n user_name: Annotated[str | None, Required] = None\n\n async def run(self, ctx: GraphRunContext[MyState]) -> \"Goodbye\":\n ctx.state.user_name = self.user_name\n return Goodbye(user_name=self.user_name)\n\n@dataclass\nclass Goodbye(BaseNode[MyState]):\n user_name: str\n\n async def run(self, ctx: GraphRunContext[MyState]) -> End:\n ctx.state.messages.append(f\"Goodbye, {self.user_name}!\")\n return End(ctx.state)\n\nmy_graph = InterruptibleGraph(nodes=[Greet, WaitForName, Goodbye])\n```\n\n### Initialize persistence\n\nWhen you just start, you have to initialize the persistence. This defines\nthe starting node of the graph and sets correct types inside the persistence.\n\n```python\n\"\"\"steps.py\"\"\"\nfrom pathlib import Path\nfrom pydantic_graph.persistence.file import FileStatePersistence\nfrom .graph import MyState, my_graph, Greet\n\nasync def start():\n persistence = FileStatePersistence(Path(\"offline_state.json\"))\n state = MyState()\n await my_graph.initialize(Greet(), persistence=persistence, state=state)\n```\n\n### Start the graph\n\n```python\n\"\"\"main.py\"\"\"\nfrom .steps import start\nawait start()\n```\n\n[State Persistence]: https://ai.pydantic.dev/graph/#state-persistence\n[pydantic-ai]: https://ai.pydantic.dev/\n",
"bugtrack_url": null,
"license": null,
"summary": "Interrupt nodes for pydantic-graph",
"version": "0.1.0",
"project_urls": {
"Documentation": "https://github.com/vduseev/pydantic-graph-interrupt",
"Homepage": "https://github.com/vduseev/pydantic-graph-interrupt",
"Issues": "https://github.com/vduseev/pydantic-graph-interrupt/issues",
"Repository": "https://github.com/vduseev/pydantic-graph-interrupt"
},
"split_keywords": [
"graph",
" interrupt",
" pydantic",
" pydantic-graph"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "48ade34e61e86db8ab28d6106575fa663fac7887feba2159273d7916e26c526f",
"md5": "bf5f2882942397e645ff6ef947e3ac92",
"sha256": "4b46de6466244f83d8cbbc9c20ce6cfb584d7b80c7c2601a28b380f264f476b5"
},
"downloads": -1,
"filename": "pydantic_graph_interrupt-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "bf5f2882942397e645ff6ef947e3ac92",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 13590,
"upload_time": "2025-10-12T22:18:53",
"upload_time_iso_8601": "2025-10-12T22:18:53.920362Z",
"url": "https://files.pythonhosted.org/packages/48/ad/e34e61e86db8ab28d6106575fa663fac7887feba2159273d7916e26c526f/pydantic_graph_interrupt-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "2e67e4c86337f0da3b02fba0c0a0f59343e95d58bde308cddd9fe21bb8e57c34",
"md5": "0bd86dcf268c731cdc4d7c0f23056394",
"sha256": "fd065c197b4e3f327da49b55809e30980734f71ac7f5ac17fb87d597a3e9ac48"
},
"downloads": -1,
"filename": "pydantic_graph_interrupt-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "0bd86dcf268c731cdc4d7c0f23056394",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 15264,
"upload_time": "2025-10-12T22:18:55",
"upload_time_iso_8601": "2025-10-12T22:18:55.325838Z",
"url": "https://files.pythonhosted.org/packages/2e/67/e4c86337f0da3b02fba0c0a0f59343e95d58bde308cddd9fe21bb8e57c34/pydantic_graph_interrupt-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-12 22:18:55",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "vduseev",
"github_project": "pydantic-graph-interrupt",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "pydantic-graph-interrupt"
}
JSON
