_get_request_payload() — langchain Function Reference

Source Code

libs/partners/anthropic/langchain_anthropic/chat_models.py lines 1056–1245

    def _get_request_payload(
        self,
        input_: LanguageModelInput,
        *,
        stop: list[str] | None = None,
        **kwargs: dict,
    ) -> dict:
        """Get the request payload for the Anthropic API."""
        messages = self._convert_input(input_).to_messages()

        for idx, message in enumerate(messages):
            # Translate v1 content
            if (
                isinstance(message, AIMessage)
                and message.response_metadata.get("output_version") == "v1"
            ):
                tcs: list[types.ToolCall] = [
                    {
                        "type": "tool_call",
                        "name": tool_call["name"],
                        "args": tool_call["args"],
                        "id": tool_call.get("id"),
                    }
                    for tool_call in message.tool_calls
                ]
                messages[idx] = message.model_copy(
                    update={
                        "content": _convert_from_v1_to_anthropic(
                            cast(list[types.ContentBlock], message.content),
                            tcs,
                            message.response_metadata.get("model_provider"),
                        )
                    }
                )

        system, formatted_messages = _format_messages(messages)

        # If cache_control is provided in kwargs, add it to the last eligible message
        # block (Anthropic requires cache_control to be nested within a message block).
        # Skip blocks related to code_execution as they cannot have cache_control.
        cache_control = kwargs.pop("cache_control", None)
        if cache_control and formatted_messages:
            # Collect tool IDs called by code_execution
            code_execution_tool_ids = _collect_code_execution_tool_ids(
                formatted_messages
            )

            cache_applied = False
            for formatted_message in reversed(formatted_messages):
                if cache_applied:
                    break
                content = formatted_message.get("content")
                if isinstance(content, list) and content:
                    # Find last eligible block (not code_execution related)
                    for block in reversed(content):
                        if isinstance(block, dict):
                            if _is_code_execution_related_block(
                                block, code_execution_tool_ids
                            ):
                                continue
                            block["cache_control"] = cache_control
                            cache_applied = True
                            break
                elif isinstance(content, str):
                    formatted_message["content"] = [
                        {
                            "type": "text",
                            "text": content,
                            "cache_control": cache_control,
                        }
                    ]
                    cache_applied = True
            # If we didn't find an eligible block we silently drop the control.
            # Anthropic would reject a payload with cache_control on
            # code_execution blocks.
        payload = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "messages": formatted_messages,
            "temperature": self.temperature,
            "top_k": self.top_k,