LangGraph Plugin for Temporal Python SDK #1263
Open
+14,131
−2,054
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Setup package structure for LangGraph Temporal integration prototypes. These are throwaway prototypes to validate technical assumptions before implementing the production integration. Package structure: - temporalio/contrib/langgraph/ - Main package (empty for now) - temporalio/contrib/langgraph/_prototypes/ - Validation prototypes Prototypes to implement: 1. Pregel loop - Validate AsyncPregelLoop submit function injection 2. Write capture - Validate CONFIG_KEY_SEND callback mechanism 3. Task interface - Document PregelExecutableTask structure 4. Serialization - Test state/message serialization 5. Graph builder - Test graph reconstruction approaches
Validates that we can inject a custom submit function into LangGraph's Pregel execution loop via CONFIG_KEY_RUNNER_SUBMIT config key. Key findings: - Submit injection works for parallel graph execution - Sequential graphs use fast path and may not call submit - PregelExecutableTask provides: name, id, input, proc, config, writes Tests cover: - Basic graph execution (async/sync) - Submit injection with sequential and parallel graphs - PregelExecutableTask attribute inspection
Import CONFIG_KEY_RUNNER_SUBMIT from langgraph._internal._constants instead of langgraph.constants to avoid the deprecation warning. The mechanism is still used internally by LangGraph - the public export just warns because it's considered private API. We document this decision and note that future LangGraph versions may change this API.
PregelExecutableTask is a dataclass, not a NamedTuple. Update test to use dataclasses.fields() instead of checking _fields attribute. Validates that: - PregelExecutableTask is a dataclass - Has 'writes' field of type deque[tuple[str, Any]] - Writes are captured correctly after task execution
Document PregelExecutableTask dataclass structure: - Core fields: name, id, path, input, proc, config, triggers - Output: writes (deque), writers - Policy: retry_policy, cache_key - Nested: subgraphs Includes config filtering for serialization: - Filters __pregel_* and __lg_* internal keys - Filters non-JSON-serializable values - Preserves user keys and standard config VALIDATION STATUS: PASSED
Validates that LangGraph state can be serialized for Temporal activities: - LangChain messages are Pydantic models (HumanMessage, AIMessage, etc.) - Temporal pydantic_data_converter handles them automatically - Default converter works for basic dict states - End-to-end tests verify workflow -> activity -> workflow round-trip Key findings: - Use pydantic_data_converter for LangChain message types - Configure sandbox to passthrough langchain_core modules - No custom serialization needed VALIDATION STATUS: PASSED
- graph_builder_proto.py: Updated with latest prototype changes - graph_registry_proto.py: Thread-safe graph caching prototype - VALIDATION_SUMMARY.md: Phase 1 validation results and findings These prototypes validated the architecture for Phase 2 production code.
Prototypes are preserved in git history (commit 5521520). Production code is now in: - _models.py, _runner.py, _plugin.py, _activities.py, _graph_registry.py
…ode config Phase 3 - Activity and Write Capture: - Fix activity to inject CONFIG_KEY_SEND callback for proper write capture - Writes are captured via LangGraph internal writer mechanism - Add 3 activity integration tests validating real node execution Phase 4 - Per-Node Configuration: - Support activity_timeout via node metadata - Support task_queue via node metadata - Support heartbeat_timeout via node metadata - Map LangGraph RetryPolicy to Temporal RetryPolicy - Add 4 configuration tests File structure follows OpenAI agents SDK pattern (all internal modules use _ prefix). All 30 tests passing.
Key changes: - Rewrote runner to use LangGraph AsyncPregelLoop for proper graph traversal - Fixed conditional edge routing by merging input_state with captured writes - Added CONFIG_KEY_READ callback to activity for state reading support - Added example.py with customer support agent demonstrating conditional routing - Fixed LangChain message serialization through Pydantic payload converter
Change from sequential to parallel task execution using asyncio.gather(). This improves performance while maintaining BSP (Bulk Synchronous Parallel) correctness - all tasks still complete before after_tick() is called. Before (sequential): Total time = sum of all activity durations After (parallel): Total time ≈ max of activity durations Also adds the consolidated design document (renamed from v3).
- Add native LangGraph interrupt API matching `__interrupt__` return pattern - Support resume with `Command(resume=value)` after interrupt - Track interrupted node name to correctly route resume values - Add PregelScratchpad setup in activities for interrupt() function - Remove GraphInterrupt exception in favor of return value API Tests added: - Unit tests for interrupt models and activity behavior - Integration tests for runner interrupt/resume flow - E2E tests with real Temporal worker for full interrupt cycle
- Fix multi-interrupt resume by preserving completed nodes across invocations instead of resetting them. This prevents nodes like step1 from re-running and re-interrupting when resuming step2. - Merge resumed node writes into input_state before starting the loop to ensure writes are included in final output even if loop doesn't schedule the resumed node. - Add invocation counter for unique activity IDs across workflow replays. - Add comprehensive e2e tests with real Temporal workers testing: - Simple graph execution without interrupts - Single interrupt with signal-based resume (approval flow) - Interrupt with rejection - Multiple sequential interrupts - Fix type errors in test_langgraph.py by renaming lambda parameter from 's' to 'state' to match LangGraph's type annotations.
- Add StateSnapshot model for checkpoint data - Add get_state() method to runner for extracting checkpoints - Add checkpoint parameter to compile() for restoring from checkpoint - Add should_continue callback to ainvoke() for external execution control - Callback is invoked once per tick (BSP superstep) - When should_continue() returns False, returns __checkpoint__ in result - Add unit tests for checkpoint extraction, restoration, and should_continue - Update ContinueAsNewWorkflow example to demonstrate the pattern
Implements Phase 1 of Store support as described in DESIGN_STORE.md: - Add ActivityLocalStore class that captures writes for replay in workflow - Add StoreItem, StoreWrite, StoreSnapshot models for serialization - Update NodeActivityInput/Output with store_snapshot and store_writes - Add _store_state to runner for canonical store data - Inject store via Runtime object so nodes can use get_store() - Include store_state in StateSnapshot for checkpoint/continue-as-new - Add comprehensive unit tests for store models and ActivityLocalStore Store operations work by: 1. Workflow maintains _store_state dict with all store data 2. Before activity: Runner creates StoreSnapshot from current state 3. In activity: ActivityLocalStore serves reads from snapshot, captures writes 4. After activity: Runner applies store_writes to _store_state 5. On checkpoint: Store state is serialized for continue-as-new
- Add test_store_persistence: verifies store data persists across nodes within a single graph invocation (node1 writes, node2 reads) - Add test_store_persistence_across_invocations: verifies store data persists across multiple ainvoke() calls within the same workflow - Fix activity to always create ActivityLocalStore (even when empty) so get_store() works on first invocation - Add MultiInvokeStoreWorkflow and counter_node for multi-invocation test
- Implement Send API for dynamic parallelism (map-reduce patterns) - Add SendPacket model to serialize Send objects - Capture Send objects separately from regular writes in activities - Execute SendPackets as separate activities with Send.arg as input - Add validation tests for Send API, Subgraphs, and Command API - Update MISSING_FEATURES.md to reflect validated features: - Send API: ✅ Implemented - Subgraphs: ✅ Implemented (native Pregel support) - Command API: ✅ Implemented (native Pregel support)
Internal design documents created during Phase 1 implementation. These will be removed in the next commit but preserved in history.
Add comprehensive README for end users covering: - Quick start example - Per-node configuration (timeouts, retries, task queues) - Human-in-the-loop with interrupt() and signals - Store API for cross-node persistence - Continue-as-new for long-running workflows - Compile options reference - Important notes and compatibility table
…ions Add a type-safe helper function for configuring LangGraph nodes with Temporal activity options, replacing untyped dict metadata. Changes: - Add temporal_node_metadata() with all execute_activity options: - schedule_to_close_timeout, schedule_to_start_timeout - start_to_close_timeout, heartbeat_timeout, task_queue - retry_policy, cancellation_type, versioning_intent - summary, priority, run_in_workflow - Consolidate _get_node_* methods into single _get_node_activity_options() - Support dict merge operator (|) for combining with other metadata - Update example.py and README.md to use the new helper - Maintain backwards compatibility with legacy "activity_timeout" key
Replace separate default_activity_timeout, default_max_retries, and default_task_queue parameters with a single defaults parameter that accepts temporal_node_metadata() output. This provides a consistent API where all Temporal configuration uses the same typed helper function, both for per-node metadata and for compile-time defaults. Changes: - compile() now accepts defaults=temporal_node_metadata(...) - TemporalLangGraphRunner.__init__() updated to accept defaults dict - _get_node_activity_options() merges defaults with node metadata - retry_policy priority: node metadata > LangGraph native > defaults > built-in - Updated README, example.py, and tests
- Rename temporal_node_metadata() to node_activity_options() - Rename defaults parameter to default_activity_options - Rename node_config parameter to per_node_activity_options The new naming is clearer about the purpose of each function/parameter and avoids confusion when used outside the langgraph contrib module.
Add default_activity_options and per_node_activity_options parameters to LangGraphPlugin, allowing users to set activity configuration at the plugin level instead of repeating it in every compile() call. Configuration priority (highest to lowest): 1. Node metadata from add_node(metadata=...) 2. per_node_activity_options from compile() 3. per_node_activity_options from LangGraphPlugin() 4. default_activity_options from compile() 5. default_activity_options from LangGraphPlugin() 6. Built-in defaults (5 min timeout, 3 retries) Options at each level are merged, so users can set base defaults at the plugin level and selectively override specific options.
…ic execution Add wrappers that allow LangChain tools and chat models to execute as Temporal activities when running inside a workflow. This enables durable execution of agentic nodes (like create_react_agent) where individual tool calls and LLM invocations are each executed as separate activities. New public APIs: - temporal_tool(): Wraps LangChain tools for activity execution - temporal_model(): Wraps LangChain chat models for activity execution - register_tool(): Register tools for activity-side lookup - register_model(): Register model instances - register_model_factory(): Register model factory functions Internal additions: - Tool and model registries for activity-side lookup - execute_tool and execute_chat_model activities - ToolActivityInput/Output and ChatModelActivityInput/Output models Includes comprehensive tests and e2e test with create_react_agent.
- Add support for LangChain 1.0+ create_agent alongside legacy create_react_agent - Add temporal_node_metadata() helper for combining activity options with run_in_workflow flag - Remove run_in_workflow from node_activity_options() - it should only be specified per-node, not as a default - Update README with Agentic Execution section and Hybrid Execution examples - Update design doc with helper functions documentation (section 5.3.6) - Add tests for temporal_node_metadata()
Reorganize the LangGraph test suite into focused, well-organized files: - conftest.py: Shared fixtures for registry clearing - e2e_graphs.py: All graph builders for E2E tests - e2e_workflows.py: Consolidated workflow definitions - test_e2e.py: All 11 E2E tests in organized classes - test_models.py: Pydantic model tests (21 tests) - test_registry.py: Registry tests (14 tests) - test_plugin.py: Plugin tests (6 tests) - test_runner.py: Runner tests (7 tests) - test_activities.py: Activity tests (7 tests) - test_store.py: Store tests (7 tests) - test_temporal_tool.py: Tool wrapper tests (7 tests) - test_temporal_model.py: Model wrapper tests (7 tests) Fix graph registry to eagerly build graphs at registration time. This ensures graph compilation happens outside the workflow sandbox, avoiding issues with Annotated type resolution inside the sandbox. All 11 E2E workflows now run with sandbox enabled, using imports_passed_through() for langchain imports where needed. Deleted: test_langgraph.py, test_validation.py, test_temporal_tool_model.py Total: 87 tests (76 unit + 11 E2E)
- Add experimental warnings to module docstring, LangGraphPlugin, compile(), temporal_tool(), and temporal_model() - Improve README with introduction section, table of contents, and architecture diagram - Remove UnsandboxedWorkflowRunner usage (e2e tests pass with sandbox) - Mark module as experimental (may be abandoned)
Add comprehensive documentation explaining why we use LangGraph's internal APIs (langgraph._internal.*) for node execution: - CONFIG_KEY_SEND/READ/SCRATCHPAD/RUNTIME/CHECKPOINT_NS - PregelScratchpad class These are needed because we execute nodes as individual activities outside of LangGraph's Pregel execution loop, requiring us to inject the same context Pregel would normally provide. Documents risks and alternatives considered.
- Create separate langgraph_tool_node activity type for tool execution - Extract tool call info from Send packet structure (tool_call_with_context) - Display tool names with arguments in activity summary instead of "tools" - Add unit tests for _build_activity_summary function - Add e2e test verifying summary and activity type in workflow history
- Rename activity types to remove redundant "langgraph" prefix: - langgraph_node -> node - langgraph_tool_node -> tool_node - resume_langgraph_node -> resume_node - Activity summaries now use node metadata "description" if available - Add _get_full_node_metadata helper for accessing node metadata - Add unit tests for node metadata description feature
…nctions These functions wrap LangGraph/LangChain agent creation with automatic Temporal durability: - Auto-wrap model with temporal_model() for durable LLM calls - Auto-wrap tools with temporal_tool() for durable tool execution - Mark agent nodes to run inline in workflow (model/tool calls as activities) This provides fine-grained durability where each LLM call and tool invocation is individually retryable and recoverable.
Update README to show the new durable agent functions as the recommended approach for creating agents with Temporal durability.
Rename the function for creating activity configuration options to a
more generic name. Also update create_durable_agent and
create_durable_react_agent to use model_activity_options and
tool_activity_options parameters that accept activity_options() values.
This provides a consistent API pattern:
create_durable_agent(
model, tools,
model_activity_options=activity_options(...),
tool_activity_options=activity_options(...),
)
The wrapped model was storing a reference to the model instance as a class attribute default, which Pydantic tried to deepcopy. This failed because the model contains HTTP clients with RLocks. Fix by using closure variables and looking up the model from the registry when needed outside workflows.
The wrappers were unnecessary complexity. Since LangGraph nodes already run as Temporal activities, models and tools execute directly inside those activities without needing special wrappers. Deleted: - _temporal_model.py, _temporal_tool.py - wrapper implementations - _model_registry.py, _tool_registry.py - registries - _react_agent.py - create_durable_agent/create_durable_react_agent - Related test files The native LangGraph API (create_react_agent with plain tools) works correctly with the Temporal integration. Added test verifying agentic loop executes multiple iterations.
- Remove references to create_durable_agent, create_durable_react_agent - Remove references to temporal_model, temporal_tool wrappers - Simplify Agentic Execution section to show native LangGraph usage - Update compatibility table - Delete obsolete STYLE_REVIEW.md
…react_agent - Update README to use langchain.agents.create_agent - Add deprecation notes to test code (tests use create_react_agent to minimize dependencies - only langchain-core needed)
…n tests - Add langchain>=1.2.0 and langgraph>=1.0.0 to dev dependencies - Use langgraph.prebuilt.create_react_agent in tests instead of langchain.agents.create_agent due to a bug in the latter - The bug: _fetch_last_ai_and_tool_messages raises UnboundLocalError when messages list has no AIMessage - Suppress deprecation warnings since we intentionally use the deprecated API to avoid the bug
The langchain.agents.create_agent routing edges use isinstance checks for AIMessage which fail when messages are serialized dicts. This commit: 1. Extract nested state from ToolCallWithContext for Send API calls 2. Merge base state with writes (append for lists like messages) 3. Convert serialized message dicts back to LangChain Message objects Also updates test docstrings to reflect create_agent is now working.
- Add model name and user query to activity summaries for model/agent nodes - Extract model name from node metadata, runnable attributes, or closure - Show format like: gpt-4o-mini: "What's the weather in Tokyo?" - Support create_agent closure pattern where model is captured in function - Add comprehensive tests for model name extraction
When a compiled subgraph (like from create_agent) is added as a node in an outer graph, the Temporal integration now automatically detects it and executes each inner node as a separate Temporal activity. This provides finer-grained durability: - Each inner node (e.g., 'model', 'tools') has its own retry/timeout - Worker crashes resume from the last completed inner node - Nested subgraphs are recursively flattened Implementation: - Add _get_subgraph() to detect subgraphs via node.subgraphs attribute - Add _execute_subgraph() to create nested runner for recursive execution - Auto-register subgraphs in registry with composite IDs (parent:node) - Extract branch writes from parent node writers for edge routing - Remove redundant heartbeat on activity start
Fix subgraph writer invocation to properly emit branch writes for conditional edge routing. Previously, branch writes were extracted from static writer attributes, which doesn't work for conditional edges that use routing functions. Now writers are invoked with the merged state (input + subgraph output) to properly execute routing functions and emit correct branch writes. Add unit tests that verify: - create_agent subgraph followed by outer nodes executes correctly - Subgraph followed by conditional edge routes to correct path - Activity counting ensures outer nodes run as activities, not inline
Remove incorrect conditional that checked if invoke() is a coroutine function (it never is) and could block the event loop when falling back to sync invoke(). All LangChain Runnables implement ainvoke.
When a subgraph node raises ParentCommand (e.g., transfer_to_* tools in langgraph-supervisor), the goto targets should route to nodes in the parent graph, not the subgraph. Changes: - Add CommandOutput model for serializable representation of LangGraph Command - Catch ParentCommand exception in activity and return CommandOutput - Store pending parent command in runner for parent graph to handle - Fix _execute_subgraph to return send_packets (was returning empty list) - Add unit tests for activity-level and runner-level ParentCommand handling
Update documentation to reference create_agent from langchain.agents instead of the deprecated create_react_agent from langgraph.prebuilt.
Add error classification to distinguish between errors that should be retried (transient failures) and those that should fail immediately (configuration errors, bugs, bad input). Non-retryable errors include: - Python built-in errors (TypeError, ValueError, KeyError, etc.) - Authentication errors (OpenAI, Anthropic) - Bad request errors (400, 401, 403, 404, 422) Retryable errors include: - Rate limit errors (429) - Server errors (5xx) - Connection/network errors - Generic exceptions (default to retryable for safety) Node execution errors are now wrapped in ApplicationError with the appropriate non_retryable flag, allowing Temporal to skip retries for errors that will never succeed. Also excludes crew-ai directory from mypy to fix duplicate module error.
Previously, activities from Send packets were executed sequentially, waiting for each to complete before starting the next. This defeated the purpose of the Send API for parallel execution. Refactored _execute_send_packets to: 1. Prepare all activity inputs first (assigns step counters) 2. Execute all activities in parallel via asyncio.gather 3. Process results sequentially (handles interrupts, parent commands) Added unit tests to verify parallel execution behavior.
For generic nodes (like "search"), extract query-like fields from the input state to show in the activity summary. This makes it easier to see what each search activity is doing in the Temporal UI. Supported field names: query, search_query, question, input, text, prompt Example: search: "LangGraph definition and core concepts"
Break down large methods for improved readability and maintainability: ainvoke (~215 lines -> ~25 lines): - _prepare_invocation_state: Handle input normalization and resume detection - _prepare_config: Setup config with defaults and filtering - _handle_resume_execution: Execute interrupted node and prepare continuation - _create_pregel_loop: Create the AsyncPregelLoop with proper configuration - _run_pregel_loop: Main loop orchestration - _inject_resumed_writes: Inject cached writes into loop tasks - _get_executable_tasks: Filter tasks that need execution - _check_checkpoint: Check for checkpointing condition - _execute_loop_tasks: Execute tasks sequentially - _finalize_output: Build final output with interrupt markers _execute_subgraph (~175 lines -> ~30 lines): - _create_nested_runner: Create nested runner for subgraph execution - _handle_subgraph_interrupt: Handle interrupt propagation from subgraph - _handle_subgraph_parent_command: Handle parent commands from subgraph - _extract_subgraph_writes: Extract writes from subgraph result - _invoke_subgraph_writers: Invoke parent node writers for edge routing
Introduce InterruptState and ExecutionState dataclasses to organize the ~15 instance variables in TemporalLangGraphRunner into logical groups: InterruptState: - interrupted_state: State snapshot when interrupt occurred - interrupted_node_name: Name of the node that triggered the interrupt - resume_value: Value to resume with after interrupt - resume_used: Whether the resume value has been consumed - is_resume_invocation: Whether current invocation is resuming - pending_interrupt: Pending interrupt from current execution ExecutionState: - step_counter: Counter for unique activity IDs within a step - invocation_counter: Counter for unique activity IDs across replays - completed_nodes_in_cycle: Nodes completed in current resume cycle - resumed_node_writes: Cached writes from resumed nodes - last_output: Last output state for get_state() - pending_parent_command: Pending parent command from subgraph - store_state: Store state for cross-node persistence This improves code organization and makes state management easier to reason about.
Create _constants.py with centralized definitions for: - START_NODE, TOOLS_NODE (node names) - INTERRUPT_KEY, CHECKPOINT_KEY (output keys) - BRANCH_PREFIX (channel prefix) - MODEL_NODE_NAMES, MODEL_NAME_ATTRS (model extraction) Update _runner.py to use these constants instead of hardcoded strings.
Improve testability and code organization by extracting: - _convert_messages_if_needed: Module-level pure function - _merge_channel_value: Module-level pure function - StateReader: Class encapsulating state read logic - _get_null_resume: Module-level function for scratchpad The _interrupt_counter function remains as a small nested function since it requires mutable state capture from the enclosing scope.
Mark as completed: - Long methods refactored into smaller functions - Instance variables grouped into dataclasses - Magic strings extracted to constants module - Nested functions extracted from _execute_node_impl
- Add Any type annotations to agent variables in e2e_graphs.py for mypy - Add __init__ docstrings for GraphAlreadyRegisteredError and GraphRegistry - Fix pydocstyle D402 in invoke() method docstring
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
This PR adds a LangGraph plugin (
temporalio.contrib.langgraph) that enables running LangGraph graphs as durable Temporal workflows. Each graph node executes as a separate Temporal activity, providing:Key Features
Full LangGraph API Support:
interrupt()andCommand(resume=...)for human-in-the-loop workflowsNative Agent Support: Works with
create_agent/create_react_agentwithout special wrappersContinue-as-New: Checkpoint support for long-running workflows via
get_state()Architecture
_plugin.py_graph_registry.py_runner.py_activities.py_models.py_exceptions.py_store.py_constants.pyUsage Example
Test Plan
Notes
langgraph._internal.*) which may change in future versionslanggraph>=0.3.31andlangchain-core>=0.3.31