github/langchain
1
0
Fork 0
mirror of https://github.com/hwchase17/langchain.git synced 2026-07-15 15:40:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(anthropic): v1 support (#32623)

Browse Source
This commit is contained in:
ccurme
2025-08-22 17:06:53 -03:00
committed by GitHub
parent 5bcf7d006f
commit 26833f2ebc
33 changed files with 2123 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
libs/core/langchain_core/language_models/_utils.py
View File

@@ -212,7 +212,7 @@ def _normalize_messages(
}
"""
from langchain_core.messages.block_translators.langchain import (
from langchain_core.messages.block_translators.langchain_v0 import (
_convert_legacy_v0_content_block_to_v1,
_convert_openai_format_to_data_block,
)

2
libs/core/langchain_core/language_models/chat_models.py
View File

@@ -124,7 +124,7 @@ def _format_for_tracing(messages: list[BaseMessage]) -> list[BaseMessage]:
if (
block.get("type") == "image"
and is_data_content_block(block)
and block.get("source_type") != "id"
and not ("file_id" in block or block.get("source_type") == "id")
):
if message_to_trace is message:
# Shallow copy

10
libs/core/langchain_core/messages/ai.py
View File

@@ -231,7 +231,10 @@ class AIMessage(BaseMessage):
translator = get_translator(model_provider)
if translator:
return translator["translate_content"](self)
try:
return translator["translate_content_chunk"](self)
except NotImplementedError:
pass
# Otherwise, use best-effort parsing
blocks = super().content_blocks
@@ -380,7 +383,10 @@ class AIMessageChunk(AIMessage, BaseMessageChunk):
translator = get_translator(model_provider)
if translator:
return translator["translate_content_chunk"](self)
try:
return translator["translate_content_chunk"](self)
except NotImplementedError:
pass
# Otherwise, use best-effort parsing
blocks = super().content_blocks

43
libs/core/langchain_core/messages/base.py
View File

@@ -7,13 +7,7 @@ from typing import TYPE_CHECKING, Any, Optional, Union, cast, overload
from pydantic import ConfigDict, Field
from langchain_core.load.serializable import Serializable
from langchain_core.messages.block_translators.langchain import (
_convert_legacy_v0_content_block_to_v1,
_convert_v0_multimodal_input_to_v1,
)
from langchain_core.messages.block_translators.openai import (
_convert_to_v1_from_chat_completions_input,
)
from langchain_core.messages import content as types
from langchain_core.utils import get_bolded_text
from langchain_core.utils._merge import merge_dicts, merge_lists
from langchain_core.utils.interactive_env import is_interactive_env
@@ -21,7 +15,6 @@ from langchain_core.utils.interactive_env import is_interactive_env
if TYPE_CHECKING:
from collections.abc import Sequence
from langchain_core.messages import content as types
from langchain_core.prompts.chat import ChatPromptTemplate
@@ -129,6 +122,15 @@ class BaseMessage(Serializable):
"""
from langchain_core.messages import content as types
from langchain_core.messages.block_translators.anthropic import (
_convert_to_v1_from_anthropic_input,
)
from langchain_core.messages.block_translators.langchain_v0 import (
_convert_v0_multimodal_input_to_v1,
)
from langchain_core.messages.block_translators.openai import (
_convert_to_v1_from_chat_completions_input,
)
blocks: list[types.ContentBlock] = []
@@ -143,26 +145,19 @@ class BaseMessage(Serializable):
blocks.append({"type": "text", "text": item})
elif isinstance(item, dict):
item_type = item.get("type")
# Try to convert potential v0 format first
converted_block = _convert_legacy_v0_content_block_to_v1(item)
if converted_block is not item: # Conversion happened
blocks.append(cast("types.ContentBlock", converted_block))
elif item_type is None or item_type not in types.KNOWN_BLOCK_TYPES:
blocks.append(
cast(
"types.ContentBlock",
{"type": "non_standard", "value": item},
)
)
if item_type not in types.KNOWN_BLOCK_TYPES:
blocks.append({"type": "non_standard", "value": item})
else:
blocks.append(cast("types.ContentBlock", item))
# Subsequent passes: attempt to unpack non-standard blocks
blocks = _convert_v0_multimodal_input_to_v1(blocks)
# blocks = _convert_to_v1_from_anthropic_input(blocks)
# ...
return _convert_to_v1_from_chat_completions_input(blocks)
for parsing_step in [
_convert_v0_multimodal_input_to_v1,
_convert_to_v1_from_chat_completions_input,
_convert_to_v1_from_anthropic_input,
]:
blocks = parsing_step(blocks)
return blocks
def text(self) -> str:
"""Get the text content of the message.

66
libs/core/langchain_core/messages/block_translators/__init__.py
View File

@@ -45,37 +45,45 @@ def get_translator(
return PROVIDER_TRANSLATORS.get(provider)
def _auto_register_translators() -> None:
"""Automatically register all available block translators."""
import contextlib
import importlib
import pkgutil
from pathlib import Path
def _register_translators() -> None:
"""Register all translators in langchain-core.
package_path = Path(__file__).parent
A unit test ensures all modules in ``block_translators`` are represented here.
# Discover all sub-modules
for module_info in pkgutil.iter_modules([str(package_path)]):
module_name = module_info.name
For translators implemented outside langchain-core, they can be registered by
calling ``register_translator`` from within the integration package.
"""
from langchain_core.messages.block_translators.anthropic import (
_register_anthropic_translator,
)
from langchain_core.messages.block_translators.bedrock import (
_register_bedrock_translator,
)
from langchain_core.messages.block_translators.bedrock_converse import (
_register_bedrock_converse_translator,
)
from langchain_core.messages.block_translators.google_genai import (
_register_google_genai_translator,
)
from langchain_core.messages.block_translators.google_vertexai import (
_register_google_vertexai_translator,
)
from langchain_core.messages.block_translators.groq import _register_groq_translator
from langchain_core.messages.block_translators.ollama import (
_register_ollama_translator,
)
from langchain_core.messages.block_translators.openai import (
_register_openai_translator,
)
# Skip the __init__ module and any private modules
if module_name.startswith("_"):
continue
if module_info.ispkg:
# For subpackages, discover their submodules
subpackage_path = package_path / module_name
for submodule_info in pkgutil.iter_modules([str(subpackage_path)]):
submodule_name = submodule_info.name
if not submodule_name.startswith("_"):
with contextlib.suppress(ImportError, AttributeError):
importlib.import_module(
f".{module_name}.{submodule_name}", package=__name__
)
else:
# Import top-level translator modules
with contextlib.suppress(ImportError, AttributeError):
importlib.import_module(f".{module_name}", package=__name__)
_register_bedrock_translator()
_register_bedrock_converse_translator()
_register_anthropic_translator()
_register_google_genai_translator()
_register_google_vertexai_translator()
_register_groq_translator()
_register_ollama_translator()
_register_openai_translator()
_auto_register_translators()
_register_translators()

1
libs/core/langchain_core/messages/block_translators/amazon/__init__.py
View File

@@ -1 +0,0 @@
"""Derivations of standard content blocks from Amazon content."""

414
libs/core/langchain_core/messages/block_translators/anthropic.py
View File

@@ -1,17 +1,423 @@
"""Derivations of standard content blocks from Anthropic content."""
import json
from collections.abc import Iterable
from typing import Any, cast
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
def _populate_extras(
standard_block: types.ContentBlock, block: dict[str, Any], known_fields: set[str]
) -> types.ContentBlock:
"""Mutate a block, populating extras."""
if standard_block.get("type") == "non_standard":
return standard_block
for key, value in block.items():
if key not in known_fields:
if "extras" not in block:
# Below type-ignores are because mypy thinks a non-standard block can
# get here, although we exclude them above.
standard_block["extras"] = {} # type: ignore[typeddict-unknown-key]
standard_block["extras"][key] = value # type: ignore[typeddict-item]
return standard_block
def _convert_to_v1_from_anthropic_input(
content: list[types.ContentBlock],
) -> list[types.ContentBlock]:
"""Attempt to unpack non-standard blocks."""
def _iter_blocks() -> Iterable[types.ContentBlock]:
blocks: list[dict[str, Any]] = [
cast("dict[str, Any]", block)
if block.get("type") != "non_standard"
else block["value"] # type: ignore[typeddict-item] # this is only non-standard blocks
for block in content
]
for block in blocks:
block_type = block.get("type")
if (
block_type == "document"
and "source" in block
and "type" in block["source"]
):
if block["source"]["type"] == "base64":
file_block: types.FileContentBlock = {
"type": "file",
"base64": block["source"]["data"],
"mime_type": block["source"]["media_type"],
}
_populate_extras(file_block, block, {"type", "source"})
yield file_block
elif block["source"]["type"] == "url":
file_block = {
"type": "file",
"url": block["source"]["url"],
}
_populate_extras(file_block, block, {"type", "source"})
yield file_block
elif block["source"]["type"] == "file":
file_block = {
"type": "file",
"id": block["source"]["file_id"],
}
_populate_extras(file_block, block, {"type", "source"})
yield file_block
elif block["source"]["type"] == "text":
plain_text_block: types.PlainTextContentBlock = {
"type": "text-plain",
"text": block["source"]["data"],
"mime_type": block.get("media_type", "text/plain"),
}
_populate_extras(plain_text_block, block, {"type", "source"})
yield plain_text_block
else:
yield {"type": "non_standard", "value": block}
elif (
block_type == "image"
and "source" in block
and "type" in block["source"]
):
if block["source"]["type"] == "base64":
image_block: types.ImageContentBlock = {
"type": "image",
"base64": block["source"]["data"],
"mime_type": block["source"]["media_type"],
}
_populate_extras(image_block, block, {"type", "source"})
yield image_block
elif block["source"]["type"] == "url":
image_block = {
"type": "image",
"url": block["source"]["url"],
}
_populate_extras(image_block, block, {"type", "source"})
yield image_block
elif block["source"]["type"] == "file":
image_block = {
"type": "image",
"id": block["source"]["file_id"],
}
_populate_extras(image_block, block, {"type", "source"})
yield image_block
else:
yield {"type": "non_standard", "value": block}
elif block_type in types.KNOWN_BLOCK_TYPES:
yield cast("types.ContentBlock", block)
else:
yield {"type": "non_standard", "value": block}
return list(_iter_blocks())
def _convert_citation_to_v1(citation: dict[str, Any]) -> types.Annotation:
citation_type = citation.get("type")
if citation_type == "web_search_result_location":
url_citation: types.Citation = {
"type": "citation",
"cited_text": citation["cited_text"],
"url": citation["url"],
}
if title := citation.get("title"):
url_citation["title"] = title
known_fields = {"type", "cited_text", "url", "title", "index", "extras"}
for key, value in citation.items():
if key not in known_fields:
if "extras" not in url_citation:
url_citation["extras"] = {}
url_citation["extras"][key] = value
return url_citation
if citation_type in (
"char_location",
"content_block_location",
"page_location",
"search_result_location",
):
document_citation: types.Citation = {
"type": "citation",
"cited_text": citation["cited_text"],
}
if "document_title" in citation:
document_citation["title"] = citation["document_title"]
elif title := citation.get("title"):
document_citation["title"] = title
else:
pass
known_fields = {
"type",
"cited_text",
"document_title",
"title",
"index",
"extras",
}
for key, value in citation.items():
if key not in known_fields:
if "extras" not in document_citation:
document_citation["extras"] = {}
document_citation["extras"][key] = value
return document_citation
return {
"type": "non_standard_annotation",
"value": citation,
}
def _convert_to_v1_from_anthropic(message: AIMessage) -> list[types.ContentBlock]:
"""Convert Anthropic message content to v1 format."""
if isinstance(message.content, str):
message.content = [{"type": "text", "text": message.content}]
def _iter_blocks() -> Iterable[types.ContentBlock]:
for block in message.content:
if not isinstance(block, dict):
continue
block_type = block.get("type")
if block_type == "text":
if citations := block.get("citations"):
text_block: types.TextContentBlock = {
"type": "text",
"text": block.get("text", ""),
"annotations": [_convert_citation_to_v1(a) for a in citations],
}
else:
text_block = {"type": "text", "text": block["text"]}
if "index" in block:
text_block["index"] = block["index"]
yield text_block
elif block_type == "thinking":
reasoning_block: types.ReasoningContentBlock = {
"type": "reasoning",
"reasoning": block.get("thinking", ""),
}
if "index" in block:
reasoning_block["index"] = block["index"]
known_fields = {"type", "thinking", "index", "extras"}
for key in block:
if key not in known_fields:
if "extras" not in reasoning_block:
reasoning_block["extras"] = {}
reasoning_block["extras"][key] = block[key]
yield reasoning_block
elif block_type == "tool_use":
if (
isinstance(message, AIMessageChunk)
and len(message.tool_call_chunks) == 1
):
tool_call_chunk: types.ToolCallChunk = (
message.tool_call_chunks[0].copy() # type: ignore[assignment]
)
if "type" not in tool_call_chunk:
tool_call_chunk["type"] = "tool_call_chunk"
yield tool_call_chunk
elif (
not isinstance(message, AIMessageChunk)
and len(message.tool_calls) == 1
):
tool_call_block = message.tool_calls[0]
if "index" in block:
tool_call_block["index"] = block["index"]
yield tool_call_block
else:
tool_call_block = {
"type": "tool_call",
"name": block.get("name", ""),
"args": block.get("input", {}),
"id": block.get("id", ""),
}
yield tool_call_block
elif (
block_type == "input_json_delta"
and isinstance(message, AIMessageChunk)
and len(message.tool_call_chunks) == 1
):
tool_call_chunk = (
message.tool_call_chunks[0].copy() # type: ignore[assignment]
)
if "type" not in tool_call_chunk:
tool_call_chunk["type"] = "tool_call_chunk"
yield tool_call_chunk
elif block_type == "server_tool_use":
if block.get("name") == "web_search":
web_search_call: types.WebSearchCall = {"type": "web_search_call"}
if query := block.get("input", {}).get("query"):
web_search_call["query"] = query
elif block.get("input") == {} and "partial_json" in block:
try:
input_ = json.loads(block["partial_json"])
if isinstance(input_, dict) and "query" in input_:
web_search_call["query"] = input_["query"]
except json.JSONDecodeError:
pass
if "id" in block:
web_search_call["id"] = block["id"]
if "index" in block:
web_search_call["index"] = block["index"]
known_fields = {"type", "name", "input", "id", "index"}
for key, value in block.items():
if key not in known_fields:
if "extras" not in web_search_call:
web_search_call["extras"] = {}
web_search_call["extras"][key] = value
yield web_search_call
elif block.get("name") == "code_execution":
code_interpreter_call: types.CodeInterpreterCall = {
"type": "code_interpreter_call"
}
if code := block.get("input", {}).get("code"):
code_interpreter_call["code"] = code
elif block.get("input") == {} and "partial_json" in block:
try:
input_ = json.loads(block["partial_json"])
if isinstance(input_, dict) and "code" in input_:
code_interpreter_call["code"] = input_["code"]
except json.JSONDecodeError:
pass
if "id" in block:
code_interpreter_call["id"] = block["id"]
if "index" in block:
code_interpreter_call["index"] = block["index"]
known_fields = {"type", "name", "input", "id", "index"}
for key, value in block.items():
if key not in known_fields:
if "extras" not in code_interpreter_call:
code_interpreter_call["extras"] = {}
code_interpreter_call["extras"][key] = value
yield code_interpreter_call
else:
new_block: types.NonStandardContentBlock = {
"type": "non_standard",
"value": block,
}
if "index" in new_block["value"]:
new_block["index"] = new_block["value"].pop("index")
yield new_block
elif block_type == "web_search_tool_result":
web_search_result: types.WebSearchResult = {"type": "web_search_result"}
if "tool_use_id" in block:
web_search_result["id"] = block["tool_use_id"]
if "index" in block:
web_search_result["index"] = block["index"]
if web_search_result_content := block.get("content", []):
if "extras" not in web_search_result:
web_search_result["extras"] = {}
urls = []
extra_content = []
for result_content in web_search_result_content:
if isinstance(result_content, dict):
if "url" in result_content:
urls.append(result_content["url"])
extra_content.append(result_content)
web_search_result["extras"]["content"] = extra_content
if urls:
web_search_result["urls"] = urls
yield web_search_result
elif block_type == "code_execution_tool_result":
code_interpreter_result: types.CodeInterpreterResult = {
"type": "code_interpreter_result",
"output": [],
}
if "tool_use_id" in block:
code_interpreter_result["id"] = block["tool_use_id"]
if "index" in block:
code_interpreter_result["index"] = block["index"]
code_interpreter_output: types.CodeInterpreterOutput = {
"type": "code_interpreter_output"
}
code_execution_content = block.get("content", {})
if code_execution_content.get("type") == "code_execution_result":
if "return_code" in code_execution_content:
code_interpreter_output["return_code"] = code_execution_content[
"return_code"
]
if "stdout" in code_execution_content:
code_interpreter_output["stdout"] = code_execution_content[
"stdout"
]
if stderr := code_execution_content.get("stderr"):
code_interpreter_output["stderr"] = stderr
if (
output := code_interpreter_output.get("content")
) and isinstance(output, list):
if "extras" not in code_interpreter_result:
code_interpreter_result["extras"] = {}
code_interpreter_result["extras"]["content"] = output
for output_block in output:
if "file_id" in output_block:
if "file_ids" not in code_interpreter_output:
code_interpreter_output["file_ids"] = []
code_interpreter_output["file_ids"].append(
output_block["file_id"]
)
code_interpreter_result["output"].append(code_interpreter_output)
elif (
code_execution_content.get("type")
== "code_execution_tool_result_error"
):
if "extras" not in code_interpreter_result:
code_interpreter_result["extras"] = {}
code_interpreter_result["extras"]["error_code"] = (
code_execution_content.get("error_code")
)
yield code_interpreter_result
else:
new_block = {"type": "non_standard", "value": block}
if "index" in new_block["value"]:
new_block["index"] = new_block["value"].pop("index")
yield new_block
return list(_iter_blocks())
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
"""Derive standard content blocks from a message with Anthropic content."""
raise NotImplementedError
"""Derive standard content blocks from a message with OpenAI content."""
return _convert_to_v1_from_anthropic(message)
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
"""Derive standard content blocks from a message chunk with Anthropic content."""
raise NotImplementedError
"""Derive standard content blocks from a message chunk with OpenAI content."""
return _convert_to_v1_from_anthropic(message)
def _register_anthropic_translator() -> None:

26
libs/core/langchain_core/messages/block_translators/amazon/bedrock.py → libs/core/langchain_core/messages/block_translators/bedrock.py
View File

@@ -1,16 +1,34 @@
"""Derivations of standard content blocks from Amazon (Bedrock) content."""
import warnings
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
WARNED = False
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
def translate_content(message: AIMessage) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message with Bedrock content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Bedrock."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a chunk with Bedrock content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Bedrock."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
@@ -21,9 +39,7 @@ def _register_bedrock_translator() -> None:
"""
from langchain_core.messages.block_translators import register_translator
register_translator(
"amazon_bedrock_chat", translate_content, translate_content_chunk
)
register_translator("bedrock", translate_content, translate_content_chunk)
_register_bedrock_translator()

