Update docs/docs/agents/models.md

Co-authored-by: Lauren Hirata Singh <lauren@langchain.dev>

.github/workflows/deploy_docs.yml

@@ -35,16 +35,7 @@ jobs:
         with:
           filter: "docs/docs/**"
 
-  # TODO: Uncomment this to run on PRs
-  # run-changed-notebooks:
-  #   needs: get-changed-files
-  #   uses: ./.github/workflows/run_notebooks.yml
-  #   secrets: inherit
-  #   with:
-  #     changed-files: ${{ needs.get-changed-files.outputs.changed-files }}
-
   deploy:
-    # needs: run-changed-notebooks
     runs-on: ubuntu-latest
     timeout-minutes: 10 # Job will be cancelled if it runs for more than 10 minutes
     env:

.github/workflows/pr_lint.yml

@@ -39,6 +39,7 @@ jobs:
             scheduler-kafka
             sdk-py
             docs
+            ci
           requireScope: false
           ignoreLabels: |
             ignore-lint-pr-title

.github/workflows/release.yml

@@ -137,7 +137,9 @@ jobs:
     needs:
       - build
       - release-notes
-    permissions: write-all
+    permissions:
+      contents: read
+      id-token: write
     uses: ./.github/workflows/_test_release.yml
     with:
       working-directory: ${{ inputs.working-directory }}

docs/docs/agents/models.md

@@ -70,6 +70,104 @@ When using `create_react_agent` you can specify the model by its name string, wh
       )
       ```
 
+### Dynamic model selection
+
+Pass a callable function to `create_react_agent` to dynamically select the model at runtime. This is useful for scenarios where you want to choose a model based on user input, configuration settings, or other runtime conditions.
+
+The selector function must return an instance of a `BaseChatModel`. If you're using tools, you must bind the tools to the model within the selector function.
+
+```python
+openai_model = init_chat_model("openai:gpt-4o")
+anthropic_model = init_chat_model("anthropic:claude-sonnet-4-20250514")
+
+# highlight-next-line
+def select_model(state, runtime: Runtime[CustomContext]) -> BaseChatModel:
+   if runtime.context.provider == "anthropic":
+      model = anthropic_model
+   elif runtime.context.provider == "openai":
+      model = openai_model
+   else:
+      raise ValueError(f"Unsupported provider: {runtime.context.provider}")
+   # With dynamic model selection, you must bind tools explicitly
+   # highlight-next-line
+   return model.bind_tools(tools_to_use)
+
+agent = create_react_agent(
+   # highlight-next-line
+   select_model, 
+   tools=all_known_tools
+)
+```
+
+!!! version-added "New in LangGraph v0.6"
+
+
+??? example "Extended example: dynamically select model and tools"
+
+      ```python
+      from dataclasses import dataclass
+      from typing import Literal
+      from langchain.chat_models import init_chat_model
+      from langchain_core.language_models import BaseChatModel
+      from langchain_core.tools import tool
+      from langgraph.prebuilt import create_react_agent
+      from langgraph.prebuilt.chat_agent_executor import AgentState
+      from langgraph.runtime import Runtime
+
+      # Define the runtime context
+      @dataclass
+      class CustomContext:
+          provider: Literal["anthropic", "openai"]
+
+      @tool
+      def weather() -> str:
+          """Returns the current weather conditions."""
+          return "It's nice and sunny."
+
+      # Initialize models
+      openai_model = init_chat_model("openai:gpt-4o")
+      anthropic_model = init_chat_model("anthropic:claude-sonnet-4-20250514")
+
+      @dataclass
+      class CustomContext:
+         provider: Literal["anthropic", "openai"]
+
+      # Initialize models
+      openai_model = init_chat_model("openai:gpt-4o")
+      anthropic_model = init_chat_model("anthropic:claude-sonnet-4-20250514")
+
+      # Selector function for model choice
+      def select_model(state: AgentState, runtime: Runtime[CustomContext]) -> BaseChatModel:
+         if runtime.context.provider == "anthropic":
+            model = anthropic_model
+         elif runtime.context.provider == "openai":
+            model = openai_model
+         else:
+            raise ValueError(f"Unsupported provider: {runtime.context.provider}")
+        
+         # With dynamic model selection, you must bind tools explicitly
+         return model.bind_tools([weather])  
+
+      # Create agent with dynamic model selection
+      agent = create_react_agent(select_model, tools=[weather])
+
+      # Invoke with context to select model
+      output = agent.invoke(
+         {
+            "messages": [
+               {
+                  "role": "user",
+                  "content": "Which model is handling this?",
+               }
+            ]
+         },
+         context=CustomContext(provider="openai"),
+      )
+
+      print(output["messages"][-1].text())
+      ```
+
+
 ## Advanced model configuration
 
 ### Disable streaming

docs/docs/cloud/reference/langgraph_server_changelog.md

@@ -4,6 +4,19 @@
 
 ---
 
+## v0.2.108 (2025-07-28)
+- Added compatibility for langgraph v0.6, including new context API support and a migration to enhance context handling in assistant operations.
+
+## v0.2.107 (2025-07-27)
+- Implemented caching for authentication processes to improve performance.
+- Merged count and select queries to improve database query efficiency.
+
+## v0.2.106 (2025-07-27)
+- Log whether run uses resumable streams.
+
+## v0.2.105 (2025-07-27)
+- Added a `/heapdump` endpoint to capture and save JS process heap data.
+
 ## v0.2.103 (2025-07-25)
 - Corrected the metadata endpoint to ensure accurate data retrieval.
 

docs/docs/how-tos/tool-calling.md

@@ -66,6 +66,108 @@ agent = create_react_agent(
 agent.invoke({"messages": [{"role": "user", "content": "what's 42 x 7?"}]})
 ```
 
+### Dynamically select tools
+
+Configure tool availability at runtime based on context:
+
+```python
+from langgraph.runtime import Runtime
+
+@dataclass
+class CustomContext:
+    tools: list[Literal["weather", "compass"]]
+    
+# highlight-next-line
+def configure_model(state: AgentState, runtime: Runtime[CustomContext]):
+    """Configure the model with tools based on runtime context."""
+    selected_tools = [
+        tool
+        for tool in [weather, compass]
+        if tool.name in runtime.context.tools
+    ]
+    return model.bind_tools(selected_tools)
+
+
+agent = create_react_agent(
+    # Dynamically configure the model with tools based on runtime context
+    # highlight-next-line
+    configure_model,
+    # Initialize with all tools available
+    # highlight-next-line
+    tools=[weather, compass]
+)
+```
+
+!!! version-added "Supported with langgraph>=0.6"
+
+??? example "Extended example: dynamically select tools based on context"
+
+    ```python
+    from dataclasses import dataclass
+    from typing import Literal
+
+    from langchain.chat_models import init_chat_model
+    from langchain_core.tools import tool
+
+    from langgraph.prebuilt import create_react_agent
+    from langgraph.prebuilt.chat_agent_executor import AgentState
+    from langgraph.runtime import Runtime
+
+
+    @dataclass
+    class CustomContext:
+        tools: list[Literal["weather", "compass"]]
+
+
+    @tool
+    def weather() -> str:
+        """Returns the current weather conditions."""
+        return "It's nice and sunny."
+
+
+    @tool
+    def compass() -> str:
+        """Returns the direction the user is facing."""
+        return "North"
+
+    model = init_chat_model("anthropic:claude-sonnet-4-20250514")
+
+    # highlight-next-line
+    def configure_model(state: AgentState, runtime: Runtime[CustomContext]):
+        """Configure the model with tools based on runtime context."""
+        selected_tools = [
+            tool
+            for tool in [weather, compass]
+            if tool.name in runtime.context.tools
+        ]
+        return model.bind_tools(selected_tools)
+
+
+    agent = create_react_agent(
+        # Dynamically configure the model with tools based on runtime context
+        # highlight-next-line
+        configure_model,
+        # Initialize with all tools available
+        # highlight-next-line
+        tools=[weather, compass]
+    )
+
+    output = agent.invoke(
+        {
+            "messages": [
+                {
+                    "role": "user",
+                    "content": "Who are you and what tools do you have access to?",
+                }
+            ]
+        },
+        # highlight-next-line
+        context=CustomContext(tools=["weather"]),  # Only enable the weather tool
+    )
+
+    print(output["messages"][-1].text())
+    ```
+
 ## Use in a workflow
 
 If you are writing a custom workflow, you will need to:

docs/yarn.lock

@@ -152,6 +152,14 @@ base64-js@^1.5.1:
   resolved "https://registry.yarnpkg.com/base64-js/-/base64-js-1.5.1.tgz#1b1b440160a5bf7ad40b650f095963481903930a"
   integrity sha512-AKpaYlHn8t4SVbOHCy+b5+KKgvR4vrsD8vbvrbiQJps7fKDTkjkDry6ji0rUJjC0kzbNePLwzxq8iypo41qeWA==
 
