54 lines
2.5 KiB
Python
54 lines
2.5 KiB
Python
"""Opt-in tiny model probe: never sends project files/history or prints secrets."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(Path(__file__).resolve().parents[1] / "backend" / ".env")
|
|
|
|
|
|
def probe(thinking_disabled=False):
|
|
endpoint = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com").rstrip("/")
|
|
if not endpoint.endswith("/v1") and endpoint.endswith("api.deepseek.com"):
|
|
endpoint += "/v1"
|
|
payload = {"model": os.getenv("DEEPSEEK_MODEL", "deepseek-chat"), "messages": [{"role":"user", "content":"请只回复：连接正常。"}], "stream": True, "max_tokens": 1024}
|
|
if thinking_disabled:
|
|
payload["thinking"] = {"type":"disabled"}
|
|
started = time.monotonic()
|
|
result = {"mode":"thinking_disabled" if thinking_disabled else "default", "content_chars":0, "reasoning_chars":0, "keys":[], "finished":False}
|
|
try:
|
|
with httpx.Client(trust_env=False, timeout=30) as client:
|
|
with client.stream("POST", endpoint + "/chat/completions", headers={"Authorization":"Bearer " + os.getenv("DEEPSEEK_API_KEY", "")}, json=payload) as response:
|
|
result["http_status"] = response.status_code
|
|
if response.status_code != 200:
|
|
return result
|
|
for line in response.iter_lines():
|
|
if time.monotonic() - started > 40:
|
|
result["timeout"] = True
|
|
break
|
|
if not line.startswith("data:"):
|
|
continue
|
|
data = line[5:].strip()
|
|
if data == "[DONE]":
|
|
result["finished"] = True
|
|
break
|
|
item = json.loads(data)
|
|
for choice in item.get("choices", []):
|
|
delta = choice.get("delta", {})
|
|
result["keys"] = sorted(set(result["keys"]) | set(delta))
|
|
result["content_chars"] += len(delta.get("content") or "")
|
|
result["reasoning_chars"] += len(delta.get("reasoning_content") or "")
|
|
if choice.get("finish_reason"):
|
|
result["finish_reason"] = choice["finish_reason"]
|
|
except Exception as exc:
|
|
result["error_type"] = type(exc).__name__
|
|
result["duration_seconds"] = round(time.monotonic() - started, 2)
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(json.dumps(probe("--thinking-disabled" in sys.argv), ensure_ascii=False))
|