mirror of
https://github.com/hwchase17/langchain.git
synced 2026-07-14 13:47:41 +00:00
317 lines
12 KiB
Python
317 lines
12 KiB
Python
"""Test chat model integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Literal
|
|
from unittest.mock import MagicMock
|
|
|
|
from langchain_core.messages import AIMessageChunk, ToolMessage
|
|
from langchain_tests.unit_tests import ChatModelUnitTests
|
|
from openai import BaseModel
|
|
from openai.types.chat import ChatCompletionMessage
|
|
from pydantic import BaseModel as PydanticBaseModel
|
|
from pydantic import Field, SecretStr
|
|
|
|
from langchain_deepseek.chat_models import DEFAULT_API_BASE, ChatDeepSeek
|
|
|
|
MODEL_NAME = "deepseek-chat"
|
|
|
|
|
|
class MockOpenAIResponse(BaseModel):
|
|
"""Mock OpenAI response model."""
|
|
|
|
choices: list
|
|
error: None = None
|
|
|
|
def model_dump( # type: ignore[override]
|
|
self,
|
|
*,
|
|
mode: Literal["json", "python"] | str = "python", # noqa: PYI051
|
|
include: Any = None,
|
|
exclude: Any = None,
|
|
by_alias: bool = False,
|
|
exclude_unset: bool = False,
|
|
exclude_defaults: bool = False,
|
|
exclude_none: bool = False,
|
|
round_trip: bool = False,
|
|
warnings: Literal["none", "warn", "error"] | bool = True,
|
|
context: dict[str, Any] | None = None,
|
|
serialize_as_any: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Convert to dictionary, ensuring `reasoning_content` is included."""
|
|
choices_list = []
|
|
for choice in self.choices:
|
|
if isinstance(choice.message, ChatCompletionMessage):
|
|
message_dict = choice.message.model_dump()
|
|
# Ensure model_extra fields are at top level
|
|
if "model_extra" in message_dict:
|
|
message_dict.update(message_dict["model_extra"])
|
|
else:
|
|
message_dict = {
|
|
"role": "assistant",
|
|
"content": choice.message.content,
|
|
}
|
|
# Add reasoning_content if present
|
|
if hasattr(choice.message, "reasoning_content"):
|
|
message_dict["reasoning_content"] = choice.message.reasoning_content
|
|
# Add model_extra fields at the top level if present
|
|
if hasattr(choice.message, "model_extra"):
|
|
message_dict.update(choice.message.model_extra)
|
|
message_dict["model_extra"] = choice.message.model_extra
|
|
choices_list.append({"message": message_dict})
|
|
|
|
return {"choices": choices_list, "error": self.error}
|
|
|
|
|
|
class TestChatDeepSeekUnit(ChatModelUnitTests):
|
|
"""Standard unit tests for `ChatDeepSeek` chat model."""
|
|
|
|
@property
|
|
def chat_model_class(self) -> type[ChatDeepSeek]:
|
|
"""Chat model class being tested."""
|
|
return ChatDeepSeek
|
|
|
|
@property
|
|
def init_from_env_params(self) -> tuple[dict, dict, dict]:
|
|
"""Parameters to initialize from environment variables."""
|
|
return (
|
|
{
|
|
"DEEPSEEK_API_KEY": "api_key",
|
|
"DEEPSEEK_API_BASE": "api_base",
|
|
},
|
|
{
|
|
"model": MODEL_NAME,
|
|
},
|
|
{
|
|
"api_key": "api_key",
|
|
"api_base": "api_base",
|
|
},
|
|
)
|
|
|
|
@property
|
|
def chat_model_params(self) -> dict:
|
|
"""Parameters to create chat model instance for testing."""
|
|
return {
|
|
"model": MODEL_NAME,
|
|
"api_key": "api_key",
|
|
}
|
|
|
|
def get_chat_model(self) -> ChatDeepSeek:
|
|
"""Get a chat model instance for testing."""
|
|
return ChatDeepSeek(**self.chat_model_params)
|
|
|
|
|
|
class TestChatDeepSeekCustomUnit:
|
|
"""Custom tests specific to DeepSeek chat model."""
|
|
|
|
def test_create_chat_result_with_reasoning_content(self) -> None:
|
|
"""Test that reasoning_content is properly extracted from response."""
|
|
chat_model = ChatDeepSeek(model=MODEL_NAME, api_key=SecretStr("api_key"))
|
|
mock_message = MagicMock()
|
|
mock_message.content = "Main content"
|
|
mock_message.reasoning_content = "This is the reasoning content"
|
|
mock_message.role = "assistant"
|
|
mock_response = MockOpenAIResponse(
|
|
choices=[MagicMock(message=mock_message)],
|
|
error=None,
|
|
)
|
|
|
|
result = chat_model._create_chat_result(mock_response)
|
|
assert (
|
|
result.generations[0].message.additional_kwargs.get("reasoning_content")
|
|
== "This is the reasoning content"
|
|
)
|
|
|
|
def test_create_chat_result_with_model_extra_reasoning(self) -> None:
|
|
"""Test that reasoning is properly extracted from `model_extra`."""
|
|
chat_model = ChatDeepSeek(model=MODEL_NAME, api_key=SecretStr("api_key"))
|
|
mock_message = MagicMock(spec=ChatCompletionMessage)
|
|
mock_message.content = "Main content"
|
|
mock_message.role = "assistant"
|
|
mock_message.model_extra = {"reasoning": "This is the reasoning"}
|
|
mock_message.model_dump.return_value = {
|
|
"role": "assistant",
|
|
"content": "Main content",
|
|
"model_extra": {"reasoning": "This is the reasoning"},
|
|
}
|
|
mock_choice = MagicMock()
|
|
mock_choice.message = mock_message
|
|
mock_response = MockOpenAIResponse(choices=[mock_choice], error=None)
|
|
|
|
result = chat_model._create_chat_result(mock_response)
|
|
assert (
|
|
result.generations[0].message.additional_kwargs.get("reasoning_content")
|
|
== "This is the reasoning"
|
|
)
|
|
|
|
def test_convert_chunk_with_reasoning_content(self) -> None:
|
|
"""Test that reasoning_content is properly extracted from streaming chunk."""
|
|
chat_model = ChatDeepSeek(model=MODEL_NAME, api_key=SecretStr("api_key"))
|
|
chunk: dict[str, Any] = {
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"content": "Main content",
|
|
"reasoning_content": "Streaming reasoning content",
|
|
},
|
|
},
|
|
],
|
|
}
|
|
|
|
chunk_result = chat_model._convert_chunk_to_generation_chunk(
|
|
chunk,
|
|
AIMessageChunk,
|
|
None,
|
|
)
|
|
if chunk_result is None:
|
|
msg = "Expected chunk_result not to be None"
|
|
raise AssertionError(msg)
|
|
assert (
|
|
chunk_result.message.additional_kwargs.get("reasoning_content")
|
|
== "Streaming reasoning content"
|
|
)
|
|
|
|
def test_convert_chunk_with_reasoning(self) -> None:
|
|
"""Test that reasoning is properly extracted from streaming chunk."""
|
|
chat_model = ChatDeepSeek(model=MODEL_NAME, api_key=SecretStr("api_key"))
|
|
chunk: dict[str, Any] = {
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"content": "Main content",
|
|
"reasoning": "Streaming reasoning",
|
|
},
|
|
},
|
|
],
|
|
}
|
|
|
|
chunk_result = chat_model._convert_chunk_to_generation_chunk(
|
|
chunk,
|
|
AIMessageChunk,
|
|
None,
|
|
)
|
|
if chunk_result is None:
|
|
msg = "Expected chunk_result not to be None"
|
|
raise AssertionError(msg)
|
|
assert (
|
|
chunk_result.message.additional_kwargs.get("reasoning_content")
|
|
== "Streaming reasoning"
|
|
)
|
|
|
|
def test_convert_chunk_without_reasoning(self) -> None:
|
|
"""Test that chunk without reasoning fields works correctly."""
|
|
chat_model = ChatDeepSeek(model=MODEL_NAME, api_key=SecretStr("api_key"))
|
|
chunk: dict[str, Any] = {"choices": [{"delta": {"content": "Main content"}}]}
|
|
|
|
chunk_result = chat_model._convert_chunk_to_generation_chunk(
|
|
chunk,
|
|
AIMessageChunk,
|
|
None,
|
|
)
|
|
if chunk_result is None:
|
|
msg = "Expected chunk_result not to be None"
|
|
raise AssertionError(msg)
|
|
assert chunk_result.message.additional_kwargs.get("reasoning_content") is None
|
|
|
|
def test_convert_chunk_with_empty_delta(self) -> None:
|
|
"""Test that chunk with empty delta works correctly."""
|
|
chat_model = ChatDeepSeek(model=MODEL_NAME, api_key=SecretStr("api_key"))
|
|
chunk: dict[str, Any] = {"choices": [{"delta": {}}]}
|
|
|
|
chunk_result = chat_model._convert_chunk_to_generation_chunk(
|
|
chunk,
|
|
AIMessageChunk,
|
|
None,
|
|
)
|
|
if chunk_result is None:
|
|
msg = "Expected chunk_result not to be None"
|
|
raise AssertionError(msg)
|
|
assert chunk_result.message.additional_kwargs.get("reasoning_content") is None
|
|
|
|
def test_get_request_payload(self) -> None:
|
|
"""Test that tool message content is converted from list to string."""
|
|
chat_model = ChatDeepSeek(model=MODEL_NAME, api_key=SecretStr("api_key"))
|
|
|
|
tool_message = ToolMessage(content=[], tool_call_id="test_id")
|
|
payload = chat_model._get_request_payload([tool_message])
|
|
assert payload["messages"][0]["content"] == "[]"
|
|
|
|
tool_message = ToolMessage(content=["item1", "item2"], tool_call_id="test_id")
|
|
payload = chat_model._get_request_payload([tool_message])
|
|
assert payload["messages"][0]["content"] == '["item1", "item2"]'
|
|
|
|
tool_message = ToolMessage(content="test string", tool_call_id="test_id")
|
|
payload = chat_model._get_request_payload([tool_message])
|
|
assert payload["messages"][0]["content"] == "test string"
|
|
|
|
|
|
class SampleTool(PydanticBaseModel):
|
|
"""Sample tool schema for testing."""
|
|
|
|
value: str = Field(description="A test value")
|
|
|
|
|
|
class TestChatDeepSeekStrictMode:
|
|
"""Tests for DeepSeek strict mode support.
|
|
|
|
This tests the experimental beta feature that uses the beta API endpoint
|
|
when `strict=True` is used. These tests can be removed when strict mode
|
|
becomes stable in the default base API.
|
|
"""
|
|
|
|
def test_bind_tools_with_strict_mode_uses_beta_endpoint(self) -> None:
|
|
"""Test that bind_tools with strict=True uses the beta endpoint."""
|
|
llm = ChatDeepSeek(
|
|
model="deepseek-chat",
|
|
api_key=SecretStr("test_key"),
|
|
)
|
|
|
|
# Verify default endpoint
|
|
assert llm.api_base == DEFAULT_API_BASE
|
|
|
|
# Bind tools with strict=True
|
|
bound_model = llm.bind_tools([SampleTool], strict=True)
|
|
|
|
# The bound model should have its internal model using beta endpoint
|
|
# We can't directly access the internal model, but we can verify the behavior
|
|
# by checking that the binding operation succeeds
|
|
assert bound_model is not None
|
|
|
|
def test_bind_tools_without_strict_mode_uses_default_endpoint(self) -> None:
|
|
"""Test bind_tools without strict or with strict=False uses default endpoint."""
|
|
llm = ChatDeepSeek(
|
|
model="deepseek-chat",
|
|
api_key=SecretStr("test_key"),
|
|
)
|
|
|
|
# Test with strict=False
|
|
bound_model_false = llm.bind_tools([SampleTool], strict=False)
|
|
assert bound_model_false is not None
|
|
|
|
# Test with strict=None (default)
|
|
bound_model_none = llm.bind_tools([SampleTool])
|
|
assert bound_model_none is not None
|
|
|
|
def test_with_structured_output_strict_mode_uses_beta_endpoint(self) -> None:
|
|
"""Test that with_structured_output with strict=True uses beta endpoint."""
|
|
llm = ChatDeepSeek(
|
|
model="deepseek-chat",
|
|
api_key=SecretStr("test_key"),
|
|
)
|
|
|
|
# Verify default endpoint
|
|
assert llm.api_base == DEFAULT_API_BASE
|
|
|
|
# Create structured output with strict=True
|
|
structured_model = llm.with_structured_output(SampleTool, strict=True)
|
|
|
|
# The structured model should work with beta endpoint
|
|
assert structured_model is not None
|
|
|
|
|
|
def test_profile() -> None:
|
|
"""Test that model profile is loaded correctly."""
|
|
model = ChatDeepSeek(model="deepseek-reasoner", api_key=SecretStr("test_key"))
|
|
assert model.profile is not None
|
|
assert model.profile["reasoning_output"]
|