+call-bind-apply-helpers@^1.0.1, call-bind-apply-helpers@^1.0.2:
+  version "1.0.2"
+  resolved "https://registry.yarnpkg.com/call-bind-apply-helpers/-/call-bind-apply-helpers-1.0.2.tgz#4b5428c222be985d79c3d82657479dbe0b59b2d6"
+  integrity sha512-Sp1ablJ0ivDkSzjcaJdxEunN5/XvksFJ2sMBFfq6x0ryhQV/2b/KwFe21cMpmHtPOSij8K99/wSfoEuTObmuMQ==
+  dependencies:
+    es-errors "^1.3.0"
+    function-bind "^1.1.2"
+
 camelcase@6:
   version "6.3.0"
   resolved "https://registry.yarnpkg.com/camelcase/-/camelcase-6.3.0.tgz#5685b95eb209ac9c0c177467778c9c84df58ba9a"
@@ -201,6 +209,42 @@ delayed-stream@~1.0.0:
   resolved "https://registry.yarnpkg.com/delayed-stream/-/delayed-stream-1.0.0.tgz#df3ae199acadfb7d440aaae0b29e2272b24ec619"
   integrity sha512-ZySD7Nf91aLB0RxL4KGrKHBXl7Eds1DAmEdcoVawXnLD7SDhpNgtuII2aAkg7a7QS41jxPSZ17p4VdGnMHk3MQ==
 
+dunder-proto@^1.0.1:
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/dunder-proto/-/dunder-proto-1.0.1.tgz#d7ae667e1dc83482f8b70fd0f6eefc50da30f58a"
+  integrity sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==
+  dependencies:
+    call-bind-apply-helpers "^1.0.1"
+    es-errors "^1.3.0"
+    gopd "^1.2.0"
+
+es-define-property@^1.0.1:
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/es-define-property/-/es-define-property-1.0.1.tgz#983eb2f9a6724e9303f61addf011c72e09e0b0fa"
+  integrity sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g==
+
+es-errors@^1.3.0:
+  version "1.3.0"
+  resolved "https://registry.yarnpkg.com/es-errors/-/es-errors-1.3.0.tgz#05f75a25dab98e4fb1dcd5e1472c0546d5057c8f"
+  integrity sha512-Zf5H2Kxt2xjTvbJvP2ZWLEICxA6j+hAmMzIlypy4xcBg1vKVnx89Wy0GbS+kf5cwCVFFzdCFh2XSCFNULS6csw==
+
+es-object-atoms@^1.0.0, es-object-atoms@^1.1.1:
+  version "1.1.1"
+  resolved "https://registry.yarnpkg.com/es-object-atoms/-/es-object-atoms-1.1.1.tgz#1c4f2c4837327597ce69d2ca190a7fdd172338c1"
+  integrity sha512-FGgH2h8zKNim9ljj7dankFPcICIK9Cp5bm+c2gQSYePhpaG5+esrLODihIorn+Pe6FGJzWhXQotPv73jTaldXA==
+  dependencies:
+    es-errors "^1.3.0"
+
+es-set-tostringtag@^2.1.0:
+  version "2.1.0"
+  resolved "https://registry.yarnpkg.com/es-set-tostringtag/-/es-set-tostringtag-2.1.0.tgz#f31dbbe0c183b00a6d26eb6325c810c0fd18bd4d"
+  integrity sha512-j6vWzfrGVfyXxge+O0x5sh6cvxAog0a/4Rdd2K36zCMV5eJ+/+tOAngRO8cODMNWbVRdVlmGZQL2YS3yR8bIUA==
+  dependencies:
+    es-errors "^1.3.0"
+    get-intrinsic "^1.2.6"
+    has-tostringtag "^1.0.2"
+    hasown "^2.0.2"
+
 event-lite@^0.1.1:
   version "0.1.3"
   resolved "https://registry.yarnpkg.com/event-lite/-/event-lite-0.1.3.tgz#3dfe01144e808ac46448f0c19b4ab68e403a901d"
@@ -222,12 +266,14 @@ form-data-encoder@1.7.2:
   integrity sha512-qfqtYan3rxrnCk1VYaA4H+Ms9xdpPqvLZa6xmMgFvhO32x7/3J/ExcTd6qpxM0vH2GdMI+poehyBZvqfMTto8A==
 
 form-data@^4.0.0:
-  version "4.0.1"
-  resolved "https://registry.yarnpkg.com/form-data/-/form-data-4.0.1.tgz#ba1076daaaa5bfd7e99c1a6cb02aa0a5cff90d48"
-  integrity sha512-tzN8e4TX8+kkxGPK8D5u0FNmjPUjw3lwC9lSLxxoB/+GtsJG91CO8bSWy73APlgAZzZbXEYZJuxjkHH2w+Ezhw==
+  version "4.0.4"
+  resolved "https://registry.yarnpkg.com/form-data/-/form-data-4.0.4.tgz#784cdcce0669a9d68e94d11ac4eea98088edd2c4"
+  integrity sha512-KrGhL9Q4zjj0kiUt5OO4Mr/A/jlI2jDYs5eHBpYHPcBEVSiipAvn2Ko2HnPe20rmcuuvMHNdZFp+4IlGTMF0Ow==
   dependencies:
     asynckit "^0.4.0"
     combined-stream "^1.0.8"
+    es-set-tostringtag "^2.1.0"
+    hasown "^2.0.2"
     mime-types "^2.1.12"
 
 formdata-node@^4.3.2:
@@ -238,11 +284,69 @@ formdata-node@^4.3.2:
     node-domexception "1.0.0"
     web-streams-polyfill "4.0.0-beta.3"
 
+function-bind@^1.1.2:
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/function-bind/-/function-bind-1.1.2.tgz#2c02d864d97f3ea6c8830c464cbd11ab6eab7a1c"
+  integrity sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA==
+
+get-intrinsic@^1.2.6:
+  version "1.3.0"
+  resolved "https://registry.yarnpkg.com/get-intrinsic/-/get-intrinsic-1.3.0.tgz#743f0e3b6964a93a5491ed1bffaae054d7f98d01"
+  integrity sha512-9fSjSaos/fRIVIp+xSJlE6lfwhES7LNtKaCBIamHsjr2na1BiABJPo0mOjjz8GJDURarmCPGqaiVg5mfjb98CQ==
+  dependencies:
+    call-bind-apply-helpers "^1.0.2"
+    es-define-property "^1.0.1"
+    es-errors "^1.3.0"
+    es-object-atoms "^1.1.1"
+    function-bind "^1.1.2"
+    get-proto "^1.0.1"
+    gopd "^1.2.0"
+    has-symbols "^1.1.0"
+    hasown "^2.0.2"
+    math-intrinsics "^1.1.0"
+
+get-proto@^1.0.1:
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/get-proto/-/get-proto-1.0.1.tgz#150b3f2743869ef3e851ec0c49d15b1d14d00ee1"
+  integrity sha512-sTSfBjoXBp89JvIKIefqw7U2CCebsc74kiY6awiGogKtoSGbgjYE/G/+l9sF3MWFPNc9IcoOC4ODfKHfxFmp0g==
+  dependencies:
+    dunder-proto "^1.0.1"
+    es-object-atoms "^1.0.0"
+
+gopd@^1.2.0:
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/gopd/-/gopd-1.2.0.tgz#89f56b8217bdbc8802bd299df6d7f1081d7e51a1"
+  integrity sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg==
+
 has-flag@^4.0.0:
   version "4.0.0"
   resolved "https://registry.yarnpkg.com/has-flag/-/has-flag-4.0.0.tgz#944771fd9c81c81265c4d6941860da06bb59479b"
   integrity sha512-EykJT/Q1KjTWctppgIAgfSO0tKVuZUjhgMr17kqTumMl6Afv3EISleU7qZUzoXDFTAHTDC4NOoG/ZxU3EvlMPQ==
 
