codekingpro/portable-devtools
114k
1"""Agent factory for creating agents with middleware support."""2 3from __future__ import annotations4 5import functools6import itertools7from dataclasses import dataclass, field, fields8from typing import (9 TYPE_CHECKING,10 Annotated,11 Any,12 Generic,13 cast,14 get_args,15 get_origin,16 get_type_hints,17)18 19from langchain_core.language_models.chat_models import BaseChatModel20from langchain_core.messages import AIMessage, AnyMessage, SystemMessage, ToolMessage21from langchain_core.tools import BaseTool22from langgraph._internal._runnable import RunnableCallable23from langgraph.constants import END, START24from langgraph.graph.state import StateGraph25from langgraph.prebuilt import ToolCallTransformer26from langgraph.prebuilt.tool_node import ToolNode27from langgraph.types import Command, Send28from langsmith import traceable29from typing_extensions import NotRequired, Required, TypedDict30 31from langchain.agents.middleware.types import (32 AgentMiddleware,33 AgentState,34 ContextT,35 ExtendedModelResponse,36 JumpTo,37 ModelRequest,38 ModelResponse,39 OmitFromSchema,40 ResponseT,41 StateT_co,42 ToolCallRequest,43 _InputAgentState,44 _OutputAgentState,45)46from langchain.agents.structured_output import (47 AutoStrategy,48 MultipleStructuredOutputsError,49 OutputToolBinding,50 ProviderStrategy,51 ProviderStrategyBinding,52 ResponseFormat,53 StructuredOutputError,54 StructuredOutputValidationError,55 ToolStrategy,56)57from langchain.chat_models import init_chat_model58 59 60@dataclass61class _ComposedExtendedModelResponse(Generic[ResponseT]):62 """Internal result from composed ``wrap_model_call`` middleware.63 64 Unlike ``ExtendedModelResponse`` (user-facing, single command), this holds the65 full list of commands accumulated across all middleware layers during66 composition.67 """68 69 model_response: ModelResponse[ResponseT]70 """The underlying model response."""71 72 commands: list[Command[Any]] = field(default_factory=list)73 """Commands accumulated from all middleware layers (inner-first, then outer)."""74 75 76if TYPE_CHECKING:77 from collections.abc import Awaitable, Callable, Sequence78 79 from langchain_core.runnables import Runnable, RunnableConfig80 from langgraph.cache.base import BaseCache81 from langgraph.graph.state import CompiledStateGraph82 from langgraph.runtime import Runtime83 from langgraph.store.base import BaseStore84 from langgraph.types import Checkpointer85 86 from langchain.agents.middleware.types import ToolCallWrapper87 88 _ModelCallHandler = Callable[89 [ModelRequest[ContextT], Callable[[ModelRequest[ContextT]], ModelResponse]],90 ModelResponse | AIMessage | ExtendedModelResponse,91 ]92 93 _ComposedModelCallHandler = Callable[94 [ModelRequest[ContextT], Callable[[ModelRequest[ContextT]], ModelResponse]],95 _ComposedExtendedModelResponse,96 ]97 98 _AsyncModelCallHandler = Callable[99 [ModelRequest[ContextT], Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse]]],100 Awaitable[ModelResponse | AIMessage | ExtendedModelResponse],101 ]102 103 _ComposedAsyncModelCallHandler = Callable[104 [ModelRequest[ContextT], Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse]]],105 Awaitable[_ComposedExtendedModelResponse],106 ]107 108 109STRUCTURED_OUTPUT_ERROR_TEMPLATE = "Error: {error}\n Please fix your mistakes."110 111DYNAMIC_TOOL_ERROR_TEMPLATE = """112Middleware added tools that the agent doesn't know how to execute.113 114Unknown tools: {unknown_tool_names}115Registered tools: {available_tool_names}116 117This happens when middleware modifies `request.tools` in `wrap_model_call` to include118tools that weren't passed to `create_agent()`.119 120How to fix this:121 122Option 1: Register tools at agent creation (recommended for most cases)123 Pass the tools to `create_agent(tools=[...])` or set them on `middleware.tools`.124 This makes tools available for every agent invocation.125 126Option 2: Handle dynamic tools in middleware (for tools created at runtime)127 Implement `wrap_tool_call` to execute tools that are added dynamically:128 129 class MyMiddleware(AgentMiddleware):130 def wrap_tool_call(self, request, handler):131 if request.tool_call["name"] == "dynamic_tool":132 # Execute the dynamic tool yourself or override with tool instance133 return handler(request.override(tool=my_dynamic_tool))134 return handler(request)135""".strip()136 137 138def _scrub_inputs(inputs: dict[str, Any]) -> dict[str, Any]:139 """Remove ``runtime`` and ``handler`` from trace inputs before sending to LangSmith."""140 filtered = inputs.copy()141 filtered.pop("handler", None)142 req = filtered.get("request")143 if isinstance(req, (ModelRequest, ToolCallRequest)):144 filtered["request"] = {145 f.name: getattr(req, f.name) for f in fields(req) if f.name != "runtime"146 }147 return filtered148 149 150FALLBACK_MODELS_WITH_STRUCTURED_OUTPUT = [151 # if model profile data are not available, these models are assumed to support152 # structured output153 "grok",154 "gpt-5",155 "gpt-4.1",156 "gpt-4o",157 "gpt-oss",158 "o3-pro",159 "o3-mini",160]161 162 163def _normalize_to_model_response(164 result: ModelResponse | AIMessage | ExtendedModelResponse,165) -> ModelResponse:166 """Normalize middleware return value to ModelResponse.167 168 At inner composition boundaries, ``ExtendedModelResponse`` is unwrapped to its169 underlying ``ModelResponse`` so that inner middleware always sees ``ModelResponse``170 from the handler.171 """172 if isinstance(result, AIMessage):173 return ModelResponse(result=[result], structured_response=None)174 if isinstance(result, ExtendedModelResponse):175 return result.model_response176 return result177 178 179def _build_commands(180 model_response: ModelResponse,181 middleware_commands: list[Command[Any]] | None = None,182) -> list[Command[Any]]:183 """Build a list of Commands from a model response and middleware commands.184 185 The first Command contains the model response state (messages and optional186 structured_response). Middleware commands are appended as-is.187 188 Args:189 model_response: The model response containing messages and optional190 structured output.191 middleware_commands: Commands accumulated from middleware layers during192 composition (inner-first ordering).193 194 Returns:195 List of ``Command`` objects ready to be returned from a model node.196 """197 state: dict[str, Any] = {"messages": model_response.result}198 199 if model_response.structured_response is not None:200 state["structured_response"] = model_response.structured_response201 202 for cmd in middleware_commands or []:203 if cmd.goto:204 msg = (205 "Command goto is not yet supported in wrap_model_call middleware. "206 "Use the jump_to state field with before_model/after_model hooks instead."207 )208 raise NotImplementedError(msg)209 if cmd.resume:210 msg = "Command resume is not yet supported in wrap_model_call middleware."211 raise NotImplementedError(msg)212 if cmd.graph:213 msg = "Command graph is not yet supported in wrap_model_call middleware."214 raise NotImplementedError(msg)215 216 commands: list[Command[Any]] = [Command(update=state)]217 commands.extend(middleware_commands or [])218 return commands219 220 221def _chain_model_call_handlers(222 handlers: Sequence[_ModelCallHandler[ContextT]],223) -> _ComposedModelCallHandler[ContextT] | None:224 """Compose multiple ``wrap_model_call`` handlers into single middleware stack.225 226 Composes handlers so first in list becomes outermost layer. Each handler receives a227 handler callback to execute inner layers. Commands from each layer are accumulated228 into a list (inner-first, then outer) without merging.229 230 Args:231 handlers: List of handlers.232 233 First handler wraps all others.234 235 Returns:236 Composed handler returning ``_ComposedExtendedModelResponse``,237 or ``None`` if handlers empty.238 """239 if not handlers:240 return None241 242 def _to_composed_result(243 result: ModelResponse | AIMessage | ExtendedModelResponse | _ComposedExtendedModelResponse,244 extra_commands: list[Command[Any]] | None = None,245 ) -> _ComposedExtendedModelResponse:246 """Normalize any handler result to _ComposedExtendedModelResponse."""247 commands: list[Command[Any]] = list(extra_commands or [])248 if isinstance(result, _ComposedExtendedModelResponse):249 commands.extend(result.commands)250 model_response = result.model_response251 elif isinstance(result, ExtendedModelResponse):252 model_response = result.model_response253 if result.command is not None:254 commands.append(result.command)255 else:256 model_response = _normalize_to_model_response(result)257 258 return _ComposedExtendedModelResponse(model_response=model_response, commands=commands)259 260 if len(handlers) == 1:261 single_handler = handlers[0]262 263 def normalized_single(264 request: ModelRequest[ContextT],265 handler: Callable[[ModelRequest[ContextT]], ModelResponse],266 ) -> _ComposedExtendedModelResponse:267 return _to_composed_result(single_handler(request, handler))268 269 return normalized_single270 271 def compose_two(272 outer: _ModelCallHandler[ContextT] | _ComposedModelCallHandler[ContextT],273 inner: _ModelCallHandler[ContextT] | _ComposedModelCallHandler[ContextT],274 ) -> _ComposedModelCallHandler[ContextT]:275 """Compose two handlers where outer wraps inner."""276 277 def composed(278 request: ModelRequest[ContextT],279 handler: Callable[[ModelRequest[ContextT]], ModelResponse],280 ) -> _ComposedExtendedModelResponse:281 # Closure variable to capture inner's commands before normalizing282 accumulated_commands: list[Command[Any]] = []283 284 def inner_handler(req: ModelRequest[ContextT]) -> ModelResponse:285 # Clear on each call for retry safety286 accumulated_commands.clear()287 inner_result = inner(req, handler)288 if isinstance(inner_result, _ComposedExtendedModelResponse):289 accumulated_commands.extend(inner_result.commands)290 return inner_result.model_response291 if isinstance(inner_result, ExtendedModelResponse):292 if inner_result.command is not None:293 accumulated_commands.append(inner_result.command)294 return inner_result.model_response295 return _normalize_to_model_response(inner_result)296 297 outer_result = outer(request, inner_handler)298 return _to_composed_result(299 outer_result,300 extra_commands=accumulated_commands or None,301 )302 303 return composed304 305 # Compose right-to-left: outer(inner(innermost(handler)))306 composed_handler = compose_two(handlers[-2], handlers[-1])307 for h in reversed(handlers[:-2]):308 composed_handler = compose_two(h, composed_handler)309 310 return composed_handler311 312 313def _chain_async_model_call_handlers(314 handlers: Sequence[_AsyncModelCallHandler[ContextT]],315) -> _ComposedAsyncModelCallHandler[ContextT] | None:316 """Compose multiple async ``wrap_model_call`` handlers into single middleware stack.317 318 Commands from each layer are accumulated into a list (inner-first, then outer)319 without merging.320 321 Args:322 handlers: List of async handlers.323 324 First handler wraps all others.325 326 Returns:327 Composed async handler returning ``_ComposedExtendedModelResponse``,328 or ``None`` if handlers empty.329 """330 if not handlers:331 return None332 333 def _to_composed_result(334 result: ModelResponse | AIMessage | ExtendedModelResponse | _ComposedExtendedModelResponse,335 extra_commands: list[Command[Any]] | None = None,336 ) -> _ComposedExtendedModelResponse:337 """Normalize any handler result to _ComposedExtendedModelResponse."""338 commands: list[Command[Any]] = list(extra_commands or [])339 if isinstance(result, _ComposedExtendedModelResponse):340 commands.extend(result.commands)341 model_response = result.model_response342 elif isinstance(result, ExtendedModelResponse):343 model_response = result.model_response344 if result.command is not None:345 commands.append(result.command)346 else:347 model_response = _normalize_to_model_response(result)348 349 return _ComposedExtendedModelResponse(model_response=model_response, commands=commands)350 351 if len(handlers) == 1:352 single_handler = handlers[0]353 354 async def normalized_single(355 request: ModelRequest[ContextT],356 handler: Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse]],357 ) -> _ComposedExtendedModelResponse:358 return _to_composed_result(await single_handler(request, handler))359 360 return normalized_single361 362 def compose_two(363 outer: _AsyncModelCallHandler[ContextT] | _ComposedAsyncModelCallHandler[ContextT],364 inner: _AsyncModelCallHandler[ContextT] | _ComposedAsyncModelCallHandler[ContextT],365 ) -> _ComposedAsyncModelCallHandler[ContextT]:366 """Compose two async handlers where outer wraps inner."""367 368 async def composed(369 request: ModelRequest[ContextT],370 handler: Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse]],371 ) -> _ComposedExtendedModelResponse:372 # Closure variable to capture inner's commands before normalizing373 accumulated_commands: list[Command[Any]] = []374 375 async def inner_handler(req: ModelRequest[ContextT]) -> ModelResponse:376 # Clear on each call for retry safety377 accumulated_commands.clear()378 inner_result = await inner(req, handler)379 if isinstance(inner_result, _ComposedExtendedModelResponse):380 accumulated_commands.extend(inner_result.commands)381 return inner_result.model_response382 if isinstance(inner_result, ExtendedModelResponse):383 if inner_result.command is not None:384 accumulated_commands.append(inner_result.command)385 return inner_result.model_response386 return _normalize_to_model_response(inner_result)387 388 outer_result = await outer(request, inner_handler)389 return _to_composed_result(390 outer_result,391 extra_commands=accumulated_commands or None,392 )393 394 return composed395 396 # Compose right-to-left: outer(inner(innermost(handler)))397 composed_handler = compose_two(handlers[-2], handlers[-1])398 for h in reversed(handlers[:-2]):399 composed_handler = compose_two(h, composed_handler)400 401 return composed_handler402 403 404@functools.lru_cache(maxsize=100)405def _get_schema_type_hints(schema: type) -> dict[str, Any]:406 """Return cached type hints for a schema."""407 return get_type_hints(schema, include_extras=True)408 409 410def _resolve_schemas(schemas: list[type]) -> tuple[type, type, type]:411 """Resolve state, input, and output schemas for the given schemas.412 413 Schemas are merged in list order; later entries override earlier ones when the414 same field is declared by multiple schemas. Duplicates are harmless — a type415 that appears more than once is processed at its last position.416 """417 schema_hints = {schema: _get_schema_type_hints(schema) for schema in schemas}418 return (419 _resolve_schema(schema_hints, "StateSchema", None),420 _resolve_schema(schema_hints, "InputSchema", "input"),421 _resolve_schema(schema_hints, "OutputSchema", "output"),422 )423 424 425def _resolve_schema(426 schema_hints: dict[type, dict[str, Any]],427 schema_name: str,428 omit_flag: str | None = None,429) -> type:430 """Resolve schema by merging schemas and optionally respecting `OmitFromSchema` annotations.431 432 Args:433 schema_hints: Resolved schema annotations to merge434 schema_name: Name for the generated `TypedDict`435 omit_flag: If specified, omit fields with this flag set (`'input'` or436 `'output'`)437 438 Returns:439 Merged schema as `TypedDict`440 """441 all_annotations = {}442 443 for hints in schema_hints.values():444 for field_name, field_type in hints.items():445 should_omit = False446 447 if omit_flag:448 metadata = _extract_metadata(field_type)449 for meta in metadata:450 if isinstance(meta, OmitFromSchema) and getattr(meta, omit_flag) is True:451 should_omit = True452 break453 454 if not should_omit:455 all_annotations[field_name] = field_type456 457 return TypedDict(schema_name, all_annotations) # type: ignore[operator]458 459 460def _extract_metadata(type_: type) -> list[Any]:461 """Extract metadata from a field type, handling `Required`/`NotRequired` and `Annotated` wrappers.""" # noqa: E501462 # Handle Required[Annotated[...]] or NotRequired[Annotated[...]]463 if get_origin(type_) in {Required, NotRequired}:464 inner_type = get_args(type_)[0]465 if get_origin(inner_type) is Annotated:466 return list(get_args(inner_type)[1:])467 468 # Handle direct Annotated[...]469 elif get_origin(type_) is Annotated:470 return list(get_args(type_)[1:])471 472 return []473 474 475def _get_can_jump_to(middleware: AgentMiddleware[Any, Any], hook_name: str) -> list[JumpTo]:476 """Get the `can_jump_to` list from either sync or async hook methods.477 478 Args:479 middleware: The middleware instance to inspect.480 hook_name: The name of the hook (`'before_model'` or `'after_model'`).481 482 Returns:483 List of jump destinations, or empty list if not configured.484 """485 # Get the base class method for comparison486 base_sync_method = getattr(AgentMiddleware, hook_name, None)487 base_async_method = getattr(AgentMiddleware, f"a{hook_name}", None)488 489 # Try sync method first - only if it's overridden from base class490 sync_method = getattr(middleware.__class__, hook_name, None)491 if (492 sync_method493 and sync_method is not base_sync_method494 and hasattr(sync_method, "__can_jump_to__")495 ):496 return sync_method.__can_jump_to__497 498 # Try async method - only if it's overridden from base class499 async_method = getattr(middleware.__class__, f"a{hook_name}", None)500 if (501 async_method502 and async_method is not base_async_method503 and hasattr(async_method, "__can_jump_to__")504 ):505 return async_method.__can_jump_to__506 507 return []508 509 510def _supports_provider_strategy(511 model: str | BaseChatModel, tools: list[BaseTool | dict[str, Any]] | None = None512) -> bool:513 """Check if a model supports provider-specific structured output.514 515 Args:516 model: Model name string or `BaseChatModel` instance.517 tools: Optional list of tools provided to the agent.518 519 Needed because some models don't support structured output together with tool calling.520 521 Returns:522 `True` if the model supports provider-specific structured output, `False` otherwise.523 """524 model_name: str | None = None525 if isinstance(model, str):526 model_name = model527 elif isinstance(model, BaseChatModel):528 model_name = (529 getattr(model, "model_name", None)530 or getattr(model, "model", None)531 or getattr(model, "model_id", "")532 )533 model_profile = model.profile534 if (535 model_profile is not None536 and model_profile.get("structured_output")537 # We make an exception for Gemini < 3-series models, which currently do not support538 # simultaneous tool use with structured output; 3-series can.539 and not (540 tools541 and isinstance(model_name, str)542 and "gemini" in model_name.lower()543 and "gemini-3" not in model_name.lower()544 )545 ):546 return True547 548 return (549 any(part in model_name.lower() for part in FALLBACK_MODELS_WITH_STRUCTURED_OUTPUT)550 if model_name551 else False552 )553 554 555def _handle_structured_output_error(556 exception: Exception,557 response_format: ResponseFormat[Any],558) -> tuple[bool, str]:559 """Handle structured output error.560 561 Returns `(should_retry, retry_tool_message)`.562 """563 if not isinstance(response_format, ToolStrategy):564 return False, ""565 566 handle_errors = response_format.handle_errors567 568 if handle_errors is False:569 return False, ""570 if handle_errors is True:571 return True, STRUCTURED_OUTPUT_ERROR_TEMPLATE.format(error=str(exception))572 if isinstance(handle_errors, str):573 return True, handle_errors574 if isinstance(handle_errors, type):575 if issubclass(handle_errors, Exception) and isinstance(exception, handle_errors):576 return True, STRUCTURED_OUTPUT_ERROR_TEMPLATE.format(error=str(exception))577 return False, ""578 if isinstance(handle_errors, tuple):579 if any(isinstance(exception, exc_type) for exc_type in handle_errors):580 return True, STRUCTURED_OUTPUT_ERROR_TEMPLATE.format(error=str(exception))581 return False, ""582 return True, handle_errors(exception)583 584 585def _chain_tool_call_wrappers(586 wrappers: Sequence[ToolCallWrapper],587) -> ToolCallWrapper | None:588 """Compose wrappers into middleware stack (first = outermost).589 590 Args:591 wrappers: Wrappers in middleware order.592 593 Returns:594 Composed wrapper, or `None` if empty.595 596 Example:597 ```python598 wrapper = _chain_tool_call_wrappers([auth, cache, retry])599 # Request flows: auth -> cache -> retry -> tool600 # Response flows: tool -> retry -> cache -> auth601 ```602 """603 if not wrappers:604 return None605 606 if len(wrappers) == 1:607 return wrappers[0]608 609 def compose_two(outer: ToolCallWrapper, inner: ToolCallWrapper) -> ToolCallWrapper:610 """Compose two wrappers where outer wraps inner."""611 612 def composed(613 request: ToolCallRequest,614 execute: Callable[[ToolCallRequest], ToolMessage | Command[Any]],615 ) -> ToolMessage | Command[Any]:616 # Create a callable that invokes inner with the original execute617 def call_inner(req: ToolCallRequest) -> ToolMessage | Command[Any]:618 return inner(req, execute)619 620 # Outer can call call_inner multiple times621 return outer(request, call_inner)622 623 return composed624 625 # Chain all wrappers: first -> second -> ... -> last626 result = wrappers[-1]627 for wrapper in reversed(wrappers[:-1]):628 result = compose_two(wrapper, result)629 630 return result631 632 633def _chain_async_tool_call_wrappers(634 wrappers: Sequence[635 Callable[636 [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],637 Awaitable[ToolMessage | Command[Any]],638 ]639 ],640) -> (641 Callable[642 [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],643 Awaitable[ToolMessage | Command[Any]],644 ]645 | None646):647 """Compose async wrappers into middleware stack (first = outermost).648 649 Args:650 wrappers: Async wrappers in middleware order.651 652 Returns:653 Composed async wrapper, or `None` if empty.654 """655 if not wrappers:656 return None657 658 if len(wrappers) == 1:659 return wrappers[0]660 661 def compose_two(662 outer: Callable[663 [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],664 Awaitable[ToolMessage | Command[Any]],665 ],666 inner: Callable[667 [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],668 Awaitable[ToolMessage | Command[Any]],669 ],670 ) -> Callable[671 [ToolCallRequest, Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]],672 Awaitable[ToolMessage | Command[Any]],673 ]:674 """Compose two async wrappers where outer wraps inner."""675 676 async def composed(677 request: ToolCallRequest,678 execute: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],679 ) -> ToolMessage | Command[Any]:680 # Create an async callable that invokes inner with the original execute681 async def call_inner(req: ToolCallRequest) -> ToolMessage | Command[Any]:682 return await inner(req, execute)683 684 # Outer can call call_inner multiple times685 return await outer(request, call_inner)686 687 return composed688 689 # Chain all wrappers: first -> second -> ... -> last690 result = wrappers[-1]691 for wrapper in reversed(wrappers[:-1]):692 result = compose_two(wrapper, result)693 694 return result695 696 697def create_agent(698 model: str | BaseChatModel,699 tools: Sequence[BaseTool | Callable[..., Any] | dict[str, Any]] | None = None,700 *,701 system_prompt: str | SystemMessage | None = None,702 middleware: Sequence[AgentMiddleware[StateT_co, ContextT]] = (),703 response_format: ResponseFormat[ResponseT] | type[ResponseT] | dict[str, Any] | None = None,704 state_schema: type[AgentState[ResponseT]] | None = None,705 context_schema: type[ContextT] | None = None,706 checkpointer: Checkpointer | None = None,707 store: BaseStore | None = None,708 interrupt_before: list[str] | None = None,709 interrupt_after: list[str] | None = None,710 debug: bool = False,711 name: str | None = None,712 cache: BaseCache[Any] | None = None,713 transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,714) -> CompiledStateGraph[715 AgentState[ResponseT], ContextT, _InputAgentState, _OutputAgentState[ResponseT]716]:717 """Creates an agent graph that calls tools in a loop until a stopping condition is met.718 719 For more details on using `create_agent`,720 visit the [Agents](https://docs.langchain.com/oss/python/langchain/agents) docs.721 722 Args:723 model: The language model for the agent.724 725 Can be a string identifier (e.g., `"openai:gpt-4"`) or a direct chat model726 instance (e.g., [`ChatOpenAI`][langchain_openai.ChatOpenAI] or other another727 [LangChain chat model](https://docs.langchain.com/oss/python/integrations/chat)).728 729 For a full list of supported model strings, see730 [`init_chat_model`][langchain.chat_models.init_chat_model(model_provider)].731 732 !!! tip ""733 734 See the [Models](https://docs.langchain.com/oss/python/langchain/models)735 docs for more information.736 tools: A list of tools, `dict`, or `Callable`.737 738 If `None` or an empty list, the agent will consist of a model node without a739 tool calling loop.740 741 742 !!! tip ""743 744 See the [Tools](https://docs.langchain.com/oss/python/langchain/tools)745 docs for more information.746 system_prompt: An optional system prompt for the LLM.747 748 Can be a `str` (which will be converted to a `SystemMessage`) or a749 `SystemMessage` instance directly. The system message is added to the750 beginning of the message list when calling the model.751 middleware: A sequence of middleware instances to apply to the agent.752 753 Middleware can intercept and modify agent behavior at various stages.754 755 !!! tip ""756 757 See the [Middleware](https://docs.langchain.com/oss/python/langchain/middleware)758 docs for more information.759 response_format: An optional configuration for structured responses.760 761 Can be a `ToolStrategy`, `ProviderStrategy`, or a Pydantic model class.762 763 If provided, the agent will handle structured output during the764 conversation flow.765 766 Raw schemas will be wrapped in an appropriate strategy based on model767 capabilities.768 769 !!! tip ""770 771 See the [Structured output](https://docs.langchain.com/oss/python/langchain/structured-output)772 docs for more information.773 state_schema: An optional `TypedDict` schema that extends `AgentState`.774 775 When provided, this schema is used instead of `AgentState` as the base776 schema for merging with middleware state schemas. This allows users to777 add custom state fields without needing to create custom middleware.778 779 Generally, it's recommended to use `state_schema` extensions via middleware780 to keep relevant extensions scoped to corresponding hooks / tools.781 context_schema: An optional schema for runtime context.782 checkpointer: An optional checkpoint saver object.783 784 Used for persisting the state of the graph (e.g., as chat memory) for a785 single thread (e.g., a single conversation).786 store: An optional store object.787 788 Used for persisting data across multiple threads (e.g., multiple789 conversations / users).790 interrupt_before: An optional list of node names to interrupt before.791 792 Useful if you want to add a user confirmation or other interrupt793 before taking an action.794 interrupt_after: An optional list of node names to interrupt after.795 796 Useful if you want to return directly or run additional processing797 on an output.798 debug: Whether to enable verbose logging for graph execution.799 800 When enabled, prints detailed information about each node execution, state801 updates, and transitions during agent runtime. Useful for debugging802 middleware behavior and understanding agent execution flow.803 name: An optional name for the `CompiledStateGraph`.804 805 This name will be automatically used when adding the agent graph to806 another graph as a subgraph node - particularly useful for building807 multi-agent systems.808 cache: An optional `BaseCache` instance to enable caching of graph execution.809 transformers: Optional sequence of scope-aware `StreamTransformer`810 factories to register on the compiled graph in addition to811 the agent defaults. Each factory is invoked per-scope812 (`factory(scope)`) so subgraph mini-muxes get fresh813 instances. Appended after the built-in `ToolCallTransformer`.814 815 Returns:816 A compiled `StateGraph` that can be used for chat interactions.817 818 Raises:819 AssertionError: If duplicate middleware instances are provided.820 821 The agent node calls the language model with the messages list (after applying822 the system prompt). If the resulting [`AIMessage`][langchain.messages.AIMessage]823 contains `tool_calls`, the graph will then call the tools. The tools node executes824 the tools and adds the responses to the messages list as825 [`ToolMessage`][langchain.messages.ToolMessage] objects. The agent node then calls826 the language model again. The process repeats until no more `tool_calls` are present827 in the response. The agent then returns the full list of messages.828 829 Example:830 ```python831 from langchain.agents import create_agent832 833 834 def check_weather(location: str) -> str:835 '''Return the weather forecast for the specified location.'''836 return f"It's always sunny in {location}"837 838 839 graph = create_agent(840 model="anthropic:claude-sonnet-4-5-20250929",841 tools=[check_weather],842 system_prompt="You are a helpful assistant",843 )844 inputs = {"messages": [{"role": "user", "content": "what is the weather in sf"}]}845 for chunk in graph.stream(inputs, stream_mode="updates"):846 print(chunk)847 ```848 """849 # init chat model850 if isinstance(model, str):851 model = init_chat_model(model)852 853 # Convert system_prompt to SystemMessage if needed854 system_message: SystemMessage | None = None855 if system_prompt is not None:856 if isinstance(system_prompt, SystemMessage):857 system_message = system_prompt858 else:859 system_message = SystemMessage(content=system_prompt)860 861 # Handle tools being None or empty862 if tools is None:863 tools = []864 865 # Convert response format and setup structured output tools866 # Raw schemas are wrapped in AutoStrategy to preserve auto-detection intent.867 # AutoStrategy is converted to ToolStrategy upfront to calculate tools during agent creation,868 # but may be replaced with ProviderStrategy later based on model capabilities.869 initial_response_format: ToolStrategy[Any] | ProviderStrategy[Any] | AutoStrategy[Any] | None870 if response_format is None:871 initial_response_format = None872 elif isinstance(response_format, (ToolStrategy, ProviderStrategy)):873 # Preserve explicitly requested strategies874 initial_response_format = response_format875 elif isinstance(response_format, AutoStrategy):876 # AutoStrategy provided - preserve it for later auto-detection877 initial_response_format = response_format878 else:879 # Raw schema - wrap in AutoStrategy to enable auto-detection880 initial_response_format = AutoStrategy(schema=response_format)881 882 # For AutoStrategy, convert to ToolStrategy to setup tools upfront883 # (may be replaced with ProviderStrategy later based on model)884 tool_strategy_for_setup: ToolStrategy[Any] | None = None885 if isinstance(initial_response_format, AutoStrategy):886 tool_strategy_for_setup = ToolStrategy(schema=initial_response_format.schema)887 elif isinstance(initial_response_format, ToolStrategy):888 tool_strategy_for_setup = initial_response_format889 890 structured_output_tools: dict[str, OutputToolBinding[Any]] = {}891 if tool_strategy_for_setup:892 for response_schema in tool_strategy_for_setup.schema_specs:893 structured_tool_info = OutputToolBinding.from_schema_spec(response_schema)894 structured_output_tools[structured_tool_info.tool.name] = structured_tool_info895 middleware_tools = [t for m in middleware for t in getattr(m, "tools", [])]896 897 # Collect middleware with wrap_tool_call or awrap_tool_call hooks898 # Include middleware with either implementation to ensure NotImplementedError is raised899 # when middleware doesn't support the execution path900 middleware_w_wrap_tool_call = [901 m902 for m in middleware903 if m.__class__.wrap_tool_call is not AgentMiddleware.wrap_tool_call904 or m.__class__.awrap_tool_call is not AgentMiddleware.awrap_tool_call905 ]906 907 # Chain all wrap_tool_call handlers into a single composed handler908 wrap_tool_call_wrapper = None909 if middleware_w_wrap_tool_call:910 wrappers = [911 traceable(name=f"{m.name}.wrap_tool_call", process_inputs=_scrub_inputs)(912 m.wrap_tool_call913 )914 for m in middleware_w_wrap_tool_call915 ]916 wrap_tool_call_wrapper = _chain_tool_call_wrappers(wrappers)917 918 # Collect middleware with awrap_tool_call or wrap_tool_call hooks919 # Include middleware with either implementation to ensure NotImplementedError is raised920 # when middleware doesn't support the execution path921 middleware_w_awrap_tool_call = [922 m923 for m in middleware924 if m.__class__.awrap_tool_call is not AgentMiddleware.awrap_tool_call925 or m.__class__.wrap_tool_call is not AgentMiddleware.wrap_tool_call926 ]927 928 # Chain all awrap_tool_call handlers into a single composed async handler929 awrap_tool_call_wrapper = None930 if middleware_w_awrap_tool_call:931 async_wrappers = [932 traceable(name=f"{m.name}.awrap_tool_call", process_inputs=_scrub_inputs)(933 m.awrap_tool_call934 )935 for m in middleware_w_awrap_tool_call936 ]937 awrap_tool_call_wrapper = _chain_async_tool_call_wrappers(async_wrappers)938 939 # Setup tools940 tool_node: ToolNode | None = None941 # Extract built-in provider tools (dict format) and regular tools (BaseTool/callables)942 built_in_tools = [t for t in tools if isinstance(t, dict)]943 regular_tools = [t for t in tools if not isinstance(t, dict)]944 945 # Tools that require client-side execution (must be in ToolNode)946 available_tools = middleware_tools + regular_tools947 948 # Create ToolNode if we have client-side tools OR if middleware defines wrap_tool_call949 # (which may handle dynamically registered tools)950 tool_node = (951 ToolNode(952 tools=available_tools,953 wrap_tool_call=wrap_tool_call_wrapper,954 awrap_tool_call=awrap_tool_call_wrapper,955 )956 if available_tools or wrap_tool_call_wrapper or awrap_tool_call_wrapper957 else None958 )959 960 # Default tools for ModelRequest initialization961 # Use converted BaseTool instances from ToolNode (not raw callables)962 # Include built-ins and converted tools (can be changed dynamically by middleware)963 # Structured tools are NOT included - they're added dynamically based on response_format964 if tool_node:965 default_tools = list(tool_node.tools_by_name.values()) + built_in_tools966 else:967 default_tools = list(built_in_tools)968 969 # validate middleware970 if len({m.name for m in middleware}) != len(middleware):971 msg = "Please remove duplicate middleware instances."972 raise AssertionError(msg)973 middleware_w_before_agent = [974 m975 for m in middleware976 if m.__class__.before_agent is not AgentMiddleware.before_agent977 or m.__class__.abefore_agent is not AgentMiddleware.abefore_agent978 ]979 middleware_w_before_model = [980 m981 for m in middleware982 if m.__class__.before_model is not AgentMiddleware.before_model983 or m.__class__.abefore_model is not AgentMiddleware.abefore_model984 ]985 middleware_w_after_model = [986 m987 for m in middleware988 if m.__class__.after_model is not AgentMiddleware.after_model989 or m.__class__.aafter_model is not AgentMiddleware.aafter_model990 ]991 middleware_w_after_agent = [992 m993 for m in middleware994 if m.__class__.after_agent is not AgentMiddleware.after_agent995 or m.__class__.aafter_agent is not AgentMiddleware.aafter_agent996 ]997 # Collect middleware with wrap_model_call or awrap_model_call hooks998 # Include middleware with either implementation to ensure NotImplementedError is raised999 # when middleware doesn't support the execution path1000 middleware_w_wrap_model_call = [1001 m1002 for m in middleware1003 if m.__class__.wrap_model_call is not AgentMiddleware.wrap_model_call1004 or m.__class__.awrap_model_call is not AgentMiddleware.awrap_model_call1005 ]1006 # Collect middleware with awrap_model_call or wrap_model_call hooks1007 # Include middleware with either implementation to ensure NotImplementedError is raised1008 # when middleware doesn't support the execution path1009 middleware_w_awrap_model_call = [1010 m1011 for m in middleware1012 if m.__class__.awrap_model_call is not AgentMiddleware.awrap_model_call1013 or m.__class__.wrap_model_call is not AgentMiddleware.wrap_model_call1014 ]1015 1016 # Compose wrap_model_call handlers into a single middleware stack (sync)1017 wrap_model_call_handler = None1018 if middleware_w_wrap_model_call:1019 sync_handlers = [1020 traceable(name=f"{m.name}.wrap_model_call", process_inputs=_scrub_inputs)(1021 m.wrap_model_call1022 )1023 for m in middleware_w_wrap_model_call1024 ]1025 wrap_model_call_handler = _chain_model_call_handlers(sync_handlers)1026 1027 # Compose awrap_model_call handlers into a single middleware stack (async)1028 awrap_model_call_handler = None1029 if middleware_w_awrap_model_call:1030 async_handlers = [1031 traceable(name=f"{m.name}.awrap_model_call", process_inputs=_scrub_inputs)(1032 m.awrap_model_call1033 )1034 for m in middleware_w_awrap_model_call1035 ]1036 awrap_model_call_handler = _chain_async_model_call_handlers(async_handlers)1037 1038 base_state = state_schema if state_schema is not None else AgentState1039 # Build an ordered list: middleware schemas first (in registration order),1040 # base_state last so it wins any field conflict. This lets the caller's1041 # explicit state_schema override middleware annotations — e.g. passing1042 # a DeltaChannel-annotated schema wins over BinaryOperatorAggregate from1043 # AgentState without requiring a post-compilation patch.1044 state_schemas: list[type] = [*(m.state_schema for m in middleware), base_state]1045 1046 resolved_state_schema, input_schema, output_schema = _resolve_schemas(state_schemas)1047 1048 # create graph, add nodes1049 graph: StateGraph[1050 AgentState[ResponseT], ContextT, _InputAgentState, _OutputAgentState[ResponseT]1051 ] = StateGraph(1052 state_schema=resolved_state_schema,1053 input_schema=input_schema,1054 output_schema=output_schema,1055 context_schema=context_schema,1056 )1057 1058 def _handle_model_output(1059 output: AIMessage, effective_response_format: ResponseFormat[Any] | None1060 ) -> dict[str, Any]:1061 """Handle model output including structured responses.1062 1063 Args:1064 output: The AI message output from the model.1065 effective_response_format: The actual strategy used (may differ from initial1066 if auto-detected).1067 """1068 # Handle structured output with provider strategy1069 if isinstance(effective_response_format, ProviderStrategy):1070 if not output.tool_calls:1071 provider_strategy_binding = ProviderStrategyBinding.from_schema_spec(1072 effective_response_format.schema_spec1073 )1074 try:1075 structured_response = provider_strategy_binding.parse(output)1076 except Exception as exc:1077 schema_name = getattr(1078 effective_response_format.schema_spec.schema, "__name__", "response_format"1079 )1080 validation_error = StructuredOutputValidationError(schema_name, exc, output)1081 raise validation_error from exc1082 else:1083 return {"messages": [output], "structured_response": structured_response}1084 return {"messages": [output]}1085 1086 # Handle structured output with tool strategy1087 if (1088 isinstance(effective_response_format, ToolStrategy)1089 and isinstance(output, AIMessage)1090 and output.tool_calls1091 ):1092 structured_tool_calls = [1093 tc for tc in output.tool_calls if tc["name"] in structured_output_tools1094 ]1095 1096 if structured_tool_calls:1097 exception: StructuredOutputError | None = None1098 if len(structured_tool_calls) > 1:1099 # Handle multiple structured outputs error1100 tool_names = [tc["name"] for tc in structured_tool_calls]1101 exception = MultipleStructuredOutputsError(tool_names, output)1102 should_retry, error_message = _handle_structured_output_error(1103 exception, effective_response_format1104 )1105 if not should_retry:1106 raise exception1107 1108 # Add error messages and retry1109 tool_messages = [1110 ToolMessage(1111 content=error_message,1112 tool_call_id=tc["id"],1113 name=tc["name"],1114 )1115 for tc in structured_tool_calls1116 ]1117 return {"messages": [output, *tool_messages]}1118 1119 # Handle single structured output1120 tool_call = structured_tool_calls[0]1121 try:1122 structured_tool_binding = structured_output_tools[tool_call["name"]]1123 structured_response = structured_tool_binding.parse(tool_call["args"])1124 1125 tool_message_content = (1126 effective_response_format.tool_message_content1127 or f"Returning structured response: {structured_response}"1128 )1129 1130 return {1131 "messages": [1132 output,1133 ToolMessage(1134 content=tool_message_content,1135 tool_call_id=tool_call["id"],1136 name=tool_call["name"],1137 ),1138 ],1139 "structured_response": structured_response,1140 }1141 except Exception as exc:1142 exception = StructuredOutputValidationError(tool_call["name"], exc, output)1143 should_retry, error_message = _handle_structured_output_error(1144 exception, effective_response_format1145 )1146 if not should_retry:1147 raise exception from exc1148 1149 return {1150 "messages": [1151 output,1152 ToolMessage(1153 content=error_message,1154 tool_call_id=tool_call["id"],1155 name=tool_call["name"],1156 ),1157 ],1158 }1159 1160 return {"messages": [output]}1161 1162 def _get_bound_model(1163 request: ModelRequest[ContextT],1164 ) -> tuple[Runnable[Any, Any], ResponseFormat[Any] | None]:1165 """Get the model with appropriate tool bindings.1166 1167 Performs auto-detection of strategy if needed based on model capabilities.1168 1169 Args:1170 request: The model request containing model, tools, and response format.1171 1172 Returns:1173 Tuple of `(bound_model, effective_response_format)` where1174 `effective_response_format` is the actual strategy used (may differ from1175 initial if auto-detected).1176 1177 Raises:1178 ValueError: If middleware returned unknown client-side tool names.1179 ValueError: If `ToolStrategy` specifies tools not declared upfront.1180 """1181 # Validate ONLY client-side tools that need to exist in tool_node1182 # Skip validation when wrap_tool_call is defined, as middleware may handle1183 # dynamic tools that are added at runtime via wrap_model_call1184 has_wrap_tool_call = wrap_tool_call_wrapper or awrap_tool_call_wrapper1185 1186 # Build map of available client-side tools from the ToolNode1187 # (which has already converted callables)1188 available_tools_by_name = {}1189 if tool_node:1190 available_tools_by_name = tool_node.tools_by_name.copy()1191 1192 # Check if any requested tools are unknown CLIENT-SIDE tools1193 # Only validate if wrap_tool_call is NOT defined (no dynamic tool handling)1194 if not has_wrap_tool_call:1195 unknown_tool_names = []1196 for t in request.tools:1197 # Only validate BaseTool instances (skip built-in dict tools)1198 if isinstance(t, dict):1199 continue1200 if isinstance(t, BaseTool) and t.name not in available_tools_by_name: