awrap_model_call() — langchain Function Reference

Source Code

libs/langchain_v1/langchain/agents/middleware/tool_selection.py lines 317–358

    async def awrap_model_call(
        self,
        request: ModelRequest[ContextT],
        handler: Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse[ResponseT]]],
    ) -> ModelResponse[ResponseT] | AIMessage:
        """Filter tools based on LLM selection before invoking the model via handler.

        Args:
            request: Model request to execute (includes state and runtime).
            handler: Async callback that executes the model request and returns
                `ModelResponse`.

        Returns:
            The model call result.

        Raises:
            AssertionError: If the selection model response is not a dict.
        """
        selection_request = self._prepare_selection_request(request)
        if selection_request is None:
            return await handler(request)

        # Create dynamic response model with Literal enum of available tool names
        type_adapter = _create_tool_selection_response(selection_request.available_tools)
        schema = type_adapter.json_schema()
        structured_model = selection_request.model.with_structured_output(schema)

        response = await structured_model.ainvoke(
            [
                {"role": "system", "content": selection_request.system_message},
                selection_request.last_user_message,
            ]
        )

        # Response should be a dict since we're passing a schema (not a Pydantic model class)
        if not isinstance(response, dict):
            msg = f"Expected dict response, got {type(response)}"
            raise AssertionError(msg)  # noqa: TRY004
        modified_request = self._process_selection_response(
            response, selection_request.available_tools, selection_request.valid_tool_names, request
        )
        return await handler(modified_request)