[AI-72] LangGraph plugin samples (#289)

* Add LangGraph plugin samples

Seven samples demonstrating the Temporal LangGraph plugin across both the
Graph API and Functional API:

- Human-in-the-loop: interrupt() + Temporal signals for chatbot approval
- Continue-as-new: task result caching across workflow boundaries
- ReAct agent: tool-calling loop with conditional edges / while loop
- Control flow (Functional API only): parallel, for-loop, if/else

Related SDK PR: temporalio/sdk-python#1448

* Remove explicit langchain deps from langgraph group to avoid conflict

* Declare langchain and langgraph as conflicting dependency groups

* Fix import sorting and formatting

* Update dependencies and lint config (temporarily - waiting for SDK PR merge) so that CI otherwise passes

* Add hello_world samples, per-sample READMEs, tests, __init__ docstrings, and configurable client

* Address review: consistent v2 API, unique thread_id, SDK prereq note

* Remove SDK prereq note - PR won't merge until plugin is released

* Add LangSmith tracing sample (Graph API + Functional API)

Combines LangGraphPlugin (durable execution) with LangSmithPlugin
(observability) for full tracing of LLM calls through Temporal workflows.

* Update all samples to match SDK PR #1448 API

- graphs parameter is now a list, not a dict
- No graph()/entrypoint() lookup functions — reference objects directly
- No entrypoints parameter on LangGraphPlugin
- Use set_cache()/get_cache() for continue-as-new caching
- Update tests and READMEs accordingly

* Update LangGraph samples to current sdk-python main API

The plugin API has evolved past PR #1448:
- LangGraphPlugin(graphs=...) now takes dict[str, StateGraph], not list
- Same for entrypoints (dict[str, Pregel])
- Each node/task must declare execute_in: "activity" | "workflow" in metadata
- Workflows look up registered graphs/entrypoints via graph(name) /
  entrypoint(name) instead of importing them directly
- get_cache/set_cache replaced by cache() and graph(name, cache=...) /
  entrypoint(name, cache=...)

Other changes:
- State schemas converted from primitives (str, int) to TypedDict, since
  langgraph's StateT is bound to TypedDict / dataclass / pydantic models
- Module-level graph instances replaced with make_<name>_graph() factories
  because LangGraphPlugin mutates node metadata in-place at construction
  time, which prevented reuse across tests
- langgraph dep group now pulls langchain, langchain-anthropic, and
  temporalio[langgraph,langsmith] so the langsmith_tracing samples and
  human_in_the_loop's RunnableConfig usage actually resolve

* CI: install protoc for building temporalio from sdk-python git source

* Skip LangGraph functional API + interrupt tests on Python < 3.11

These features rely on contextvars propagation through asyncio.create_task(),
which is only available in Python 3.11+. On 3.10 they hang rather than fail
cleanly, so CI sat for an hour before today's fix landed.

Mirrors the pytestmark skipif used in sdk-python's own contrib/langgraph
tests.

* Address review feedback on LangGraph samples

- Drop cross-references between Functional API and Graph API READMEs so
  each stands alone (review comment 1)
- Combine run_worker.py + run_workflow.py into a single main.py for the
  langsmith_tracing samples to reduce setup steps (review comment 2)
- Fix workflow -> Workflow casing (review comment 3)
- Alias temporalio's entrypoint and graph helpers as temporal_entrypoint /
  temporal_graph at import sites to avoid the langgraph.func.entrypoint
  collision (review comment 4)
- Demonstrate @Traceable in three places in both langsmith_tracing samples:
  on the Activity/task itself, on a helper called from the Activity, and
  on a helper called from the Workflow (review comment 5)
- Drop unnecessary single-task list comprehension for activity_options in
  hello_world and langsmith_tracing functional samples (review comment 6)
- Spell out 'temporal server start-dev' in every sample README's
  prerequisites (review comment 7)
- Fix stale README references to get_cache() now that the API is cache()
  (review comment 8)
- Rename arbitrarily-named ETL pipeline functions in continue_as_new to
  honest math names (double / add_50 / triple) per review comment 9

* Update LangGraph samples for temporalio 1.27.0 release

- Drop the [tool.uv.sources] git pin; consume temporalio from PyPI now
  that 1.27.0 ships the LangGraph plugin.
- Keep the protoc CI step around as a permanent fixture so we don't have
  to add and remove it each time we iterate on a new plugin sample.
- Replace dict-comprehension activity_options with explicit per-task
  literals in the Functional API samples.
- Emphasize workflow.wait_condition() in the human-in-the-loop READMEs
  as the durable wait primitive that pairs with interrupt() and signals.
- Clarify that the task result cache is provided by the LangGraph plugin
  (Temporal), not by upstream LangGraph.
- Drop "Same pattern as the Graph API version" cross-references from
  Functional API workflow docstrings; readers pick one API style.
- Tighten the Functional API LangSmith tracing README copy.
- Point the parent README to the released contrib package docs.

Author: DABH

.github/workflows/ci.yml

@@ -33,6 +33,14 @@ jobs:
       - uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6
         with:
           python-version: ${{ matrix.python }}
+      # protoc is needed whenever uv builds temporalio from a git source
+      # (e.g. when [tool.uv.sources] pins a branch while a feature is in
+      # development). Kept on by default so we don't have to add and remove
+      # this step every time we iterate on a new plugin.
+      - uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3
+        with:
+          version: "23.x"
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
       - run: uv tool install poethepoet
       - run: uv sync --group=dsl --group=encryption --group=trio-async
       - run: poe lint

README.md

@@ -72,6 +72,7 @@ Some examples require extra dependencies. See each sample's directory for specif
 * [gevent_async](gevent_async) - Combine gevent and Temporal.
 * [hello_standalone_activity](hello_standalone_activity) - Use activities without using a workflow.
 * [langchain](langchain) - Orchestrate workflows for LangChain.
+* [langgraph_plugin](langgraph_plugin) - Run LangGraph workflows as durable Temporal workflows (Graph API and Functional API).
 * [message_passing/introduction](message_passing/introduction/) - Introduction to queries, signals, and updates.
 * [message_passing/safe_message_handlers](message_passing/safe_message_handlers/) - Safely handling updates and signals.
 * [message_passing/update_with_start/lazy_initialization](message_passing/update_with_start/lazy_initialization/) - Use update-with-start to update a Shopping Cart, starting it if it does not exist.

langgraph_plugin/README.md

@@ -0,0 +1,73 @@
+# LangGraph Plugin Samples
+
+These samples demonstrate the [Temporal LangGraph plugin](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/langgraph), which runs LangGraph workflows as durable Temporal workflows. Each LangGraph graph node (Graph API) or `@task` (Functional API) executes as a Temporal activity with automatic retries, timeouts, and crash recovery.
+
+Samples are organized by API style:
+
+- **Graph API** (`graph_api/`) -- Define workflows as `StateGraph` with nodes and edges.
+- **Functional API** (`functional_api/`) -- Define workflows with `@task` and `@entrypoint` decorators for an imperative programming style.
+
+## Samples
+
+| Sample | Graph API | Functional API | Description |
+|--------|:---------:|:--------------:|-------------|
+| **Hello World** | [graph_api/hello_world](graph_api/hello_world) | [functional_api/hello_world](functional_api/hello_world) | Minimal sample -- single node/task that processes a query string. Start here. |
+| **Human-in-the-loop** | [graph_api/human_in_the_loop](graph_api/human_in_the_loop) | [functional_api/human_in_the_loop](functional_api/human_in_the_loop) | Chatbot that uses `interrupt()` to pause for human approval, Temporal signals to receive feedback, and queries to expose the pending draft. |
+| **Continue-as-new** | [graph_api/continue_as_new](graph_api/continue_as_new) | [functional_api/continue_as_new](functional_api/continue_as_new) | Multi-stage data pipeline that uses `continue-as-new` with task result caching so previously-completed stages are not re-executed. |
+| **ReAct Agent** | [graph_api/react_agent](graph_api/react_agent) | [functional_api/react_agent](functional_api/react_agent) | Tool-calling agent loop. Graph API uses conditional edges; Functional API uses a `while` loop. |
+| **Control Flow** | -- | [functional_api/control_flow](functional_api/control_flow) | Demonstrates parallel task execution, `for` loops, and `if/else` branching -- patterns that are natural in the Functional API. |
+| **LangSmith Tracing** | [graph_api/langsmith_tracing](graph_api/langsmith_tracing) | [functional_api/langsmith_tracing](functional_api/langsmith_tracing) | Combines `LangGraphPlugin` with Temporal's `LangSmithPlugin` for durable execution + full observability of LLM calls. Requires API keys. |
+
+## Prerequisites
+
+1. Install dependencies:
+
+   ```bash
+   uv sync --group langgraph
+   ```
+
+2. Start a [Temporal dev server](https://docs.temporal.io/cli#start-dev-server):
+
+   ```bash
+   temporal server start-dev
+   ```
+
+## Running a Sample
+
+Most samples have two scripts -- start the Worker first, then the Workflow starter in a separate terminal.
+
+```bash
+# Terminal 1: start the Worker
+uv run langgraph_plugin/<api>/<sample>/run_worker.py
+
+# Terminal 2: start the Workflow
+uv run langgraph_plugin/<api>/<sample>/run_workflow.py
+```
+
+For example, to run the Graph API human-in-the-loop chatbot:
+
+```bash
+# Terminal 1
+uv run langgraph_plugin/graph_api/human_in_the_loop/run_worker.py
+
+# Terminal 2
+uv run langgraph_plugin/graph_api/human_in_the_loop/run_workflow.py
+```
+
+The LangSmith Tracing samples bundle the Worker and Workflow execution into a single `main.py`:
+
+```bash
+uv run langgraph_plugin/<api>/langsmith_tracing/main.py
+```
+
+## Key Features Demonstrated
+
+- **Durable execution** -- Every graph node / `@task` runs as a Temporal activity with configurable timeouts and retry policies.
+- **Human-in-the-loop** -- LangGraph's `interrupt()` pauses the graph; Temporal signals deliver human input; queries expose pending state to UIs.
+- **Continue-as-new with caching** -- `cache()` captures completed task results; passing the cache to the next execution avoids re-running them.
+- **Conditional routing** -- Graph API's `add_conditional_edges` and Functional API's native `if/else`/`while` for agent loops.
+- **Parallel execution** -- Functional API launches multiple tasks concurrently by creating futures before awaiting them.
+
+## Related
+
+- [Temporal LangGraph plugin docs](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/langgraph)

langgraph_plugin/__init__.py

@@ -0,0 +1 @@
+"""Temporal LangGraph plugin samples."""

langgraph_plugin/functional_api/__init__.py

@@ -0,0 +1 @@
+"""LangGraph Functional API samples using @task and @entrypoint."""

langgraph_plugin/functional_api/continue_as_new/README.md

@@ -0,0 +1,36 @@
+# Continue-as-New with Caching (Functional API)
+
+Demonstrates combining Temporal's continue-as-new with the LangGraph plugin's task result cache to avoid re-executing completed `@task` functions across workflow boundaries.
+
+## What This Sample Demonstrates
+
+- Task result caching across continue-as-new boundaries with `cache()`
+- Restoring cached results with `entrypoint(name, cache=...)`
+- Each `@task` executes exactly once despite multiple workflow invocations
+
+## How It Works
+
+1. Three tasks run sequentially: `double` (x2) -> `add_50` (+50) -> `triple` (x3).
+2. After the first invocation, the workflow continues-as-new with the cache.
+3. On subsequent invocations, all tasks return cached results instantly.
+4. Input 10 -> 20 -> 70 -> 210.
+
+## Running the Sample
+
+Prerequisites: `uv sync --group langgraph` and a running Temporal dev server (`temporal server start-dev`).
+
+```bash
+# Terminal 1
+uv run langgraph_plugin/functional_api/continue_as_new/run_worker.py
+
+# Terminal 2
+uv run langgraph_plugin/functional_api/continue_as_new/run_workflow.py
+```
+
+## Files
+
+| File | Description |
+|------|-------------|
+| `workflow.py` | `@task` functions, `@entrypoint`, `PipelineInput`, and `PipelineFunctionalWorkflow` |
+| `run_worker.py` | Registers tasks and entrypoint with `LangGraphPlugin`, starts Worker |
+| `run_workflow.py` | Executes the pipeline Workflow and prints the result |

langgraph_plugin/functional_api/continue_as_new/__init__.py

@@ -0,0 +1 @@
+"""Continue-as-new pipeline with task result caching."""

langgraph_plugin/functional_api/continue_as_new/run_worker.py

@@ -0,0 +1,37 @@
+"""Worker for the continue-as-new pipeline (Functional API)."""
+
+import asyncio
+import os
+
+from temporalio.client import Client
+from temporalio.contrib.langgraph import LangGraphPlugin
+from temporalio.worker import Worker
+
+from langgraph_plugin.functional_api.continue_as_new.workflow import (
+    PipelineFunctionalWorkflow,
+    activity_options,
+    all_tasks,
+    pipeline_entrypoint,
+)
+
+
+async def main() -> None:
+    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
+    plugin = LangGraphPlugin(
+        entrypoints={"pipeline": pipeline_entrypoint},
+        tasks=all_tasks,
+        activity_options=activity_options,
+    )
+
+    worker = Worker(
+        client,
+        task_queue="langgraph-pipeline-functional",
+        workflows=[PipelineFunctionalWorkflow],
+        plugins=[plugin],
+    )
+    print("Worker started. Ctrl+C to exit.")
+    await worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

langgraph_plugin/functional_api/continue_as_new/run_workflow.py

@@ -0,0 +1,31 @@
+"""Start the continue-as-new pipeline workflow (Functional API)."""
+
+import asyncio
+import os
+from datetime import timedelta
+
+from temporalio.client import Client
+
+from langgraph_plugin.functional_api.continue_as_new.workflow import (
+    PipelineFunctionalWorkflow,
+    PipelineInput,
+)
+
+
+async def main() -> None:
+    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
+
+    result = await client.execute_workflow(
+        PipelineFunctionalWorkflow.run,
+        PipelineInput(data=10),
+        id="pipeline-functional-workflow",
+        task_queue="langgraph-pipeline-functional",
+        execution_timeout=timedelta(seconds=60),
+    )
+
+    # 10*2=20 -> 20+50=70 -> 70*3=210
+    print(f"Pipeline result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

langgraph_plugin/functional_api/continue_as_new/workflow.py

@@ -0,0 +1,92 @@
+"""Continue-as-new with caching using the LangGraph Functional API with Temporal.
+
+Demonstrates Temporal's continue-as-new with the LangGraph plugin's task
+result cache to avoid re-executing completed @task functions across
+workflow boundaries.
+"""
+
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Any
+
+from langgraph.func import entrypoint, task
+from temporalio import workflow
+from temporalio.contrib.langgraph import cache
+from temporalio.contrib.langgraph import entrypoint as temporal_entrypoint
+
+
+@task
+def double(data: int) -> int:
+    """Stage 1: double the input."""
+    return data * 2
+
+
+@task
+def add_50(data: int) -> int:
+    """Stage 2: add 50."""
+    return data + 50
+
+
+@task
+def triple(data: int) -> int:
+    """Stage 3: triple the result."""
+    return data * 3
+
+
+@entrypoint()
+async def pipeline_entrypoint(data: int) -> dict:
+    """Run the 3-stage pipeline: double -> add_50 -> triple."""
+    doubled = await double(data)
+    plus_50 = await add_50(doubled)
+    tripled = await triple(plus_50)
+    return {"result": tripled}
+
+
+all_tasks = [double, add_50, triple]
+
+activity_options = {
+    "double": {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=30),
+    },
+    "add_50": {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=30),
+    },
+    "triple": {
+        "execute_in": "activity",
+        "start_to_close_timeout": timedelta(seconds=30),
+    },
+}
+
+
+@dataclass
+class PipelineInput:
+    data: int
+    cache: dict[str, Any] | None = None
+    phase: int = 1
+
+
+@workflow.defn
+class PipelineFunctionalWorkflow:
+    """Runs the pipeline, continuing-as-new after each phase.
+
+    Input 10: 10*2=20 -> 20+50=70 -> 70*3=210
+    Each task executes once; phases 2 and 3 use cached results.
+    """
+
+    @workflow.run
+    async def run(self, input_data: PipelineInput) -> dict[str, Any]:
+        app = temporal_entrypoint("pipeline", cache=input_data.cache)
+        result = await app.ainvoke(input_data.data)
+
+        if input_data.phase < 3:
+            workflow.continue_as_new(
+                PipelineInput(
+                    data=input_data.data,
+                    cache=cache(),
+                    phase=input_data.phase + 1,
+                )
+            )
+
+        return result