DeepSeekProvider#

Connect a provider and configure retries, credentials, and transport. See the guide.

class zett_agent.providers.deepseek.DeepSeekProvider(model, api_key, transport=None, *, base_url=None, temperature=None, response=False, retry=RetryOptions(base_delay=0.25, max_delay=8.0, max_retries=3), send_prompt_cache_key=False, extra_headers=None, extra_query=None, extra_body=None, timeout=TimeoutOptions(connect=5.0, first_byte=60.0, idle=600.0))[源代码]#

Stream DeepSeek through its OpenAI-compatible protocol.

参数:

  • model (str) -- DeepSeek model identifier accepted by the endpoint.

  • api_key (str) -- API credential supplied by the application.

  • transport (httpx.AsyncBaseTransport | None) -- Optional HTTPX transport for custom routing or tests.

  • base_url (str | None) -- API root; None uses the default DeepSeek v1 endpoint.

  • temperature (float | None) -- Optional provider sampling temperature.

  • response (bool) -- Use DeepSeek's Responses-compatible endpoint instead of Chat Completions.

  • retry (RetryOptions) -- Model-owned retry/backoff policy; never restarts an emitted stream.

  • timeout (TimeoutOptions) -- How long the provider may stay silent before the request is given up on: connecting, waiting for the response headers, and between chunks of a response that has started. See TimeoutOptions; a timeout raises ProviderTimeoutError and is not retried.

  • send_prompt_cache_key (bool) -- Declare the run's prompt cache key on every request. Off by default: DeepSeek caches context automatically and its compatibility endpoint does not document the field. Enable it for gateways that route on the key.

  • extra_headers (Mapping[str, str] | None) -- HTTP headers sent on every request after the adapter's own, for gateways that route or bill on a header. A header given here replaces the SDK default of the same name.

  • extra_query (Mapping[str, str] | None) -- URL query parameters sent on every request, such as the api-version an Azure-style endpoint requires.

  • extra_body (Mapping[str, Any] | None) -- JSON fields merged into every request body last, so they add a vendor option the adapter does not model and may override one it does, such as thinking.

备注

Cache-hit counters are normalized into ModelUsage. Reasoning deltas are emitted only when returned by the provider. Returned reasoning is retained in AssistantMessage and replayed as reasoning_content in later requests, which is required by DeepSeek thinking mode. Close with aclose().

示例

Select the model explicitly rather than inferring it from the key:

model = DeepSeekProvider(
    model="deepseek-chat",
    api_key=os.environ["DEEPSEEK_API_KEY"],
)
try:
    client = await create_agent(model)
    print((await client.run("Explain this function")).content)
finally:
    await model.aclose()

备注

Reasoning effort is mapped only for models whose endpoint supports the corresponding request fields. A selected effort does not guarantee that the server will return reasoning deltas.

参见

ReasoningEffort lists provider-neutral levels, and 连接模型提供商 covers shared provider lifecycle rules.

async aclose()#

Close the SDK's owned HTTP connection pool.

返回类型:

None

native_deferred_tools: bool = False#

See zett_agent.model.NativeDeferredToolsModel.native_deferred_tools. A provider opts in by overriding this on its class; the default keeps Responses, Chat Completions, Google, and Ollama on the withhold-and-search path.

response: bool = False#

Whether this adapter uses a Responses-style endpoint instead of its provider's traditional message or chat-completions endpoint. Only the providers read it; the runtime does not.

sync(*, runtime=None)#

Return a context-managed synchronous view; optionally share a runtime.

Omitting runtime creates a new background thread for this view. Passing an existing runtime reuses its loop and leaves its shutdown to the caller. This API is for service entry points, not Extension lifecycle hooks.

示例

Use a model or tool from a plain function:

with read_file.sync() as read:
    result = read({"path": "README.md"})
参数:

runtime (SyncRuntime | None)

返回类型:

SyncObject[Self]