28
libs/core/langchain_core/messages/block_translators/amazon/bedrock_converse.py → libs/core/langchain_core/messages/block_translators/bedrock_converse.py
View File

@@ -1,16 +1,36 @@
"""Derivations of standard content blocks from Amazon (Bedrock Converse) content."""
import warnings
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
WARNED = False
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
def translate_content(message: AIMessage) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message with Bedrock Converse content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Bedrock "
"Converse."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a chunk with Bedrock Converse content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Bedrock "
"Converse."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
@@ -21,9 +41,7 @@ def _register_bedrock_converse_translator() -> None:
"""
from langchain_core.messages.block_translators import register_translator
register_translator(
"amazon_bedrock_converse_chat", translate_content, translate_content_chunk
)
register_translator("bedrock_converse", translate_content, translate_content_chunk)
_register_bedrock_converse_translator()

27
libs/core/langchain_core/messages/block_translators/chroma.py
View File

@@ -1,27 +0,0 @@
"""Derivations of standard content blocks from Chroma content."""
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
"""Derive standard content blocks from a message with Chroma content."""
raise NotImplementedError
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
"""Derive standard content blocks from a message chunk with Chroma content."""
raise NotImplementedError
def _register_chroma_translator() -> None:
"""Register the Chroma translator with the central registry.
Run automatically when the module is imported.
"""
from langchain_core.messages.block_translators import register_translator
register_translator("chroma", translate_content, translate_content_chunk)
_register_chroma_translator()

1
libs/core/langchain_core/messages/block_translators/google/__init__.py
View File

@@ -1 +0,0 @@
"""Derivations of standard content blocks from Google content."""

22
libs/core/langchain_core/messages/block_translators/google/genai.py → libs/core/langchain_core/messages/block_translators/google_genai.py
View File

@@ -1,16 +1,34 @@
"""Derivations of standard content blocks from Google (GenAI) content."""
import warnings
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
WARNED = False
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
def translate_content(message: AIMessage) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message with Google (GenAI) content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Google GenAI."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a chunk with Google (GenAI) content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Google GenAI."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError

24
libs/core/langchain_core/messages/block_translators/google/vertexai.py → libs/core/langchain_core/messages/block_translators/google_vertexai.py
View File

@@ -1,16 +1,36 @@
"""Derivations of standard content blocks from Google (VertexAI) content."""
import warnings
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
WARNED = False
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
def translate_content(message: AIMessage) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message with Google (VertexAI) content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Google "
"VertexAI."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a chunk with Google (VertexAI) content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Google "
"VertexAI."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError

22
libs/core/langchain_core/messages/block_translators/groq.py
View File

@@ -1,16 +1,34 @@
"""Derivations of standard content blocks from Groq content."""
import warnings
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
WARNED = False
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
def translate_content(message: AIMessage) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message with Groq content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Groq."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message chunk with Groq content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Groq."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError

43
libs/core/langchain_core/messages/block_translators/langchain.py → libs/core/langchain_core/messages/block_translators/langchain_v0.py
View File

@@ -1,4 +1,4 @@
"""Derivations of standard content blocks from LangChain content."""
"""Derivations of standard content blocks from LangChain v0 multimodal content."""
from typing import Any, Union, cast
@@ -21,26 +21,20 @@ def _convert_v0_multimodal_input_to_v1(
Updated list with v0 blocks converted to v1 format.
"""
converted_blocks = []
for block in blocks:
if (
isinstance(block, dict)
and block.get("type") == "non_standard"
and "value" in block
and isinstance(block["value"], dict) # type: ignore[typeddict-item]
):
# We know this is a NonStandardContentBlock, so we can safely access value
value = cast("Any", block)["value"]
# Check if this looks like v0 format
if (
value.get("type") in {"image", "audio", "file"}
and "source_type" in value
):
converted_block = _convert_legacy_v0_content_block_to_v1(value)
converted_blocks.append(cast("types.ContentBlock", converted_block))
else:
converted_blocks.append(block)
unpacked_blocks: list[dict[str, Any]] = [
cast("dict[str, Any]", block)
if block.get("type") != "non_standard"
else block["value"] # type: ignore[typeddict-item] # this is only non-standard blocks
for block in blocks
]
for block in unpacked_blocks:
if block.get("type") in {"image", "audio", "file"} and "source_type" in block:
converted_block = _convert_legacy_v0_content_block_to_v1(block)
converted_blocks.append(cast("types.ContentBlock", converted_block))
elif block.get("type") in types.KNOWN_BLOCK_TYPES:
converted_blocks.append(cast("types.ContentBlock", block))
else:
converted_blocks.append(block)
converted_blocks.append({"type": "non_standard", "value": block})
return converted_blocks
@@ -213,7 +207,7 @@ def _convert_openai_format_to_data_block(
return types.create_image_block(
# Even though this is labeled as `url`, it can be base64-encoded
base64=block["image_url"]["url"],
base64=parsed["data"],
mime_type=parsed["mime_type"],
**all_extras,
)
@@ -278,9 +272,7 @@ def _convert_openai_format_to_data_block(
)
# base64-style file block
if (block["type"] == "file") and (
parsed := _parse_data_uri(block["file"]["file_data"])
):
if block["type"] == "file":
known_keys = {"type", "file"}
extras = _extract_extras(block, known_keys)
@@ -291,11 +283,10 @@ def _convert_openai_format_to_data_block(
for key, value in file_extras.items():
all_extras[f"file_{key}"] = value
mime_type = parsed["mime_type"]
filename = block["file"].get("filename")
return types.create_file_block(
base64=block["file"]["file_data"],
mime_type=mime_type,
mime_type="application/pdf",
filename=filename,
**all_extras,
)

22
libs/core/langchain_core/messages/block_translators/ollama.py
View File

@@ -1,16 +1,34 @@
"""Derivations of standard content blocks from Ollama content."""
import warnings
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
WARNED = False
def translate_content(message: AIMessage) -> list[types.ContentBlock]:
def translate_content(message: AIMessage) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message with Ollama content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Ollama."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]:
def translate_content_chunk(message: AIMessageChunk) -> list[types.ContentBlock]: # noqa: ARG001
"""Derive standard content blocks from a message chunk with Ollama content."""
global WARNED # noqa: PLW0603
if not WARNED:
warning_message = (
"Content block standardization is not yet fully supported for Ollama."
)
warnings.warn(warning_message, stacklevel=2)
WARNED = True
raise NotImplementedError

51
libs/core/langchain_core/messages/block_translators/openai.py
View File

@@ -9,7 +9,7 @@ from langchain_core.language_models._utils import (
_is_openai_data_block,
)
from langchain_core.messages import content as types
from langchain_core.messages.block_translators.langchain import (
from langchain_core.messages.block_translators.langchain_v0 import (
_convert_openai_format_to_data_block,
)
@@ -52,34 +52,31 @@ def _convert_to_v1_from_chat_completions_input(
from langchain_core.messages import content as types
converted_blocks = []
for block in blocks:
if (
isinstance(block, dict)
and block.get("type") == "non_standard"
and "value" in block
and isinstance(block["value"], dict) # type: ignore[typeddict-item]
):
# We know this is a NonStandardContentBlock, so we can safely access value
value = cast("Any", block)["value"]
# Check if this looks like OpenAI format
if value.get("type") in {
"image_url",
"input_audio",
"file",
} and _is_openai_data_block(value):
converted_block = _convert_openai_format_to_data_block(value)
# If conversion succeeded, use it; otherwise keep as non_standard
if (
isinstance(converted_block, dict)
and converted_block.get("type") in types.KNOWN_BLOCK_TYPES
):
converted_blocks.append(cast("types.ContentBlock", converted_block))
else:
converted_blocks.append(block)
unpacked_blocks: list[dict[str, Any]] = [
cast("dict[str, Any]", block)
if block.get("type") != "non_standard"
else block["value"] # type: ignore[typeddict-item] # this is only non-standard blocks
for block in blocks
]
for block in unpacked_blocks:
if block.get("type") in {
"image_url",
"input_audio",
"file",
} and _is_openai_data_block(block):
converted_block = _convert_openai_format_to_data_block(block)
# If conversion succeeded, use it; otherwise keep as non_standard
if (
isinstance(converted_block, dict)
and converted_block.get("type") in types.KNOWN_BLOCK_TYPES
):
converted_blocks.append(cast("types.ContentBlock", converted_block))
else:
converted_blocks.append(block)
converted_blocks.append({"type": "non_standard", "value": block})
elif block.get("type") in types.KNOWN_BLOCK_TYPES:
converted_blocks.append(cast("types.ContentBlock", block))
else:
converted_blocks.append(block)
converted_blocks.append({"type": "non_standard", "value": block})
return converted_blocks

8
libs/core/langchain_core/messages/content.py
View File

@@ -503,12 +503,6 @@ class CodeInterpreterOutput(TypedDict):
file_ids: NotRequired[list[str]]
"""List of file IDs generated by the code interpreter."""
index: NotRequired[Union[int, str]]
"""Index of block in aggregate response. Used during streaming."""
extras: NotRequired[dict[str, Any]]
"""Provider-specific metadata."""
class CodeInterpreterResult(TypedDict):
"""Result of a code interpreter tool call."""
@@ -886,7 +880,6 @@ ToolContentBlock = Union[
ToolCall,
ToolCallChunk,
CodeInterpreterCall,
CodeInterpreterOutput,
CodeInterpreterResult,
WebSearchCall,
WebSearchResult,
@@ -918,7 +911,6 @@ KNOWN_BLOCK_TYPES = {
"video",
# Server-side tool calls
"code_interpreter_call",
"code_interpreter_output",
"code_interpreter_result",
"web_search_call",
"web_search_result",

34
libs/core/langchain_core/utils/_merge.py
View File

@@ -116,11 +116,35 @@ def merge_lists(left: Optional[list], *others: Optional[list]) -> Optional[list]
if to_merge:
# TODO: Remove this once merge_dict is updated with special
# handling for 'type'.
new_e = (
{k: v for k, v in e.items() if k != "type"}
if "type" in e
else e
)
if (left_type := merged[to_merge[0]].get("type")) and (
e.get("type") == "non_standard" and "value" in e
):
if left_type != "non_standard":
# standard + non_standard
new_e: dict[str, Any] = {
"extras": {
k: v
for k, v in e["value"].items()
if k != "type"
}
}
else:
# non_standard + non_standard
new_e = {
"value": {
k: v
for k, v in e["value"].items()
if k != "type"
}
}
if "index" in e:
new_e["index"] = e["index"]
else:
new_e = (
{k: v for k, v in e.items() if k != "type"}
if "type" in e
else e
)
merged[to_merge[0]] = merge_dicts(merged[to_merge[0]], new_e)
else:
merged.append(e)

8
libs/core/tests/unit_tests/language_models/chat_models/test_base.py
View File

@@ -621,14 +621,14 @@ def test_extend_support_to_openai_multimodal_formats() -> None:
"type": "input_audio",
"input_audio": {
"format": "wav",
"data": "data:audio/wav;base64,<base64 string>",
"data": "<base64 string>",
},
},
{ # file-base64
"type": "file",
"file": {
"filename": "draconomicon.pdf",
"file_data": "data:application/pdf;base64,<base64 string>",
"file_data": "<base64 string>",
},
},
{ # file-id
@@ -643,12 +643,12 @@ def test_extend_support_to_openai_multimodal_formats() -> None:
{"type": "text", "text": "Hello"}, # TextContentBlock
{ # AudioContentBlock
"type": "audio",
"base64": "data:audio/wav;base64,<base64 string>",
"base64": "<base64 string>",
"mime_type": "audio/wav",
},
{ # FileContentBlock
"type": "file",
"base64": "data:application/pdf;base64,<base64 string>",
"base64": "<base64 string>",
"mime_type": "application/pdf",
"extras": {"filename": "draconomicon.pdf"},
},

439
libs/core/tests/unit_tests/messages/block_translators/test_anthropic.py Normal file
View File

@@ -0,0 +1,439 @@
from typing import Optional
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage
from langchain_core.messages import content as types
def test_convert_to_v1_from_anthropic() -> None:
message = AIMessage(
[
{"type": "thinking", "thinking": "foo", "signature": "foo_signature"},
{"type": "text", "text": "Let's call a tool."},
{
"type": "tool_use",
"id": "abc_123",
"name": "get_weather",
"input": {"location": "San Francisco"},
},
{
"type": "text",
"text": "It's sunny.",
"citations": [
{
"type": "search_result_location",
"cited_text": "The weather is sunny.",
"source": "source_123",
"title": "Document Title",
"search_result_index": 1,
"start_block_index": 0,
"end_block_index": 2,
},
{"bar": "baz"},
],
},
{
"type": "server_tool_use",
"name": "web_search",
"input": {"query": "web search query"},
"id": "srvtoolu_abc123",
},
{
"type": "web_search_tool_result",
"tool_use_id": "srvtoolu_abc123",
"content": [
{
"type": "web_search_result",
"title": "Page Title 1",
"url": "<page url 1>",
"page_age": "January 1, 2025",
"encrypted_content": "<encrypted content 1>",
},
{
"type": "web_search_result",
"title": "Page Title 2",
"url": "<page url 2>",
"page_age": "January 2, 2025",
"encrypted_content": "<encrypted content 2>",
},
],
},
{
"type": "server_tool_use",
"id": "srvtoolu_def456",
"name": "code_execution",
"input": {"code": "import numpy as np..."},
},
{
"type": "code_execution_tool_result",
"tool_use_id": "srvtoolu_def456",
"content": {
"type": "code_execution_result",
"stdout": "Mean: 5.5\nStandard deviation...",
"stderr": "",
"return_code": 0,
},
},
{"type": "something_else", "foo": "bar"},
],
response_metadata={"model_provider": "anthropic"},
)
expected_content: list[types.ContentBlock] = [
{
"type": "reasoning",
"reasoning": "foo",
"extras": {"signature": "foo_signature"},
},
{"type": "text", "text": "Let's call a tool."},
{
"type": "tool_call",
"id": "abc_123",
"name": "get_weather",
"args": {"location": "San Francisco"},
},
{
"type": "text",
"text": "It's sunny.",
"annotations": [
{
"type": "citation",
"title": "Document Title",
"cited_text": "The weather is sunny.",
"extras": {
"source": "source_123",
"search_result_index": 1,
"start_block_index": 0,
"end_block_index": 2,
},
},
{"type": "non_standard_annotation", "value": {"bar": "baz"}},
],
},
{
"type": "web_search_call",
"id": "srvtoolu_abc123",
"query": "web search query",
},
{
"type": "web_search_result",
"id": "srvtoolu_abc123",
"urls": ["<page url 1>", "<page url 2>"],
"extras": {
"content": [
{
"type": "web_search_result",
"title": "Page Title 1",
"url": "<page url 1>",
"page_age": "January 1, 2025",
"encrypted_content": "<encrypted content 1>",
},
{
"type": "web_search_result",
"title": "Page Title 2",
"url": "<page url 2>",
"page_age": "January 2, 2025",
"encrypted_content": "<encrypted content 2>",
},
]
},
},
{
"type": "code_interpreter_call",
"id": "srvtoolu_def456",
"code": "import numpy as np...",
},
{
"type": "code_interpreter_result",
"id": "srvtoolu_def456",
"output": [
{
"type": "code_interpreter_output",
"return_code": 0,
"stdout": "Mean: 5.5\nStandard deviation...",
}
],
},
{
"type": "non_standard",
"value": {"type": "something_else", "foo": "bar"},
},
]
assert message.content_blocks == expected_content
# Check no mutation
assert message.content != expected_content
def test_convert_to_v1_from_anthropic_chunk() -> None:
chunks = [
AIMessageChunk(
content=[{"text": "Looking ", "type": "text", "index": 0}],
response_metadata={"model_provider": "anthropic"},
),
AIMessageChunk(
content=[{"text": "now.", "type": "text", "index": 0}],
response_metadata={"model_provider": "anthropic"},
),
AIMessageChunk(
content=[
{
"type": "tool_use",
"name": "get_weather",
"input": {},
"id": "toolu_abc123",
"index": 1,
}
],
tool_call_chunks=[
{
"type": "tool_call_chunk",
"name": "get_weather",
"args": "",
"id": "toolu_abc123",
"index": 1,
}
],
response_metadata={"model_provider": "anthropic"},
),
AIMessageChunk(
content=[{"type": "input_json_delta", "partial_json": "", "index": 1}],
tool_call_chunks=[
{
"name": None,
"args": "",
"id": None,
"index": 1,
"type": "tool_call_chunk",
}
],
response_metadata={"model_provider": "anthropic"},
),
AIMessageChunk(
content=[
{"type": "input_json_delta", "partial_json": '{"loca', "index": 1}
],
tool_call_chunks=[
{
"name": None,
"args": '{"loca',
"id": None,
"index": 1,
"type": "tool_call_chunk",
}
],
response_metadata={"model_provider": "anthropic"},
),
AIMessageChunk(
content=[
{"type": "input_json_delta", "partial_json": 'tion": "San ', "index": 1}
],
tool_call_chunks=[
{
"name": None,
"args": 'tion": "San ',
"id": None,
"index": 1,
"type": "tool_call_chunk",
}
],
response_metadata={"model_provider": "anthropic"},
),
AIMessageChunk(
content=[
{"type": "input_json_delta", "partial_json": 'Francisco"}', "index": 1}
],
tool_call_chunks=[
{
"name": None,
"args": 'Francisco"}',
"id": None,
"index": 1,
"type": "tool_call_chunk",
}
],
response_metadata={"model_provider": "anthropic"},
),
]
expected_contents: list[types.ContentBlock] = [
{"type": "text", "text": "Looking ", "index": 0},
{"type": "text", "text": "now.", "index": 0},
{
"type": "tool_call_chunk",
"name": "get_weather",
"args": "",
"id": "toolu_abc123",
"index": 1,
},
{"name": None, "args": "", "id": None, "index": 1, "type": "tool_call_chunk"},
{
"name": None,
"args": '{"loca',
"id": None,
"index": 1,
"type": "tool_call_chunk",
},
{
"name": None,
"args": 'tion": "San ',
"id": None,
"index": 1,
"type": "tool_call_chunk",
},
{
"name": None,
"args": 'Francisco"}',
"id": None,
"index": 1,
"type": "tool_call_chunk",
},
]
for chunk, expected in zip(chunks, expected_contents):
assert chunk.content_blocks == [expected]
full: Optional[AIMessageChunk] = None
for chunk in chunks:
full = chunk if full is None else full + chunk
assert isinstance(full, AIMessageChunk)
expected_content = [
{"type": "text", "text": "Looking now.", "index": 0},
{
"type": "tool_use",
"name": "get_weather",
"partial_json": '{"location": "San Francisco"}',
"input": {},
"id": "toolu_abc123",
"index": 1,
},
]
assert full.content == expected_content
expected_content_blocks = [
{"type": "text", "text": "Looking now.", "index": 0},
{
"type": "tool_call_chunk",
"name": "get_weather",
"args": '{"location": "San Francisco"}',
"id": "toolu_abc123",
"index": 1,
},
]
assert full.content_blocks == expected_content_blocks
def test_convert_to_v1_from_anthropic_input() -> None:
message = HumanMessage(
[
{"type": "text", "text": "foo"},
{
"type": "document",
"source": {
"type": "base64",
"data": "<base64 data>",
"media_type": "application/pdf",
},
},
{
"type": "document",
"source": {
"type": "url",
"url": "<document url>",
},
},
{
"type": "document",
"source": {
"type": "content",
"content": [
{"type": "text", "text": "The grass is green"},
{"type": "text", "text": "The sky is blue"},
],
},
"citations": {"enabled": True},
},
{
"type": "document",
"source": {
"type": "text",
"data": "<plain text data>",
"media_type": "text/plain",
},
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": "<base64 image data>",
},
},
{
"type": "image",
"source": {
"type": "url",
"url": "<image url>",
},
},
{
"type": "image",
"source": {
"type": "file",
"file_id": "<image file id>",
},
},
{
"type": "document",
"source": {"type": "file", "file_id": "<pdf file id>"},
},
]
)
expected: list[types.ContentBlock] = [
{"type": "text", "text": "foo"},
{
"type": "file",
"base64": "<base64 data>",
"mime_type": "application/pdf",
},
{
"type": "file",
"url": "<document url>",
},
{
"type": "non_standard",
"value": {
"type": "document",
"source": {
"type": "content",
"content": [
{"type": "text", "text": "The grass is green"},
{"type": "text", "text": "The sky is blue"},
],
},
"citations": {"enabled": True},
},
},
{
"type": "text-plain",
"text": "<plain text data>",
"mime_type": "text/plain",
},
{
"type": "image",
"base64": "<base64 image data>",
"mime_type": "image/jpeg",
},
{
"type": "image",
"url": "<image url>",
},
{
"type": "image",
"id": "<image file id>",
},
{
"type": "file",
"id": "<pdf file id>",
},
]
assert message.content_blocks == expected

79
libs/core/tests/unit_tests/messages/block_translators/test_langchain_v0.py Normal file
View File

@@ -0,0 +1,79 @@
from langchain_core.messages import HumanMessage
from langchain_core.messages import content as types
from tests.unit_tests.language_models.chat_models.test_base import (
_content_blocks_equal_ignore_id,
)
def test_convert_to_v1_from_openai_input() -> None:
message = HumanMessage(
content=[
{"type": "text", "text": "Hello"},
{
"type": "image",
"source_type": "url",
"url": "https://example.com/image.png",
},
{
"type": "image",
"source_type": "base64",
"data": "<base64 data>",
"mime_type": "image/png",
},
{
"type": "file",
"source_type": "url",
"url": "<document url>",
},
{
"type": "file",
"source_type": "base64",
"data": "<base64 data>",
"mime_type": "application/pdf",
},
{
"type": "audio",
"source_type": "base64",
"data": "<base64 data>",
"mime_type": "audio/mpeg",
},
{
"type": "file",
"source_type": "id",
"id": "<file id>",
},
]
)
expected: list[types.ContentBlock] = [
{"type": "text", "text": "Hello"},
{
"type": "image",
"url": "https://example.com/image.png",
},
{
"type": "image",
"base64": "<base64 data>",
"mime_type": "image/png",
},
{
"type": "file",
"url": "<document url>",
},
{
"type": "file",
"base64": "<base64 data>",
"mime_type": "application/pdf",
},
{
"type": "audio",
"base64": "<base64 data>",
"mime_type": "audio/mpeg",
},
{
"type": "file",
"file_id": "<file id>",
},
]
assert _content_blocks_equal_ignore_id(message.content_blocks, expected)

92
libs/core/tests/unit_tests/messages/block_translators/test_openai.py
View File

@@ -1,41 +1,12 @@
from typing import Optional
from langchain_core.language_models.fake_chat_models import ParrotFakeChatModel
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage
from langchain_core.messages import content as types
from tests.unit_tests.language_models.chat_models.test_base import (
_content_blocks_equal_ignore_id,
)
def test_v0_to_v1_content_blocks() -> None:
llm = ParrotFakeChatModel()
messages = [
{
"role": "user",
# v0 format
"content": [
{
"type": "image",
"source_type": "url",
"url": "https://example.com/image.png",
}
],
}
]
response = llm.invoke(messages)
assert len(response.content_blocks) == 1
expected_content_blocks = [
{
"type": "image",
"url": "https://example.com/image.png",
}
]
assert _content_blocks_equal_ignore_id(
response.content_blocks, expected_content_blocks
)
def test_convert_to_v1_from_responses() -> None:
message = AIMessage(
[
@@ -261,3 +232,64 @@ def test_convert_to_v1_from_responses_chunk() -> None:
},
]
assert full.content_blocks == expected_content_blocks
def test_convert_to_v1_from_openai_input() -> None:
message = HumanMessage(
content=[
{"type": "text", "text": "Hello"},
{
"type": "image_url",
"image_url": {"url": "https://example.com/image.png"},
},
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,/9j/4AAQSkZJRg..."},
},
{
"type": "input_audio",
"input_audio": {
"format": "wav",
"data": "<base64 string>",
},
},
{
"type": "file",
"file": {
"filename": "draconomicon.pdf",
"file_data": "<base64 string>",
},
},
{
"type": "file",
"file": {"file_id": "<file id>"},
},
]
)
expected: list[types.ContentBlock] = [
{"type": "text", "text": "Hello"},
{
"type": "image",
"url": "https://example.com/image.png",
},
{
"type": "image",
"base64": "/9j/4AAQSkZJRg...",
"mime_type": "image/jpeg",
},
{
"type": "audio",
"base64": "<base64 string>",
"mime_type": "audio/wav",
},
{
"type": "file",
"base64": "<base64 string>",
"mime_type": "application/pdf",
"extras": {"filename": "draconomicon.pdf"},
},
{"type": "file", "file_id": "<file id>"},
]
assert _content_blocks_equal_ignore_id(message.content_blocks, expected)

29
libs/core/tests/unit_tests/messages/block_translators/test_registration.py Normal file
View File

@@ -0,0 +1,29 @@
import pkgutil
from pathlib import Path
import pytest
from langchain_core.messages.block_translators import PROVIDER_TRANSLATORS
def test_all_providers_registered() -> None:
"""Test that all block translators implemented in langchain-core are registered.
If this test fails, it is likely that a block translator is implemented but not
registered on import. Check that the provider is included in
``langchain_core.messages.block_translators.__init__._register_translators``.
"""
package_path = (
Path(__file__).parents[4] / "langchain_core" / "messages" / "block_translators"
)
for module_info in pkgutil.iter_modules([str(package_path)]):
module_name = module_info.name
# Skip the __init__ module, any private modules, and ``langchain_v0``, which is
# only used to parse v0 multimodal inputs.
if module_name.startswith("_") or module_name == "langchain_v0":
continue
if module_name not in PROVIDER_TRANSLATORS:
pytest.fail(f"Block translator not registered: {module_name}")

93
libs/core/tests/unit_tests/messages/test_ai.py
View File

@@ -1,3 +1,7 @@
from typing import Union, cast
import pytest
from langchain_core.load import dumpd, load
from langchain_core.messages import AIMessage, AIMessageChunk
from langchain_core.messages import content as types
@@ -310,3 +314,92 @@ def test_content_blocks() -> None:
}
]
assert message.content == ""
# Non-standard
standard_content_1: list[types.ContentBlock] = [
{"type": "non_standard", "index": 0, "value": {"foo": "bar "}}
]
standard_content_2: list[types.ContentBlock] = [
{"type": "non_standard", "index": 0, "value": {"foo": "baz"}}
]
chunk_1 = AIMessageChunk(
content=cast("Union[str, list[Union[str, dict]]]", standard_content_1)
)
chunk_2 = AIMessageChunk(
content=cast("Union[str, list[Union[str, dict]]]", standard_content_2)
)
merged_chunk = chunk_1 + chunk_2
assert merged_chunk.content == [
{"type": "non_standard", "index": 0, "value": {"foo": "bar baz"}},
]
# Test non-standard + non-standard
chunk_1 = AIMessageChunk(
content=[
{
"type": "non_standard",
"index": 0,
"value": {"type": "non_standard_tool", "foo": "bar"},
}
]
)
chunk_2 = AIMessageChunk(
content=[
{
"type": "non_standard",
"index": 0,
"value": {"type": "input_json_delta", "partial_json": "a"},
}
]
)
chunk_3 = AIMessageChunk(
content=[
{
"type": "non_standard",
"index": 0,
"value": {"type": "input_json_delta", "partial_json": "b"},
}
]
)
merged_chunk = chunk_1 + chunk_2 + chunk_3
assert merged_chunk.content == [
{
"type": "non_standard",
"index": 0,
"value": {"type": "non_standard_tool", "foo": "bar", "partial_json": "ab"},
}
]
# Test standard + non-standard with same index
standard_content_1 = [
{"type": "web_search_call", "id": "ws_123", "query": "web query", "index": 0}
]
standard_content_2 = [{"type": "non_standard", "value": {"foo": "bar"}, "index": 0}]
chunk_1 = AIMessageChunk(
content=cast("Union[str, list[Union[str, dict]]]", standard_content_1)
)
chunk_2 = AIMessageChunk(
content=cast("Union[str, list[Union[str, dict]]]", standard_content_2)
)
merged_chunk = chunk_1 + chunk_2
assert merged_chunk.content == [
{
"type": "web_search_call",
"id": "ws_123",
"query": "web query",
"index": 0,
"extras": {"foo": "bar"},
}
]
def test_provider_warns() -> None:
# Test that major providers warn if content block standardization is not yet
# implemented.
# This test should be removed when all major providers support content block
# standardization.
message = AIMessage("Hello.", response_metadata={"model_provider": "groq"})
with pytest.warns(match="not yet fully supported for Groq"):
content_blocks = message.content_blocks
assert content_blocks == [{"type": "text", "text": "Hello."}]