dima/langchain
1
0 Fork 0
mirror of https://github.com/langchain-ai/langchain.git synced 2026-10-05 09:25:14 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): guard malformed Anthropic content blocks (#38670)

Browse Source
This commit is contained in:
Willow Lopez authored and GitHub committed 2026-08-11 18:50:39 -04:00
1 parent 119bf69a1e
commit e34cd76346
2 files changed
+123 -24
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

libs/core/langchain_core/messages/block_translators/anthropic.py
+58 -24
View File
@@ -26,6 +26,10 @@ def _populate_extras(
return standard_block
def _non_standard_block(block: dict[str, Any]) -> types.NonStandardContentBlock:
return {"type": "non_standard", "value": block}
def _convert_to_v1_from_anthropic_input(
content: list[types.ContentBlock],
) -> list[types.ContentBlock]:
@@ -54,87 +58,107 @@ def _convert_to_v1_from_anthropic_input(
]
for block in blocks:
block_type = block.get("type")
source = block.get("source")
if (
block_type == "document"
and "source" in block
and "type" in block["source"]
and isinstance(source, dict)
and "type" in source
):
if block["source"]["type"] == "base64":
if source["type"] == "base64":
if "data" not in source or "media_type" not in source:
yield _non_standard_block(block)
continue
file_block: types.FileContentBlock = {
"type": "file",
"base64": block["source"]["data"],
"mime_type": block["source"]["media_type"],
"base64": source["data"],
"mime_type": source["media_type"],
}
_populate_extras(file_block, block, {"type", "source"})
yield file_block
elif block["source"]["type"] == "url":
elif source["type"] == "url":
if "url" not in source:
yield _non_standard_block(block)
continue
file_block = {
"type": "file",
"url": block["source"]["url"],
"url": source["url"],
}
_populate_extras(file_block, block, {"type", "source"})
yield file_block
elif block["source"]["type"] == "file":
elif source["type"] == "file":
if "file_id" not in source:
yield _non_standard_block(block)
continue
file_block = {
"type": "file",
"id": block["source"]["file_id"],
"id": source["file_id"],
}
_populate_extras(file_block, block, {"type", "source"})
yield file_block
elif block["source"]["type"] == "text":
elif source["type"] == "text":
if "data" not in source:
yield _non_standard_block(block)
continue
plain_text_block: types.PlainTextContentBlock = {
"type": "text-plain",
"text": block["source"]["data"],
"text": 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}
yield _non_standard_block(block)
elif (
block_type == "image"
and "source" in block
and "type" in block["source"]
block_type == "image" and isinstance(source, dict) and "type" in source
):
if block["source"]["type"] == "base64":
if source["type"] == "base64":
if "data" not in source or "media_type" not in source:
yield _non_standard_block(block)
continue
image_block: types.ImageContentBlock = {
"type": "image",
"base64": block["source"]["data"],
"mime_type": block["source"]["media_type"],
"base64": source["data"],
"mime_type": source["media_type"],
}
_populate_extras(image_block, block, {"type", "source"})
yield image_block
elif block["source"]["type"] == "url":
elif source["type"] == "url":
if "url" not in source:
yield _non_standard_block(block)
continue
image_block = {
"type": "image",
"url": block["source"]["url"],
"url": source["url"],
}
_populate_extras(image_block, block, {"type", "source"})
yield image_block
elif block["source"]["type"] == "file":
elif source["type"] == "file":
if "file_id" not in source:
yield _non_standard_block(block)
continue
image_block = {
"type": "image",
"id": block["source"]["file_id"],
"id": source["file_id"],
}
_populate_extras(image_block, block, {"type", "source"})
yield image_block
else:
yield {"type": "non_standard", "value": block}
yield _non_standard_block(block)
elif block_type in types.KNOWN_BLOCK_TYPES:
yield cast("types.ContentBlock", block)
else:
yield {"type": "non_standard", "value": block}
yield _non_standard_block(block)
return list(_iter_blocks())
@@ -143,6 +167,11 @@ def _convert_citation_to_v1(citation: dict[str, Any]) -> types.Annotation:
citation_type = citation.get("type")
if citation_type == "web_search_result_location":
if "cited_text" not in citation or "url" not in citation:
return {
"type": "non_standard_annotation",
"value": citation,
}
url_citation: types.Citation = {
"type": "citation",
"cited_text": citation["cited_text"],
@@ -165,6 +194,11 @@ def _convert_citation_to_v1(citation: dict[str, Any]) -> types.Annotation:
"page_location",
"search_result_location",
}:
if "cited_text" not in citation:
return {
"type": "non_standard_annotation",
"value": citation,
}
document_citation: types.Citation = {
"type": "citation",
"cited_text": citation["cited_text"],
libs/core/tests/unit_tests/messages/block_translators/test_anthropic.py
+65
View File
@@ -1,3 +1,5 @@
from typing import Any
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage
from langchain_core.messages import content as types
@@ -507,3 +509,66 @@ def test_convert_to_v1_from_anthropic_input() -> None:
]
assert message.content_blocks == expected
def test_convert_to_v1_from_anthropic_input_malformed_sources() -> None:
content: list[str | dict[Any, Any]] = [
{"type": "document", "source": {"type": "base64", "media_type": "app/pdf"}},
{"type": "document", "source": {"type": "url"}},
{"type": "document", "source": {"type": "file"}},
{"type": "document", "source": {"type": "text"}},
{"type": "image", "source": {"type": "base64", "media_type": "image/jpeg"}},
{"type": "image", "source": {"type": "url"}},
{"type": "image", "source": {"type": "file"}},
]
message = HumanMessage(content)
assert message.content_blocks == [
*[{"type": "non_standard", "value": block} for block in content[:4]],
*content[4:],
]
def test_convert_to_v1_from_anthropic_malformed_citations() -> None:
message = AIMessage(
[
{
"type": "text",
"text": "Source-backed answer.",
"citations": [
{
"type": "web_search_result_location",
"cited_text": "Source text",
},
{
"type": "search_result_location",
"title": "Document Title",
},
],
},
],
response_metadata={"model_provider": "anthropic"},
)
assert message.content_blocks == [
{
"type": "text",
"text": "Source-backed answer.",
"annotations": [
{
"type": "non_standard_annotation",
"value": {
"type": "web_search_result_location",
"cited_text": "Source text",
},
},
{
"type": "non_standard_annotation",
"value": {
"type": "search_result_location",
"title": "Document Title",
},
},
],
},
]