"""DeepSeek chat models."""
from typing import Dict, Optional, Type, Union
import openai
from langchain_core.messages import AIMessageChunk
from langchain_core.outputs import ChatGenerationChunk, ChatResult
from langchain_core.utils import from_env, secret_from_env
from langchain_openai.chat_models.base import BaseChatOpenAI
from pydantic import ConfigDict, Field, SecretStr, model_validator
from typing_extensions import Self
DEFAULT_API_BASE = "https://api.deepseek.com/v1"
[docs]
class ChatDeepSeek(BaseChatOpenAI):
"""DeepSeek chat model integration to access models hosted in DeepSeek's API.
Setup:
Install ``langchain-deepseek`` and set environment variable ``DEEPSEEK_API_KEY``.
.. code-block:: bash
pip install -U langchain-deepseek
export DEEPSEEK_API_KEY="your-api-key"
Key init args — completion params:
model: str
Name of DeepSeek model to use, e.g. "deepseek-chat".
temperature: float
Sampling temperature.
max_tokens: Optional[int]
Max number of tokens to generate.
Key init args — client params:
timeout: Optional[float]
Timeout for requests.
max_retries: int
Max number of retries.
api_key: Optional[str]
DeepSeek API key. If not passed in will be read from env var DEEPSEEK_API_KEY.
See full list of supported init args and their descriptions in the params section.
Instantiate:
.. code-block:: python
from langchain_deepseek import ChatDeepSeek
llm = ChatDeepSeek(
model="...",
temperature=0,
max_tokens=None,
timeout=None,
max_retries=2,
# api_key="...",
# other params...
)
Invoke:
.. code-block:: python
messages = [
("system", "You are a helpful translator. Translate the user sentence to French."),
("human", "I love programming."),
]
llm.invoke(messages)
Stream:
.. code-block:: python
for chunk in llm.stream(messages):
print(chunk.text(), end="")
.. code-block:: python
stream = llm.stream(messages)
full = next(stream)
for chunk in stream:
full += chunk
full
Async:
.. code-block:: python
await llm.ainvoke(messages)
# stream:
# async for chunk in (await llm.astream(messages))
# batch:
# await llm.abatch([messages])
Tool calling:
.. code-block:: python
from pydantic import BaseModel, Field
class GetWeather(BaseModel):
'''Get the current weather in a given location'''
location: str = Field(..., description="The city and state, e.g. San Francisco, CA")
class GetPopulation(BaseModel):
'''Get the current population in a given location'''
location: str = Field(..., description="The city and state, e.g. San Francisco, CA")
llm_with_tools = llm.bind_tools([GetWeather, GetPopulation])
ai_msg = llm_with_tools.invoke("Which city is hotter today and which is bigger: LA or NY?")
ai_msg.tool_calls
See ``ChatDeepSeek.bind_tools()`` method for more.
Structured output:
.. code-block:: python
from typing import Optional
from pydantic import BaseModel, Field
class Joke(BaseModel):
'''Joke to tell user.'''
setup: str = Field(description="The setup of the joke")
punchline: str = Field(description="The punchline to the joke")
rating: Optional[int] = Field(description="How funny the joke is, from 1 to 10")
structured_llm = llm.with_structured_output(Joke)
structured_llm.invoke("Tell me a joke about cats")
See ``ChatDeepSeek.with_structured_output()`` for more.
Token usage:
.. code-block:: python
ai_msg = llm.invoke(messages)
ai_msg.usage_metadata
.. code-block:: python
{'input_tokens': 28, 'output_tokens': 5, 'total_tokens': 33}
Response metadata
.. code-block:: python
ai_msg = llm.invoke(messages)
ai_msg.response_metadata
""" # noqa: E501
model_name: str = Field(alias="model")
"""The name of the model"""
api_key: Optional[SecretStr] = Field(
default_factory=secret_from_env("DEEPSEEK_API_KEY", default=None)
)
"""DeepSeek API key"""
api_base: str = Field(
default_factory=from_env("DEEPSEEK_API_BASE", default=DEFAULT_API_BASE)
)
"""DeepSeek API base URL"""
model_config = ConfigDict(populate_by_name=True)
@property
def _llm_type(self) -> str:
"""Return type of chat model."""
return "chat-deepseek"
@property
def lc_secrets(self) -> Dict[str, str]:
"""A map of constructor argument names to secret ids."""
return {"api_key": "DEEPSEEK_API_KEY"}
@model_validator(mode="after")
def validate_environment(self) -> Self:
if self.api_base == DEFAULT_API_BASE and not (
self.api_key and self.api_key.get_secret_value()
):
raise ValueError("If using default api base, DEEPSEEK_API_KEY must be set.")
client_params: dict = {
k: v
for k, v in {
"api_key": self.api_key.get_secret_value() if self.api_key else None,
"base_url": self.api_base,
"timeout": self.request_timeout,
"max_retries": self.max_retries,
"default_headers": self.default_headers,
"default_query": self.default_query,
}.items()
if v is not None
}
if not (self.client or None):
sync_specific: dict = {"http_client": self.http_client}
self.client = openai.OpenAI(
**client_params, **sync_specific
).chat.completions
if not (self.async_client or None):
async_specific: dict = {"http_client": self.http_async_client}
self.async_client = openai.AsyncOpenAI(
**client_params, **async_specific
).chat.completions
return self
def _create_chat_result(
self,
response: Union[dict, openai.BaseModel],
generation_info: Optional[Dict] = None,
) -> ChatResult:
rtn = super()._create_chat_result(response, generation_info)
if not isinstance(response, openai.BaseModel):
return rtn
if hasattr(response.choices[0].message, "reasoning_content"): # type: ignore
rtn.generations[0].message.additional_kwargs["reasoning_content"] = (
response.choices[0].message.reasoning_content # type: ignore
)
return rtn
def _convert_chunk_to_generation_chunk(
self,
chunk: dict,
default_chunk_class: Type,
base_generation_info: Optional[Dict],
) -> Optional[ChatGenerationChunk]:
generation_chunk = super()._convert_chunk_to_generation_chunk(
chunk,
default_chunk_class,
base_generation_info,
)
if (choices := chunk.get("choices")) and generation_chunk:
top = choices[0]
if reasoning_content := top.get("delta", {}).get("reasoning_content"):
if isinstance(generation_chunk.message, AIMessageChunk):
generation_chunk.message.additional_kwargs["reasoning_content"] = (
reasoning_content
)
return generation_chunk