+has-symbols@^1.0.3, has-symbols@^1.1.0:
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/has-symbols/-/has-symbols-1.1.0.tgz#fc9c6a783a084951d0b971fe1018de813707a338"
+  integrity sha512-1cDNdwJ2Jaohmb3sg4OmKaMBwuC48sYni5HUw2DvsC8LjGTLK9h+eb1X6RyuOHe4hT0ULCW68iomhjUoKUqlPQ==
+
+has-tostringtag@^1.0.2:
+  version "1.0.2"
+  resolved "https://registry.yarnpkg.com/has-tostringtag/-/has-tostringtag-1.0.2.tgz#2cdc42d40bef2e5b4eeab7c01a73c54ce7ab5abc"
+  integrity sha512-NqADB8VjPFLM2V0VvHUewwwsw0ZWBaIdgo+ieHtK3hasLz4qeCRjYcqfB6AQrBggRKppKF8L52/VqdVsO47Dlw==
+  dependencies:
+    has-symbols "^1.0.3"
+
+hasown@^2.0.2:
+  version "2.0.2"
+  resolved "https://registry.yarnpkg.com/hasown/-/hasown-2.0.2.tgz#003eaf91be7adc372e84ec59dc37252cedb80003"
+  integrity sha512-0hJU9SCPvmMzIBdZFqNPXWa6dqh7WdH0cII9y+CyS8rG3nL48Bclra9HmKhVVUHyPWNH5Y7xDwAB7bfgSjkUMQ==
+  dependencies:
+    function-bind "^1.1.2"
+
+he@^1.2.0:
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/he/-/he-1.2.0.tgz#84ae65fa7eafb165fddb61566ae14baf05664f0f"
+  integrity sha512-F/1DnUGPopORZi0ni+CvrCgHQ5FyEAHRLSApuYWMmrbSwoN2Mn/7k+Gl38gJnR7yyDZk6WLXwiGod1JOWNDKGw==
+
 humanize-ms@^1.2.1:
   version "1.2.1"
   resolved "https://registry.yarnpkg.com/humanize-ms/-/humanize-ms-1.2.1.tgz#c46e3159a293f6b896da29316d8b6fe8bb79bbed"
@@ -295,6 +399,11 @@ json-stringify-safe@^5.0.1:
     semver "^7.6.3"
     uuid "^10.0.0"
 
+math-intrinsics@^1.1.0:
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/math-intrinsics/-/math-intrinsics-1.1.0.tgz#a0dd74be81e2aa5c2f27e65ce283605ee4e2b7f9"
+  integrity sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g==
+
 mime-db@1.52.0:
   version "1.52.0"
   resolved "https://registry.yarnpkg.com/mime-db/-/mime-db-1.52.0.tgz#bbabcdc02859f4987301c856e3387ce5ec43bf70"

libs/checkpoint-sqlite/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint-sqlite"
-version = "2.0.10"
+version = "2.0.11"
 description = "Library with a SQLite implementation of LangGraph checkpoint saver."
 authors = []
 requires-python = ">=3.9"

libs/checkpoint-sqlite/uv.lock

