Conversation
|
|
There was a problem hiding this comment.
Pull request overview
Adds a new temporalio.contrib.langgraph integration to run LangGraph Graph/Functional API nodes/tasks as Temporal Activities (with retries/timeouts), including continue-as-new caching and interrupt propagation, plus an accompanying test suite.
Changes:
- Introduces
LangGraphPlugin+ activity/config/cache plumbing for executing LangGraph nodes/tasks via Temporal Activities and supporting interrupts + continue-as-new deduping. - Adds a new
langgraphoptional extra and locks required LangGraph/LangChain dependencies. - Adds an end-to-end test suite covering graphs, subgraphs, interrupts (v1/v2), streaming, timeouts, execute-in-workflow, and continue-as-new caching.
Reviewed changes
Copilot reviewed 22 out of 24 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| uv.lock | Locks new LangGraph/LangChain-related dependencies and adds langgraph extra entries. |
| pyproject.toml | Adds langgraph optional dependency and dev dependency. |
| temporalio/contrib/langgraph/init.py | Exposes LangGraphPlugin, graph, entrypoint, and cache public API. |
| temporalio/contrib/langgraph/langgraph_plugin.py | Implements the plugin: registers activities, sandbox passthrough, and graph/entrypoint accessors. |
| temporalio/contrib/langgraph/activity.py | Activity wrappers for config propagation, interrupt handling, and cache-aware execution. |
| temporalio/contrib/langgraph/langgraph_config.py | Serializes/deserializes LangGraph runnable config (checkpoint namespace/scratchpad) across workflow/activity boundary. |
| temporalio/contrib/langgraph/task_cache.py | Implements a content-addressed task result cache for continue-as-new deduplication. |
| temporalio/contrib/langgraph/README.md | Documents installation and usage for Graph/Functional APIs and activity options. |
| tests/contrib/langgraph/init.py | Adds test package marker for the new test suite. |
| tests/contrib/langgraph/conftest.py | Provides langgraph-specific Temporal test environment and client fixtures. |
| tests/contrib/langgraph/test_two_nodes.py | Verifies basic two-node graph execution via activities. |
| tests/contrib/langgraph/test_timeout.py | Verifies activity timeout behavior is enforced through node metadata. |
| tests/contrib/langgraph/test_subgraph_workflow.py | Verifies subgraph execution with execute_in="workflow". |
| tests/contrib/langgraph/test_subgraph_activity.py | Verifies subgraph execution inside an activity node. |
| tests/contrib/langgraph/test_streaming.py | Verifies streaming (astream) behavior across activity-wrapped nodes. |
| tests/contrib/langgraph/test_interrupt.py | Verifies v1 interrupt propagation and resume semantics. |
| tests/contrib/langgraph/test_interrupt_v2.py | Verifies v2 interrupt semantics (GraphOutput.value/.interrupts) and resume. |
| tests/contrib/langgraph/test_execute_in_workflow.py | Verifies execute_in="workflow" bypasses activity wrapping. |
| tests/contrib/langgraph/test_continue_as_new.py | Verifies continue-as-new with LangGraph checkpointer state restoration. |
| tests/contrib/langgraph/test_continue_as_new_cached.py | Verifies continue-as-new caching prevents re-execution across continuations. |
| tests/contrib/langgraph/test_e2e_functional.py | End-to-end functional API workflows (including continue-as-new + partial execution scenarios). |
| tests/contrib/langgraph/test_e2e_functional_v2.py | End-to-end functional API workflows using version="v2" GraphOutput. |
| tests/contrib/langgraph/e2e_functional_entrypoints.py | Defines functional @task/@entrypoint implementations used by E2E tests. |
| tests/contrib/langgraph/e2e_functional_workflows.py | Defines Temporal workflow wrappers used by functional API E2E tests. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
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
Add a full integration guide for the new LangGraph plugin (temporalio/sdk-python#1448) covering both Graph API and Functional API usage, Activity options, checkpointer setup, human-in-the-loop, continue-as-new with caching, and links to samples. Also adds LangGraph to the Python integrations index, sidebar, plugins guide, and encyclopedia.
…en use init functions for graph compilation
|
|
||
| import langsmith | ||
|
|
||
| # langsmith conditionally imports langchain_core when it is installed. |
There was a problem hiding this comment.
Can you explain why this change is needed for langgraph?
There was a problem hiding this comment.
[edited for accuracy] The langgraph Python package directly depends on the langchain-core Python package, so in an environment where langgraph is installed, langchain-core is installed as well. This is just a env/venv-level thing. The CI pipeline is one such environment. But so in any environment where langchain-core is available, langsmith tries to import it - but fails if it's not passed through for langsmith. You can be running Python code that doesn't import LangGraph or anything, but as long as langchain-core is installed in the env, langsmith will try to import it. So for safety, for the case where a user may have langchain-core installed (e.g., when they are using LangSmith + LangGraph, which should be fairly common), we need to pass through langchain-core into the sandbox.
For this block specifically, my understanding is that this import helps bypass some restrictions on the sandbox and/or helps langsmith get the sandboxed version of langchain-core when it dynamically loads/imports it. Not 100% clarity without digging further into langsmith internals but can add a follow-up if we want to investigate more.
|
|
||
| ```python | ||
| # Graph API | ||
| graph.add_node("my_node", my_node, metadata={"execute_in": "workflow"}) |
There was a problem hiding this comment.
Just to call out, in other integrations, this is the default and activities are done explicitly. Do we want to maintain that discrepancy?
Summary
Details
New files:
Tests cover: two-node graphs, subgraphs (as activities and workflows), interrupts, streaming, timeouts, continue-as-new (with and without caching), execute-in-workflow mode, and full end-to-end functional API workflows.
Test plan