File size: 6,120 Bytes
d06fa57 8956871 d06fa57 8956871 d06fa57 8956871 d06fa57 8956871 d06fa57 ed58f07 d06fa57 ed58f07 d06fa57 ed58f07 d06fa57 ed58f07 d06fa57 ed58f07 d06fa57 ed58f07 d06fa57 ed58f07 d06fa57 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | # Copyright (c) 2024-present, Yumeow. Licensed under the MIT License.
import httpx
from openai import DefaultHttpxClient, OpenAI
from typing import Any
from .base_api import BaseAPI, ProgressCallback, ProviderSampleResult
from .utils import RemoteModelCatalog
from ..core import APICallSample, ToolCall
@BaseAPI.register("deepseek")
class DeepSeekAPI(BaseAPI):
"""LLM provider adapter for the official DeepSeek API (OpenAI-compatible endpoint)."""
def __init__(self, model: str = "deepseek-chat", **kwargs: Any) -> None:
"""Initialize the DeepSeek API adapter.
Args:
model: DeepSeek model identifier (defaults to ``"deepseek-chat"``).
**kwargs: Additional arguments forwarded to :class:`BaseAPI`.
"""
super().__init__(model=model, **kwargs)
self._model_catalog = RemoteModelCatalog(
lambda url: httpx.Client(proxy=self.proxy_url(url), trust_env=False)
)
def format_response(
self, message: dict[str, Any], tool_calls: list[ToolCall], content: str | None = None
) -> APICallSample:
"""Convert a DeepSeek assistant message into a normalized :class:`~sr_harness.core.APICallSample`.
Args:
message: Raw DeepSeek response message dictionary.
tool_calls: Normalized tool calls extracted from the message.
content: Optional explicit text content override.
Returns:
APICallSample: Normalized assistant response sample.
"""
return self._build_response(message, tool_calls, content)
@property
def model_list(self) -> list[str]:
"""Fetch and return the list of available models from the DeepSeek catalog.
Returns:
list[str]: Model identifiers offered by DeepSeek.
"""
endpoint = self.get_env("DEEPSEEK_ENDPOINT", "https://api.deepseek.com")
return self._model_catalog.get(
f"{endpoint.rstrip('/')}/models",
description="DeepSeek model catalog",
api_key=self.get_env("DEEPSEEK_API_KEY", required=True),
)
def _request_sample(
self,
messages: list[dict[str, Any]],
*,
sample_index: int,
max_tokens=None,
temperature=1,
callback: ProgressCallback | None = None,
) -> ProviderSampleResult:
endpoint = self.get_env("DEEPSEEK_ENDPOINT", "https://api.deepseek.com")
client = OpenAI(
timeout=300.0,
max_retries=0,
api_key=self.get_env("DEEPSEEK_API_KEY", required=True),
base_url=endpoint,
http_client=DefaultHttpxClient(
proxy=self.proxy_url(endpoint),
trust_env=False,
),
)
payload = {
"model": self.model,
"messages": [
{
key: value
for key, value in message.items()
if not (key == "tool_calls" and value == [])
}
for message in messages
],
**({"max_tokens": max_tokens} if max_tokens is not None else {}),
"temperature": temperature,
}
if self.tool_box is None:
pass
elif not self.tool_box.native_tool_calls:
description = (
"Use the following tools when a tool call is needed. "
"Return tool calls in the specified format.\n\n"
f"{self.tool_box.format_tools()}"
)
if payload["messages"][0]["role"] in {"system", "developer"}:
first = payload["messages"][0]
payload["messages"][0] = {
**first,
"content": f"{first['content']}\n\n{description}",
}
else:
payload["messages"].insert(0, {"role": "system", "content": description})
else:
tools = self.tool_box.format_tools()
if not isinstance(tools, list):
raise TypeError("Native tool parser must return JSON tool descriptions")
payload["tools"] = tools
payload["tool_choice"] = "auto"
completion, raw_response, message, content = self.openai_chat_completion(
client,
payload,
sample=sample_index,
callback=callback,
)
token_usage = self.openai_token_usage(completion, other_key="other")
money_usage = {
"prompt": 0.28 / 1e6 * token_usage.get("prompt", 0),
"answer": 0.42 / 1e6 * token_usage.get("answer", 0),
}
if "other" in token_usage:
money_usage["other"] = 0.42 / 1e6 * token_usage["other"]
choices = raw_response.get("choices") or [{}]
finish_reason = raw_response.get("finish_reason") or choices[0].get("finish_reason")
if finish_reason == "length":
partial_message = dict(message)
partial_message.pop("tool_calls", None)
partial_response = self.format_response(partial_message, [], content)
partial_response.finish_reason = "length"
return ProviderSampleResult(
response=partial_response,
token_usage=token_usage,
money_usage=money_usage,
raw_response=raw_response,
error=(
f"DeepSeekAPI({self.model}) exhausted max_tokens={max_tokens if max_tokens is not None else 'provider default'}; "
"increase the Agent's maximum output length and retry."
),
)
tool_calls = self.parse_tool_calls(message, content)
if not (tool_calls or content.strip()):
raise ValueError(
f"DeepSeekAPI({self.model}) returned empty content and no usable tool calls."
)
return ProviderSampleResult(
response=self.format_response(message, tool_calls, content),
token_usage=token_usage,
money_usage=money_usage,
raw_response=raw_response,
)
|