@@ -346,7 +346,7 @@ dev = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "2.0.10"
+version = "2.0.11"
 source = { editable = "." }
 dependencies = [
     { name = "aiosqlite" },

libs/langgraph/langgraph/utils/__init__.py

@@ -0,0 +1 @@
+"""Legacy utilities module, to be removed in v1."""

libs/langgraph/langgraph/utils/config.py

@@ -0,0 +1,4 @@
+"""Backwards compat imports for config utilities, to be removed in v1."""
+
+from langgraph._internal._config import ensure_config, patch_configurable  # noqa: F401
+from langgraph.config import get_config, get_store  # noqa: F401

libs/langgraph/langgraph/utils/runnable.py

@@ -0,0 +1,3 @@
+"""Backwards compat imports for runnable utilities, to be removed in v1."""
+
+from langgraph._internal._runnable import RunnableCallable, RunnableLike  # noqa: F401

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "0.6.0a1"
+version = "0.6.0"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.9"

libs/langgraph/utils/runnable.py

@@ -1,2 +0,0 @@
-# import for backwards compatibility
-from langgraph._internal._runnable import RunnableCallable, RunnableSeq  # noqa: F401

libs/langgraph/uv.lock

@@ -1192,7 +1192,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "0.6.0a1"
+version = "0.6.0"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1364,7 +1364,7 @@ dev = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "2.0.10"
+version = "2.0.11"
 source = { editable = "../checkpoint-sqlite" }
 dependencies = [
     { name = "aiosqlite" },

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

@@ -1,6 +1,7 @@
 import inspect
 from typing import (
     Any,
+    Awaitable,
     Callable,
     Literal,
     Optional,
@@ -44,8 +45,10 @@ from langgraph.graph.state import CompiledStateGraph
 from langgraph.managed import IsLastStep, RemainingSteps
 from langgraph.prebuilt._internal import ToolCallWithContext
 from langgraph.prebuilt.tool_node import ToolNode
+from langgraph.runtime import Runtime
 from langgraph.store.base import BaseStore
 from langgraph.types import Checkpointer, Send
+from langgraph.typing import ContextT
 from langgraph.warnings import LangGraphDeprecatedSinceV10
 
 StructuredResponse = Union[dict, BaseModel]
@@ -245,437 +248,13 @@ def _validate_chat_history(
     raise ValueError(error_message)
 
 
-class _AgentBuilder:
-    """Internal builder class for constructing React agents with intuitive method-to-node mapping."""
-
-    def __init__(
-        self,
-        model: Union[str, LanguageModelLike],
-        tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
-        *,
-        prompt: Optional[Prompt] = None,
-        response_format: Optional[
-            Union[StructuredResponseSchema, tuple[str, StructuredResponseSchema]]
-        ] = None,
-        pre_model_hook: Optional[RunnableLike] = None,
-        post_model_hook: Optional[RunnableLike] = None,
-        state_schema: Optional[StateSchemaType] = None,
-        context_schema: Optional[Type[Any]] = None,
-        version: Literal["v1", "v2"] = "v2",
-        name: Optional[str] = None,
-    ):
-        # Store all parameters
-        self.model = model
-        self.tools = tools
-        self.prompt = prompt
-        self.response_format = response_format
-        self.pre_model_hook = pre_model_hook
-        self.post_model_hook = post_model_hook
-        self.state_schema = state_schema
-        self.context_schema = context_schema
-        self.version = version
-        self.name = name
-
-        # Setup tools
-        if isinstance(self.tools, ToolNode):
-            self._tool_classes = list(self.tools.tools_by_name.values())
-            self._tool_node = self.tools
-        else:
-            self._llm_builtin_tools = [t for t in self.tools if isinstance(t, dict)]
-            self._tool_node = ToolNode(
-                [t for t in self.tools if not isinstance(t, dict)]
-            )
-            self._tool_classes = list(self._tool_node.tools_by_name.values())
-
-        self._should_return_direct: set[str] = {
-            t.name for t in self._tool_classes if t.return_direct
-        }
-
-        # Setup state schema
-        if self.state_schema is not None:
-            required_keys = {"messages", "remaining_steps"}
-            if self.response_format is not None:
-                required_keys.add("structured_response")
-
-            schema_keys = set(get_type_hints(self.state_schema))
-            if missing_keys := required_keys - schema_keys:
-                raise ValueError(
-                    f"Missing required key(s) {missing_keys} in state_schema"
-                )
-
-            self._final_state_schema = self.state_schema
-        else:
-            self._final_state_schema = (
-                AgentStateWithStructuredResponse
-                if self.response_format is not None
-                else AgentState
-            )
-
-        # Setup model
-        model = self.model
-
-        # Convert string models
-        if isinstance(model, str):
-            try:
-                from langchain.chat_models import init_chat_model  # type: ignore[import-not-found]
-            except ImportError:
-                raise ImportError(
-                    "Please install langchain (`pip install langchain`) to use '<provider>:<model>' string syntax for `model` parameter."
-                )
-            model = cast(BaseChatModel, init_chat_model(model))
-
-        # Bind tools if needed
-        if (
-            _should_bind_tools(
-                model, self._tool_classes, num_builtin=len(self._llm_builtin_tools)
-            )
-            and len(self._tool_classes + self._llm_builtin_tools) > 0
-        ):
-            model = cast(BaseChatModel, model).bind_tools(
-                self._tool_classes + self._llm_builtin_tools
-            )  # type: ignore[operator]
-
-        self._model_runnable = _get_prompt_runnable(self.prompt) | model
-
-    def create_model_node(self) -> RunnableCallable:
-        """Create the 'agent' node that calls the LLM."""
-
-        def _get_model_input_state(state: StateSchema) -> StateSchema:
-            if self.pre_model_hook is not None:
-                messages: Optional[Sequence[BaseMessage]] = (
-                    _get_state_value(state, "llm_input_messages")
-                ) or _get_state_value(state, "messages")
-                error_msg: str = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
-            else:
-                messages = _get_state_value(state, "messages")
-                error_msg = f"Expected input to call_model to have 'messages' key, but got {state}"
-
-            if messages is None:
-                raise ValueError(error_msg)
-
-            _validate_chat_history(messages)
-
-            if isinstance(self._final_state_schema, type) and issubclass(
-                self._final_state_schema, BaseModel
-            ):
-                state.messages = messages  # type: ignore
-            else:
-                state["messages"] = messages  # type: ignore
-            return state
-
-        def _are_more_steps_needed(state: StateSchema, response: BaseMessage) -> bool:
-            has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
-            all_tools_return_direct = (
-                all(
-                    call["name"] in self._should_return_direct
-                    for call in response.tool_calls
-                )
-                if isinstance(response, AIMessage)
-                else False
-            )
-            remaining_steps = _get_state_value(state, "remaining_steps", None)
-            is_last_step = _get_state_value(state, "is_last_step", False)
-            return (
-                (remaining_steps is None and is_last_step and has_tool_calls)
-                or (
-                    remaining_steps is not None
-                    and remaining_steps < 1
-                    and all_tools_return_direct
-                )
-                or (
-                    remaining_steps is not None
-                    and remaining_steps < 2
-                    and has_tool_calls
-                )
-            )
-
-        def call_model(state: StateSchema, config: RunnableConfig) -> StateSchema:
-            state = _get_model_input_state(state)
-            response = cast(AIMessage, self._model_runnable.invoke(state, config))  # type: ignore[union-attr]
-            response.name = self.name
-
-            if _are_more_steps_needed(state, response):
-                return {
-                    "messages": [
-                        AIMessage(
-                            id=response.id,
-                            content="Sorry, need more steps to process this request.",
-                        )
-                    ]
-                }
-            return {"messages": [response]}
-
-        async def acall_model(
-            state: StateSchema, config: RunnableConfig
-        ) -> StateSchema:
-            state = _get_model_input_state(state)
-            response = cast(
-                AIMessage, await self._model_runnable.ainvoke(state, config)
-            )  # type: ignore[union-attr]
-            response.name = self.name
-
-            if _are_more_steps_needed(state, response):
-                return {
-                    "messages": [
-                        AIMessage(
-                            id=response.id,
-                            content="Sorry, need more steps to process this request.",
-                        )
-                    ]
-                }
-            return {"messages": [response]}
-
-        # Determine input schema
-        input_schema = self._final_state_schema
-        if self.pre_model_hook is not None:
-            if isinstance(self._final_state_schema, type) and issubclass(
-                self._final_state_schema, BaseModel
-            ):
-                from pydantic import create_model
-
-                input_schema = create_model(
-                    "CallModelInputSchema",
-                    llm_input_messages=(list[AnyMessage], ...),
-                    __base__=self._final_state_schema,
-                )
-            else:
-
-                class CallModelInputSchema(self._final_state_schema):  # type: ignore
-                    llm_input_messages: list[AnyMessage]
-
-                input_schema = CallModelInputSchema
-
-        return RunnableCallable(call_model, acall_model, input_schema=input_schema)
-
-    def create_structured_response_node(self) -> Optional[RunnableCallable]:
-        """Create the 'generate_structured_response' node if configured."""
-        if self.response_format is None:
-            return None
-
-        def generate_structured_response(
-            state: StateSchema, config: RunnableConfig
-        ) -> StateSchema:
-            messages = _get_state_value(state, "messages")
-            structured_response_schema = self.response_format
-            if isinstance(self.response_format, tuple):
-                system_prompt, structured_response_schema = self.response_format
-                messages = [SystemMessage(content=system_prompt)] + list(messages)
-
-            model_with_structured_output = _get_model(
-                self._model_runnable
-            ).with_structured_output(  # type: ignore[arg-type]
-                cast(StructuredResponseSchema, structured_response_schema)
-            )
-            response = model_with_structured_output.invoke(messages, config)
-            return {"structured_response": response}
-
-        async def agenerate_structured_response(
-            state: StateSchema, config: RunnableConfig
-        ) -> StateSchema:
-            messages = _get_state_value(state, "messages")
-            structured_response_schema = self.response_format
-            if isinstance(self.response_format, tuple):
-                system_prompt, structured_response_schema = self.response_format
-                messages = [SystemMessage(content=system_prompt)] + list(messages)
-
-            model_with_structured_output = _get_model(
-                self._model_runnable
-            ).with_structured_output(  # type: ignore[arg-type]
-                cast(StructuredResponseSchema, structured_response_schema)
-            )
-            response = await model_with_structured_output.ainvoke(messages, config)
-            return {"structured_response": response}
-
-        return RunnableCallable(
-            generate_structured_response, agenerate_structured_response
-        )
-
-    def create_model_router(self) -> Callable[[StateSchema], Union[str, list[Send]]]:
-        """Create routing function for model node conditional edges."""
-
-        def should_continue(state: StateSchema) -> Union[str, list[Send]]:
-            messages = _get_state_value(state, "messages")
-            last_message = messages[-1]
-
-            if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
-                if self.post_model_hook is not None:
-                    return "post_model_hook"
-                elif self.response_format is not None:
-                    return "generate_structured_response"
-                else:
-                    return END
-            else:
-                if self.version == "v1":
-                    return "tools"
-                elif self.version == "v2":
-                    if self.post_model_hook is not None:
-                        return "post_model_hook"
-                    return [
-                        Send(
-                            "tools",
-                            ToolCallWithContext(
-                                __type="tool_call_with_context",
-                                tool_call=tool_call,
-                                state=state,
-                            ),
-                        )
-                        for tool_call in last_message.tool_calls
-                    ]
-
-        return should_continue
-
-    def post_model_hook_router(self, state: StateSchema) -> Union[str, list[Send]]:
-        """Route to the next node after post_model_hook."""
-        messages = _get_state_value(state, "messages")
-        tool_messages = [m.tool_call_id for m in messages if isinstance(m, ToolMessage)]
-        last_ai_message = next(
-            m for m in reversed(messages) if isinstance(m, AIMessage)
-        )
-        pending_tool_calls = [
-            c for c in last_ai_message.tool_calls if c["id"] not in tool_messages
-        ]
-
-        if pending_tool_calls:
-            return [
-                Send(
-                    "tools",
-                    ToolCallWithContext(
-                        __type="tool_call_with_context",
-                        tool_call=tool_call,
-                        state=state,
-                    ),
-                )
-                for tool_call in pending_tool_calls
-            ]
-        elif isinstance(messages[-1], ToolMessage):
-            return self._get_entry_point()
-        elif self.response_format is not None:
-            return "generate_structured_response"
-        else:
-            return END
-
-    def create_tools_router(self) -> Optional[Callable[[StateSchema], str]]:
-        """Create routing function for tools node conditional edges."""
-        if not self._should_return_direct:
-            return None
-
-        def route_tool_responses(state: StateSchema) -> str:
-            messages = _get_state_value(state, "messages")
-            for m in reversed(messages):
-                if not isinstance(m, ToolMessage):
-                    break
-                if m.name in self._should_return_direct:
-                    return END
-
-            if isinstance(m, AIMessage) and m.tool_calls:
-                if any(
-                    call["name"] in self._should_return_direct for call in m.tool_calls
-                ):
-                    return END
-
-            return self._get_entry_point()
-
-        return route_tool_responses
-
-    def _get_entry_point(self) -> str:
-        """Get the workflow entry point."""
-        return "pre_model_hook" if self.pre_model_hook else "agent"
-
-    def _has_tools(self) -> bool:
-        """Check if agent has tools enabled."""
-        return len(self._tool_classes) > 0
-
-    def _get_model_edges(self) -> list[str]:
-        """Get possible edge destinations from model node."""
-        edges = []
-
-        # If post_model_hook exists, we don't add edges here - we use direct edge instead
-        if not self.post_model_hook:
-            if self._has_tools():
-                edges.append("tools")
-            if self.response_format:
-                edges.append("generate_structured_response")
-            if not self._has_tools() and not self.response_format:
-                edges.append(END)
-
-        return edges
-
-    def _get_post_model_hook_edges(self) -> list[str]:
-        """Get possible edge destinations from post_model_hook node."""
-        edges = [self._get_entry_point()]
-        if self._has_tools():
-            edges.append("tools")
-        if self.response_format:
-            edges.append("generate_structured_response")
-        else:
-            edges.append(END)
-        return edges
-
-    def build(self) -> StateGraph:
-        """Build the agent workflow graph (uncompiled)."""
-        # Create workflow
-        workflow = StateGraph(
-            state_schema=self._final_state_schema,  # type: ignore[arg-type]
-            context_schema=self.context_schema,
-        )
-
-        # Add nodes
-        # Always add model node (named 'agent' for backwards compatibility)
-        workflow.add_node("agent", self.create_model_node())
-
-        # Add tools node if needed
-        if self._has_tools():
-            workflow.add_node("tools", self._tool_node)
-
-        # Add hook nodes if configured
-        if self.pre_model_hook:
-            workflow.add_node("pre_model_hook", self.pre_model_hook)  # type: ignore[arg-type]
-        if self.post_model_hook:
-            workflow.add_node("post_model_hook", self.post_model_hook)  # type: ignore[arg-type]
-
-        # Add structured response node if configured
-        structured_node = self.create_structured_response_node()
-        if structured_node:
-            workflow.add_node("generate_structured_response", structured_node)
-
-        # Add edges
-        entry_point = self._get_entry_point()
-        workflow.set_entry_point(entry_point)
-
-        # Pre-model hook edge
-        if self.pre_model_hook:
-            workflow.add_edge("pre_model_hook", "agent")
-
-        # Model node edges
-        if self.post_model_hook:
-            # Direct edge from model node to post_model_hook when post_model_hook exists
-            workflow.add_edge("agent", "post_model_hook")
-            # Post-model hook conditional edges
-            post_hook_edges = self._get_post_model_hook_edges()
-            workflow.add_conditional_edges(
-                "post_model_hook", self.post_model_hook_router, path_map=post_hook_edges
-            )  # type: ignore[arg-type]
-        else:
-            # Conditional edges from model node when no post_model_hook
-            model_router = self.create_model_router()
-            model_edges = self._get_model_edges()
-            workflow.add_conditional_edges("agent", model_router, path_map=model_edges)  # type: ignore[arg-type]
-
-        # Tools edges
-        if self._has_tools():
-            tools_router = self.create_tools_router()
-            if tools_router:
-                workflow.add_conditional_edges(
-                    "tools", tools_router, path_map=[entry_point, END]
-                )
-            else:
-                workflow.add_edge("tools", entry_point)
-
-        return workflow
-
-
 def create_react_agent(
-    model: Union[str, LanguageModelLike],
+    model: Union[
+        str,
+        LanguageModelLike,
+        Callable[[StateSchema, Runtime[ContextT]], BaseChatModel],
+        Callable[[StateSchema, Runtime[ContextT]], Awaitable[BaseChatModel]],
+    ],
     tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
     *,
     prompt: Optional[Prompt] = None,
@@ -700,7 +279,43 @@ def create_react_agent(
     For more details on using `create_react_agent`, visit [Agents](https://langchain-ai.github.io/langgraph/agents/overview/) documentation.
 
     Args:
-        model: The `LangChain` chat model that supports tool calling.
+        model: The language model for the agent. Supports static and dynamic
+            model selection.
+
+            - **Static model**: A chat model instance (e.g., `ChatOpenAI()`) or
+              string identifier (e.g., `"openai:gpt-4"`)
+            - **Dynamic model**: A callable with signature
+              `(state, runtime) -> BaseChatModel` that returns different models
+              based on runtime context
+
+            Dynamic functions receive graph state and runtime, enabling
+            context-dependent model selection. Must return a `BaseChatModel`
+            instance. For tool calling, bind tools using `.bind_tools()`.
+            Bound tools must be a subset of the `tools` parameter.
+
+            Dynamic model example:
+            ```python
+            from dataclasses import dataclass
+
+            @dataclass
+            class ModelContext:
+                model_name: str = "gpt-3.5-turbo"
+
+            # Instantiate models globally
+            gpt4_model = ChatOpenAI(model="gpt-4")
+            gpt35_model = ChatOpenAI(model="gpt-3.5-turbo")
+
+            def select_model(state: AgentState, runtime: Runtime[ModelContext]) -> ChatOpenAI:
+                model_name = runtime.context.model_name
+                model = gpt4_model if model_name == "gpt-4" else gpt35_model
+                return model.bind_tools(tools)
+            ```
+
+            !!! note "Dynamic Model Requirements"
+                Ensure returned models have appropriate tools bound via
+                `.bind_tools()` and support required functionality. Bound tools
+                must be a subset of those specified in the `tools` parameter.
+
         tools: A list of tools or a ToolNode instance.
             If an empty list is provided, the agent will consist of a single LLM node without tool calling.
         prompt: An optional prompt for the LLM. Can take a few different forms:
@@ -840,7 +455,6 @@ def create_react_agent(
             print(chunk)
         ```
     """
-    # Handle deprecated config_schema parameter
     if (
         config_schema := deprecated_kwargs.pop("config_schema", MISSING)
     ) is not MISSING:
@@ -852,29 +466,471 @@ def create_react_agent(
         if context_schema is not None:
             context_schema = config_schema
 
-    # Validate version
     if version not in ("v1", "v2"):
         raise ValueError(
             f"Invalid version {version}. Supported versions are 'v1' and 'v2'."
         )
 
-    # Build the graph using the internal builder
-    builder = _AgentBuilder(
-        model=model,
-        tools=tools,
-        prompt=prompt,
-        response_format=response_format,
-        pre_model_hook=pre_model_hook,
-        post_model_hook=post_model_hook,
-        state_schema=state_schema,
-        context_schema=context_schema,
-        version=version,
-        name=name,
+    if state_schema is not None:
+        required_keys = {"messages", "remaining_steps"}
+        if response_format is not None:
+            required_keys.add("structured_response")
+
+        schema_keys = set(get_type_hints(state_schema))
+        if missing_keys := required_keys - set(schema_keys):
+            raise ValueError(f"Missing required key(s) {missing_keys} in state_schema")
+
+    if state_schema is None:
+        state_schema = (
+            AgentStateWithStructuredResponse
+            if response_format is not None
+            else AgentState
+        )
+
+    llm_builtin_tools: list[dict] = []
+    if isinstance(tools, ToolNode):
+        tool_classes = list(tools.tools_by_name.values())
+        tool_node = tools
+    else:
+        llm_builtin_tools = [t for t in tools if isinstance(t, dict)]
+        tool_node = ToolNode([t for t in tools if not isinstance(t, dict)])
+        tool_classes = list(tool_node.tools_by_name.values())
+
+    is_dynamic_model = not isinstance(model, (str, Runnable)) and callable(model)
+    is_async_dynamic_model = is_dynamic_model and inspect.iscoroutinefunction(model)
+
+    tool_calling_enabled = len(tool_classes) > 0
+
+    if not is_dynamic_model:
+        if isinstance(model, str):
+            try:
+                from langchain.chat_models import (  # type: ignore[import-not-found]
+                    init_chat_model,
+                )
+            except ImportError:
+                raise ImportError(
+                    "Please install langchain (`pip install langchain`) to "
+                    "use '<provider>:<model>' string syntax for `model` parameter."
+                )
+
+            model = cast(BaseChatModel, init_chat_model(model))
+
+        if (
+            _should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools))  # type: ignore[arg-type]
+            and len(tool_classes + llm_builtin_tools) > 0
+        ):
+            model = cast(BaseChatModel, model).bind_tools(
+                tool_classes + llm_builtin_tools  # type: ignore[operator]
+            )
+
+        static_model: Optional[Runnable] = _get_prompt_runnable(prompt) | model  # type: ignore[operator]
+    else:
+        # For dynamic models, we'll create the runnable at runtime
+        static_model = None
+
+    # If any of the tools are configured to return_directly after running,
+    # our graph needs to check if these were called
+    should_return_direct = {t.name for t in tool_classes if t.return_direct}
+
+    def _resolve_model(
+        state: StateSchema, runtime: Runtime[ContextT]
+    ) -> LanguageModelLike:
+        """Resolve the model to use, handling both static and dynamic models."""
+        if is_dynamic_model:
+            return _get_prompt_runnable(prompt) | model(state, runtime)  # type: ignore[operator]
+        else:
+            return static_model
+
+    async def _aresolve_model(
+        state: StateSchema, runtime: Runtime[ContextT]
+    ) -> LanguageModelLike:
+        """Async resolve the model to use, handling both static and dynamic models."""
+        if is_async_dynamic_model:
+            resolved_model = await model(state, runtime)  # type: ignore[misc,operator]
+            return _get_prompt_runnable(prompt) | resolved_model
+        elif is_dynamic_model:
+            return _get_prompt_runnable(prompt) | model(state, runtime)  # type: ignore[operator]
+        else:
+            return static_model
+
+    def _are_more_steps_needed(state: StateSchema, response: BaseMessage) -> bool:
+        has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
+        all_tools_return_direct = (
+            all(call["name"] in should_return_direct for call in response.tool_calls)
+            if isinstance(response, AIMessage)
+            else False
+        )
+        remaining_steps = _get_state_value(state, "remaining_steps", None)
+        is_last_step = _get_state_value(state, "is_last_step", False)
+        return (
+            (remaining_steps is None and is_last_step and has_tool_calls)
+            or (
+                remaining_steps is not None
+                and remaining_steps < 1
+                and all_tools_return_direct
+            )
+            or (remaining_steps is not None and remaining_steps < 2 and has_tool_calls)
+        )
+
+    def _get_model_input_state(state: StateSchema) -> StateSchema:
+        if pre_model_hook is not None:
+            messages = (
+                _get_state_value(state, "llm_input_messages")
+            ) or _get_state_value(state, "messages")
+            error_msg = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
+        else:
+            messages = _get_state_value(state, "messages")
+            error_msg = (
+                f"Expected input to call_model to have 'messages' key, but got {state}"
+            )
+
+        if messages is None:
+            raise ValueError(error_msg)
+
+        _validate_chat_history(messages)
+        # we're passing messages under `messages` key, as this is expected by the prompt
+        if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
+            state.messages = messages  # type: ignore
+        else:
+            state["messages"] = messages  # type: ignore
+
+        return state
+
+    # Define the function that calls the model
+    def call_model(
+        state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
+    ) -> StateSchema:
+        if is_async_dynamic_model:
+            msg = (
+                "Async model callable provided but agent invoked synchronously. "
+                "Use agent.ainvoke() or agent.astream(), or "
+                "provide a sync model callable."
+            )
+            raise RuntimeError(msg)
+
+        model_input = _get_model_input_state(state)
+
+        if is_dynamic_model:
+            # Resolve dynamic model at runtime and apply prompt
+            dynamic_model = _resolve_model(state, runtime)
+            response = cast(AIMessage, dynamic_model.invoke(model_input, config))  # type: ignore[arg-type]
+        else:
+            response = cast(AIMessage, static_model.invoke(model_input, config))  # type: ignore[union-attr]
+
+        # add agent name to the AIMessage
+        response.name = name
+
+        if _are_more_steps_needed(state, response):
+            return {
+                "messages": [
+                    AIMessage(
+                        id=response.id,
+                        content="Sorry, need more steps to process this request.",
+                    )
+                ]
+            }
+        # We return a list, because this will get added to the existing list
+        return {"messages": [response]}
+
+    async def acall_model(
+        state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
+    ) -> StateSchema:
+        model_input = _get_model_input_state(state)
+
+        if is_dynamic_model:
+            # Resolve dynamic model at runtime and apply prompt
+            # (supports both sync and async)
+            dynamic_model = await _aresolve_model(state, runtime)
+            response = cast(AIMessage, await dynamic_model.ainvoke(model_input, config))  # type: ignore[arg-type]
+        else:
+            response = cast(AIMessage, await static_model.ainvoke(model_input, config))  # type: ignore[union-attr]
+
+        # add agent name to the AIMessage
+        response.name = name
+        if _are_more_steps_needed(state, response):
+            return {
+                "messages": [
+                    AIMessage(
+                        id=response.id,
+                        content="Sorry, need more steps to process this request.",
+                    )
+                ]
+            }
+        # We return a list, because this will get added to the existing list
+        return {"messages": [response]}
+
+    input_schema: StateSchemaType
+    if pre_model_hook is not None:
+        # Dynamically create a schema that inherits from state_schema and adds 'llm_input_messages'
+        if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
+            # For Pydantic schemas
+            from pydantic import create_model
+
+            input_schema = create_model(
+                "CallModelInputSchema",
+                llm_input_messages=(list[AnyMessage], ...),
+                __base__=state_schema,
+            )
+        else:
+            # For TypedDict schemas
+            class CallModelInputSchema(state_schema):  # type: ignore
+                llm_input_messages: list[AnyMessage]
+
+            input_schema = CallModelInputSchema
+    else:
+        input_schema = state_schema
+
+    def generate_structured_response(
+        state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
+    ) -> StateSchema:
+        if is_async_dynamic_model:
+            msg = (
+                "Async model callable provided but agent invoked synchronously. "
+                "Use agent.ainvoke() or agent.astream(), or provide a sync model callable."
+            )
+            raise RuntimeError(msg)
+
+        messages = _get_state_value(state, "messages")
+        structured_response_schema = response_format
+        if isinstance(response_format, tuple):
+            system_prompt, structured_response_schema = response_format
+            messages = [SystemMessage(content=system_prompt)] + list(messages)
+
+        resolved_model = _resolve_model(state, runtime)
+        model_with_structured_output = _get_model(
+            resolved_model
+        ).with_structured_output(
+            cast(StructuredResponseSchema, structured_response_schema)
+        )
+        response = model_with_structured_output.invoke(messages, config)
+        return {"structured_response": response}
+
+    async def agenerate_structured_response(
+        state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
+    ) -> StateSchema:
+        messages = _get_state_value(state, "messages")
+        structured_response_schema = response_format
+        if isinstance(response_format, tuple):
+            system_prompt, structured_response_schema = response_format
+            messages = [SystemMessage(content=system_prompt)] + list(messages)
+
+        resolved_model = await _aresolve_model(state, runtime)
+        model_with_structured_output = _get_model(
+            resolved_model
+        ).with_structured_output(
+            cast(StructuredResponseSchema, structured_response_schema)
+        )
+        response = await model_with_structured_output.ainvoke(messages, config)
+        return {"structured_response": response}
+
+    if not tool_calling_enabled:
+        # Define a new graph
+        workflow = StateGraph(state_schema=state_schema, context_schema=context_schema)
+        workflow.add_node(
+            "agent",
+            RunnableCallable(call_model, acall_model),
+            input_schema=input_schema,
+        )
+        if pre_model_hook is not None:
+            workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
+            workflow.add_edge("pre_model_hook", "agent")
+            entrypoint = "pre_model_hook"
+        else:
+            entrypoint = "agent"
+
+        workflow.set_entry_point(entrypoint)
+
+        if post_model_hook is not None:
+            workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
+            workflow.add_edge("agent", "post_model_hook")
+
+        if response_format is not None:
+            workflow.add_node(
+                "generate_structured_response",
+                RunnableCallable(
+                    generate_structured_response,
+                    agenerate_structured_response,
+                ),
+            )
+            if post_model_hook is not None:
+                workflow.add_edge("post_model_hook", "generate_structured_response")
+            else:
+                workflow.add_edge("agent", "generate_structured_response")
+
+        return workflow.compile(
+            checkpointer=checkpointer,
+            store=store,
+            interrupt_before=interrupt_before,
+            interrupt_after=interrupt_after,
+            debug=debug,
+            name=name,
+        )
+
+    # Define the function that determines whether to continue or not
+    def should_continue(state: StateSchema) -> Union[str, list[Send]]:
+        messages = _get_state_value(state, "messages")
+        last_message = messages[-1]
+        # If there is no function call, then we finish
+        if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
+            if post_model_hook is not None:
+                return "post_model_hook"
+            elif response_format is not None:
+                return "generate_structured_response"
+            else:
+                return END
+        # Otherwise if there is, we continue
+        else:
+            if version == "v1":
+                return "tools"
+            elif version == "v2":
+                if post_model_hook is not None:
+                    return "post_model_hook"
+                return [
+                    Send(
+                        "tools",
+                        ToolCallWithContext(
+                            __type="tool_call_with_context",
+                            tool_call=tool_call,
+                            state=state,
+                        ),
+                    )
+                    for tool_call in last_message.tool_calls
+                ]
+
+    # Define a new graph
+    workflow = StateGraph(
+        state_schema=state_schema or AgentState, context_schema=context_schema
     )
 
-    workflow = builder.build()
+    # Define the two nodes we will cycle between
+    workflow.add_node(
+        "agent",
+        RunnableCallable(call_model, acall_model),
+        input_schema=input_schema,
+    )
+    workflow.add_node("tools", tool_node)
 
-    # Compile and return the graph
+    # Optionally add a pre-model hook node that will be called
+    # every time before the "agent" (LLM-calling node)
+    if pre_model_hook is not None:
+        workflow.add_node("pre_model_hook", pre_model_hook)  # type: ignore[arg-type]
+        workflow.add_edge("pre_model_hook", "agent")
+        entrypoint = "pre_model_hook"
+    else:
+        entrypoint = "agent"
+
+    # Set the entrypoint as `agent`
+    # This means that this node is the first one called
+    workflow.set_entry_point(entrypoint)
+
+    agent_paths = []
+    post_model_hook_paths = [entrypoint, "tools"]
+
+    # Add a post model hook node if post_model_hook is provided
+    if post_model_hook is not None:
+        workflow.add_node("post_model_hook", post_model_hook)  # type: ignore[arg-type]
+        agent_paths.append("post_model_hook")
+        workflow.add_edge("agent", "post_model_hook")
+    else:
+        agent_paths.append("tools")
+
+    # Add a structured output node if response_format is provided
+    if response_format is not None:
+        workflow.add_node(
+            "generate_structured_response",
+            RunnableCallable(
+                generate_structured_response,
+                agenerate_structured_response,
+            ),
+        )
+        if post_model_hook is not None:
+            post_model_hook_paths.append("generate_structured_response")
+        else:
+            agent_paths.append("generate_structured_response")
+    else:
+        if post_model_hook is not None:
+            post_model_hook_paths.append(END)
+        else:
+            agent_paths.append(END)
+
+    if post_model_hook is not None:
+
+        def post_model_hook_router(state: StateSchema) -> Union[str, list[Send]]:
+            """Route to the next node after post_model_hook.
+
+            Routes to one of:
+            * "tools": if there are pending tool calls without a corresponding message.
+            * "generate_structured_response": if no pending tool calls exist and response_format is specified.
+            * END: if no pending tool calls exist and no response_format is specified.
+            """
+
+            messages = _get_state_value(state, "messages")
+            tool_messages = [
+                m.tool_call_id for m in messages if isinstance(m, ToolMessage)
+            ]
+            last_ai_message = next(
+                m for m in reversed(messages) if isinstance(m, AIMessage)
+            )
+            pending_tool_calls = [
+                c for c in last_ai_message.tool_calls if c["id"] not in tool_messages
+            ]
+
+            if pending_tool_calls:
+                return [
+                    Send(
+                        "tools",
+                        ToolCallWithContext(
+                            __type="tool_call_with_context",
+                            tool_call=tool_call,
+                            state=state,
+                        ),
+                    )
+                    for tool_call in pending_tool_calls
+                ]
+            elif isinstance(messages[-1], ToolMessage):
+                return entrypoint
+            elif response_format is not None:
+                return "generate_structured_response"
+            else:
+                return END
+
+        workflow.add_conditional_edges(
+            "post_model_hook",
+            post_model_hook_router,  # type: ignore[arg-type]
+            path_map=post_model_hook_paths,
+        )
+
+    workflow.add_conditional_edges(
+        "agent",
+        should_continue,  # type: ignore[arg-type]
+        path_map=agent_paths,
+    )
+
+    def route_tool_responses(state: StateSchema) -> str:
+        for m in reversed(_get_state_value(state, "messages")):
+            if not isinstance(m, ToolMessage):
+                break
+            if m.name in should_return_direct:
+                return END
+
+        # handle a case of parallel tool calls where
+        # the tool w/ `return_direct` was executed in a different `Send`
+        if isinstance(m, AIMessage) and m.tool_calls:
+            if any(call["name"] in should_return_direct for call in m.tool_calls):
+                return END
+
+        return entrypoint
+
+    if should_return_direct:
+        workflow.add_conditional_edges(
+            "tools", route_tool_responses, path_map=[entrypoint, END]
+        )
+    else:
+        workflow.add_edge("tools", entrypoint)
+
+    # Finally, we compile it!
+    # This compiles it into a LangChain Runnable,
+    # meaning you can use it as you would any other runnable
     return workflow.compile(
         checkpointer=checkpointer,
         store=store,

libs/prebuilt/tests/test_react_agent.py

@@ -13,10 +13,12 @@ from typing import (
 )
 
 import pytest
+from langchain_core.language_models import BaseChatModel
 from langchain_core.messages import (
     AIMessage,
     AnyMessage,
     HumanMessage,
+    MessageLikeRepresentation,
     RemoveMessage,
     SystemMessage,
     ToolCall,
@@ -52,6 +54,7 @@ from langgraph.prebuilt.tool_node import (
     _get_state_args,
     _infer_handled_types,
 )
+from langgraph.runtime import Runtime
 from langgraph.store.base import BaseStore
 from langgraph.store.memory import InMemoryStore
 from langgraph.types import Command, Interrupt, interrupt
@@ -1092,7 +1095,7 @@ def test_inspect_react() -> None:
 
 @pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
 def test_react_with_subgraph_tools(
-    sync_checkpointer: BaseCheckpointSaver, version: str
+    sync_checkpointer: BaseCheckpointSaver, version: Literal["v1", "v2"]
 ) -> None:
     class State(TypedDict):
         a: int
@@ -1367,6 +1370,376 @@ def test_get_model() -> None:
         _get_model(RunnableLambda(lambda message: message))
 
 
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_basic(version: str) -> None:
+    """Test basic dynamic model functionality."""
+
+    def dynamic_model(state, runtime: Runtime):
+        # Return different models based on state
+        if "urgent" in state["messages"][-1].content:
+            return FakeToolCallingModel(tool_calls=[])
+        else:
+            return FakeToolCallingModel(tool_calls=[])
+
+    agent = create_react_agent(dynamic_model, [], version=version)
+
+    result = agent.invoke({"messages": [HumanMessage("hello")]})
+    assert len(result["messages"]) == 2
+    assert result["messages"][-1].content == "hello"
+
+    result = agent.invoke({"messages": [HumanMessage("urgent help")]})
+    assert len(result["messages"]) == 2
+    assert result["messages"][-1].content == "urgent help"
+
+
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_with_tools(version: Literal["v1", "v2"]) -> None:
+    """Test dynamic model with tool calling."""
+
+    @dec_tool
+    def basic_tool(x: int) -> str:
+        """Basic tool."""
+        return f"basic: {x}"
+
+    @dec_tool
+    def advanced_tool(x: int) -> str:
+        """Advanced tool."""
+        return f"advanced: {x}"
+
+    def dynamic_model(state: dict, runtime: Runtime) -> BaseChatModel:
+        # Return model with different behaviors based on message content
+        if "advanced" in state["messages"][-1].content:
+            return FakeToolCallingModel(
+                tool_calls=[
+                    [{"args": {"x": 1}, "id": "1", "name": "advanced_tool"}],
+                    [],
+                ]
+            )
+        else:
+            return FakeToolCallingModel(
+                tool_calls=[[{"args": {"x": 1}, "id": "1", "name": "basic_tool"}], []]
+            )
+
+    agent = create_react_agent(
+        dynamic_model, [basic_tool, advanced_tool], version=version
+    )
+
+    # Test basic tool usage
+    result = agent.invoke({"messages": [HumanMessage("basic request")]})
+    assert len(result["messages"]) == 3
+    tool_message = result["messages"][-1]
+    assert tool_message.content == "basic: 1"
+    assert tool_message.name == "basic_tool"
+
+    # Test advanced tool usage
+    result = agent.invoke({"messages": [HumanMessage("advanced request")]})
+    assert len(result["messages"]) == 3
+    tool_message = result["messages"][-1]
+    assert tool_message.content == "advanced: 1"
+    assert tool_message.name == "advanced_tool"
+
+
+@dataclasses.dataclass
+class Context:
+    user_id: str
+
+
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_with_context(version: str) -> None:
+    """Test dynamic model using config parameters."""
+
+    def dynamic_model(state, runtime: Runtime[Context]):
+        # Use context to determine model behavior
+        user_id = runtime.context.user_id
+        if user_id == "user_premium":
+            return FakeToolCallingModel(tool_calls=[])
+        else:
+            return FakeToolCallingModel(tool_calls=[])
+
+    agent = create_react_agent(
+        dynamic_model, [], context_schema=Context, version=version
+    )
+
+    # Test with basic user
+    result = agent.invoke(
+        {"messages": [HumanMessage("hello")]},
+        context=Context(user_id="user_basic"),
+    )
+    assert len(result["messages"]) == 2
+
+    # Test with premium user
+    result = agent.invoke(
+        {"messages": [HumanMessage("hello")]},
+        context=Context(user_id="user_premium"),
+    )
+    assert len(result["messages"]) == 2
+
+
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_with_state_schema(version: Literal["v1", "v2"]) -> None:
+    """Test dynamic model with custom state schema."""
+
+    class CustomDynamicState(AgentState):
+        model_preference: str = "default"
+
+    def dynamic_model(state: CustomDynamicState, runtime: Runtime) -> BaseChatModel:
+        # Use custom state field to determine model
+        if state.get("model_preference") == "advanced":
+            return FakeToolCallingModel(tool_calls=[])
+        else:
+            return FakeToolCallingModel(tool_calls=[])
+
+    agent = create_react_agent(
+        dynamic_model, [], state_schema=CustomDynamicState, version=version
+    )
+
+    result = agent.invoke(
+        {"messages": [HumanMessage("hello")], "model_preference": "advanced"}
+    )
+    assert len(result["messages"]) == 2
+    assert result["model_preference"] == "advanced"
+
+
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_with_prompt(version: Literal["v1", "v2"]) -> None:
+    """Test dynamic model with different prompt types."""
+
+    def dynamic_model(state: AgentState, runtime: Runtime) -> BaseChatModel:
+        return FakeToolCallingModel(tool_calls=[])
+
+    # Test with string prompt
+    agent = create_react_agent(dynamic_model, [], prompt="system_msg", version=version)
+    result = agent.invoke({"messages": [HumanMessage("human_msg")]})
+    assert result["messages"][-1].content == "system_msg-human_msg"
+
+    # Test with callable prompt
+    def dynamic_prompt(state: AgentState) -> list[MessageLikeRepresentation]:
+        """Generate a dynamic system message based on state."""
+        return [{"role": "system", "content": "system_msg"}] + list(state["messages"])
+
+    agent = create_react_agent(
+        dynamic_model, [], prompt=dynamic_prompt, version=version
+    )
+    result = agent.invoke({"messages": [HumanMessage("human_msg")]})
+    assert result["messages"][-1].content == "system_msg-human_msg"
+
+
+async def test_dynamic_model_async() -> None:
+    """Test dynamic model with async operations."""
+
+    def dynamic_model(state: AgentState, runtime: Runtime) -> BaseChatModel:
+        return FakeToolCallingModel(tool_calls=[])
+
+    agent = create_react_agent(dynamic_model, [])
+
+    result = await agent.ainvoke({"messages": [HumanMessage("hello async")]})
+    assert len(result["messages"]) == 2
+    assert result["messages"][-1].content == "hello async"
+
+
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_with_structured_response(version: str) -> None:
+    """Test dynamic model with structured response format."""
+
+    class TestResponse(BaseModel):
+        message: str
+        confidence: float
+
+    def dynamic_model(state, runtime: Runtime):
+        expected_response = TestResponse(message="dynamic response", confidence=0.9)
+        return FakeToolCallingModel(
+            tool_calls=[], structured_response=expected_response
+        )
+
+    agent = create_react_agent(
+        dynamic_model, [], response_format=TestResponse, version=version
+    )
+
+    result = agent.invoke({"messages": [HumanMessage("hello")]})
+    assert "structured_response" in result
+    assert result["structured_response"].message == "dynamic response"
+    assert result["structured_response"].confidence == 0.9
+
+
+def test_dynamic_model_with_checkpointer(sync_checkpointer):
+    """Test dynamic model with checkpointer."""
+    call_count = 0
+
+    def dynamic_model(state: AgentState, runtime: Runtime) -> BaseChatModel:
+        nonlocal call_count
+        call_count += 1
+        return FakeToolCallingModel(
+            tool_calls=[],
+            # Incrementing the call count as it is used to assign an id
+            # to the AIMessage.
+            # The default reducer semantics are to overwrite an existing message
+            # with the new one if the id matches.
+            index=call_count,
+        )
+
+    agent = create_react_agent(dynamic_model, [], checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "test_dynamic"}}
+
+    # First call
+    result1 = agent.invoke({"messages": [HumanMessage("hello")]}, config)
+    assert len(result1["messages"]) == 2  # Human + AI message
+
+    # Second call - should load from checkpoint
+    result2 = agent.invoke({"messages": [HumanMessage("world")]}, config)
+    assert len(result2["messages"]) == 4
+
+    # Dynamic model should be called each time
+    assert call_count >= 2
+
+
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_state_dependent_tools(version: Literal["v1", "v2"]) -> None:
+    """Test dynamic model that changes available tools based on state."""
+
+    @dec_tool
+    def tool_a(x: int) -> str:
+        """Tool A."""
+        return f"A: {x}"
+
+    @dec_tool
+    def tool_b(x: int) -> str:
+        """Tool B."""
+        return f"B: {x}"
+
+    def dynamic_model(state, runtime: Runtime):
+        # Switch tools based on message history
+        if any("use_b" in msg.content for msg in state["messages"]):
+            return FakeToolCallingModel(
+                tool_calls=[[{"args": {"x": 2}, "id": "1", "name": "tool_b"}], []]
+            )
+        else:
+            return FakeToolCallingModel(
+                tool_calls=[[{"args": {"x": 1}, "id": "1", "name": "tool_a"}], []]
+            )
+
+    agent = create_react_agent(dynamic_model, [tool_a, tool_b], version=version)
+
+    # Ask to use tool B
+    result = agent.invoke({"messages": [HumanMessage("use_b please")]})
+    last_message = result["messages"][-1]
+    assert isinstance(last_message, ToolMessage)
+    assert last_message.content == "B: 2"
+
+    # Ask to use tool A
+    result = agent.invoke({"messages": [HumanMessage("hello")]})
+    last_message = result["messages"][-1]
+    assert isinstance(last_message, ToolMessage)
+    assert last_message.content == "A: 1"
+
+
+@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
+def test_dynamic_model_error_handling(version: Literal["v1", "v2"]) -> None:
+    """Test error handling in dynamic model."""
+
+    def failing_dynamic_model(state, runtime: Runtime):
+        if "fail" in state["messages"][-1].content:
+            raise ValueError("Dynamic model failed")
+        return FakeToolCallingModel(tool_calls=[])
+
+    agent = create_react_agent(failing_dynamic_model, [], version=version)
+
+    # Normal operation should work
+    result = agent.invoke({"messages": [HumanMessage("hello")]})
+    assert len(result["messages"]) == 2
+
+    # Should propagate the error
+    with pytest.raises(ValueError, match="Dynamic model failed"):
+        agent.invoke({"messages": [HumanMessage("fail now")]})
+
+
+def test_dynamic_model_vs_static_model_behavior():
+    """Test that dynamic and static models produce equivalent results when configured the same."""
+    # Static model
+    static_model = FakeToolCallingModel(tool_calls=[])
+    static_agent = create_react_agent(static_model, [])
+
+    # Dynamic model returning the same model
+    def dynamic_model(state, runtime: Runtime):
+        return FakeToolCallingModel(tool_calls=[])
+
+    dynamic_agent = create_react_agent(dynamic_model, [])
+
+    input_msg = {"messages": [HumanMessage("test message")]}
+
+    static_result = static_agent.invoke(input_msg)
+    dynamic_result = dynamic_agent.invoke(input_msg)
+
+    # Results should be equivalent (content-wise, IDs may differ)
+    assert len(static_result["messages"]) == len(dynamic_result["messages"])
+    assert static_result["messages"][0].content == dynamic_result["messages"][0].content
+    assert static_result["messages"][1].content == dynamic_result["messages"][1].content
+
+
+def test_dynamic_model_receives_correct_state():
+    """Test that the dynamic model function receives the correct state, not the model input."""
+    received_states = []
+
+    class CustomAgentState(AgentState):
+        custom_field: str
+
+    def dynamic_model(state, runtime: Runtime) -> BaseChatModel:
+        # Capture the state that's passed to the dynamic model function
+        received_states.append(state)
+        return FakeToolCallingModel(tool_calls=[])
+
+    agent = create_react_agent(dynamic_model, [], state_schema=CustomAgentState)
+
+    # Test with initial state
+    input_state = {"messages": [HumanMessage("hello")], "custom_field": "test_value"}
+    agent.invoke(input_state)
+
+    # The dynamic model function should receive the original state, not the processed model input
+    assert len(received_states) == 1
+    received_state = received_states[0]
+
+    # Should have the custom field from original state
+    assert "custom_field" in received_state
+    assert received_state["custom_field"] == "test_value"
+
+    # Should have the original messages
+    assert len(received_state["messages"]) == 1
+    assert received_state["messages"][0].content == "hello"
+
+
+async def test_dynamic_model_receives_correct_state_async():
+    """Test that the async dynamic model function receives the correct state, not the model input."""
+    received_states = []
+
+    class CustomAgentStateAsync(AgentState):
+        custom_field: str
+
+    def dynamic_model(state, runtime: Runtime):
+        # Capture the state that's passed to the dynamic model function
+        received_states.append(state)
+        return FakeToolCallingModel(tool_calls=[])
+
+    agent = create_react_agent(dynamic_model, [], state_schema=CustomAgentStateAsync)
+
+    # Test with initial state
+    input_state = {
+        "messages": [HumanMessage("hello async")],
+        "custom_field": "test_value_async",
+    }
+    await agent.ainvoke(input_state)
+
+    # The dynamic model function should receive the original state, not the processed model input
+    assert len(received_states) == 1
+    received_state = received_states[0]
+
+    # Should have the custom field from original state
+    assert "custom_field" in received_state
+    assert received_state["custom_field"] == "test_value_async"
+
+    # Should have the original messages
+    assert len(received_state["messages"]) == 1
+    assert received_state["messages"][0].content == "hello async"
+
+
 def test_pre_model_hook() -> None:
     model = FakeToolCallingModel(tool_calls=[])
 

libs/prebuilt/uv.lock

@@ -316,7 +316,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "0.6.0a1"
+version = "0.6.0"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -430,7 +430,7 @@ dev = [
 
 [[package]]
 name = "langgraph-checkpoint-sqlite"
-version = "2.0.10"
+version = "2.0.11"
 source = { editable = "../checkpoint-sqlite" }
 dependencies = [
     { name = "aiosqlite" },