michele
0d817c876f
HubAudio U11/U12/U38 400'd after a thinking-only review turn because the retry dropped reasoning_content. Native client now appends that field and continues; hierarchical sheet ingest stays in 2.60.10.
407 lines
14 KiB
Python
407 lines
14 KiB
Python
"""DeepSeek provider: PDF ingest, OpenAI message translation, routing, pricing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from backend.config import settings
|
|
from backend.services.llm.pdf_ingest import extract_pdf_text, make_text_pdf
|
|
from backend.services.llm.deepseek_provider import (
|
|
_is_vision_model,
|
|
_repair_assistant_messages,
|
|
_to_openai_tool,
|
|
_to_openai_tool_choice,
|
|
completion_from_openai,
|
|
messages_to_openai,
|
|
)
|
|
from backend.services.llm.local_skill import load_skill_markdown, load_skill_validator
|
|
from backend.services.llm.pricing import PRICING, cost_for_entry
|
|
from backend.services.llm.types import (
|
|
Message,
|
|
PdfBlock,
|
|
TextBlock,
|
|
ToolCall,
|
|
ToolResultBlock,
|
|
ToolSchema,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_pdf(tmp_path: Path) -> Path:
|
|
pdf = tmp_path / "ds.pdf"
|
|
pdf.write_bytes(make_text_pdf([
|
|
"Pin configuration\n1 VCC Power\n2 GND Ground\n3 TXD UART transmit",
|
|
"Absolute maximum ratings\nVCC 6.0 V",
|
|
]))
|
|
return pdf
|
|
|
|
|
|
def test_extract_pdf_text_includes_page_markers(sample_pdf: Path):
|
|
text = extract_pdf_text(sample_pdf)
|
|
assert "page 1" in text.lower() or "--- page 1 ---" in text
|
|
assert "VCC" in text
|
|
assert sample_pdf.name in text
|
|
|
|
|
|
def test_vision_model_detection():
|
|
assert _is_vision_model("deepseek-v4-flash-vision-exp")
|
|
assert _is_vision_model("deepseek-flash")
|
|
assert _is_vision_model("deepseek-v4-flash")
|
|
assert not _is_vision_model("deepseek-v4-pro")
|
|
|
|
|
|
def test_messages_to_openai_pdf_becomes_text(sample_pdf: Path):
|
|
messages = [
|
|
Message("user", [
|
|
PdfBlock(path=sample_pdf, cacheable=True),
|
|
TextBlock("Extract the pin table."),
|
|
]),
|
|
]
|
|
out = messages_to_openai(messages, vision=False)
|
|
assert len(out) == 1
|
|
assert out[0]["role"] == "user"
|
|
content = out[0]["content"]
|
|
if isinstance(content, str):
|
|
blob = content
|
|
else:
|
|
blob = " ".join(p.get("text", "") for p in content if p.get("type") == "text")
|
|
assert not any(p.get("type") == "image_url" for p in content)
|
|
assert "VCC" in blob
|
|
assert "Extract the pin table" in blob
|
|
|
|
|
|
def test_thinking_only_assistant_sends_empty_string_content():
|
|
"""DeepSeek accepts content=\"\" and rejects content=null / omitted."""
|
|
out = messages_to_openai(
|
|
[Message("assistant", [
|
|
TextBlock("", reasoning_content="Need to inspect pin 3 first."),
|
|
])],
|
|
vision=False,
|
|
)
|
|
assert out[0]["role"] == "assistant"
|
|
assert out[0]["content"] == ""
|
|
assert "content" in out[0]
|
|
assert out[0]["reasoning_content"] == "Need to inspect pin 3 first."
|
|
assert "tool_calls" not in out[0]
|
|
|
|
|
|
def test_empty_assistant_blocks_still_set_content():
|
|
out = messages_to_openai([Message("assistant", [])], vision=False)
|
|
assert out[0]["content"] == ""
|
|
|
|
|
|
def test_tool_call_assistant_sends_empty_string_content():
|
|
out = messages_to_openai(
|
|
[Message("assistant", [
|
|
ToolCall(id="c1", name="get_pintable", input={"ref": "U2"},
|
|
reasoning_content="look up pins"),
|
|
])],
|
|
vision=False,
|
|
)
|
|
assert out[0]["content"] == ""
|
|
assert out[0]["tool_calls"][0]["function"]["name"] == "get_pintable"
|
|
assert out[0]["reasoning_content"] == "look up pins"
|
|
|
|
|
|
def test_repair_null_content_even_when_tool_calls_present():
|
|
repaired = _repair_assistant_messages([
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "c1"}]},
|
|
{"role": "user", "content": "ok"},
|
|
])
|
|
assert repaired[0]["content"] == ""
|
|
assert repaired[0]["tool_calls"][0]["id"] == "c1"
|
|
|
|
|
|
def test_messages_to_openai_tool_roundtrip():
|
|
messages = [
|
|
Message("assistant", [
|
|
TextBlock("checking", reasoning_content="I should query the net."),
|
|
ToolCall(id="call_1", name="get_net_for_pin", input={"ref": "U2", "pin": "3"}),
|
|
]),
|
|
Message("user", [
|
|
ToolResultBlock(tool_use_id="call_1", name="get_net_for_pin", content="UART_TX"),
|
|
TextBlock("continue"),
|
|
]),
|
|
]
|
|
out = messages_to_openai(messages, vision=False)
|
|
assert out[0]["role"] == "assistant"
|
|
assert out[0]["reasoning_content"] == "I should query the net."
|
|
assert out[0]["tool_calls"][0]["function"]["name"] == "get_net_for_pin"
|
|
args = json.loads(out[0]["tool_calls"][0]["function"]["arguments"])
|
|
assert args["ref"] == "U2"
|
|
assert out[1]["role"] == "tool"
|
|
assert out[1]["tool_call_id"] == "call_1"
|
|
assert out[1]["content"] == "UART_TX"
|
|
assert out[2]["role"] == "user"
|
|
|
|
|
|
def test_forced_tool_choice_disables_thinking():
|
|
captured: dict = {}
|
|
|
|
class FakeCompletions:
|
|
async def create(self, **kwargs):
|
|
captured.update(kwargs)
|
|
fn = SimpleNamespace(name="save_resolved_specs", arguments="{}")
|
|
tc = SimpleNamespace(id="c1", function=fn)
|
|
msg = SimpleNamespace(content=None, reasoning_content=None, tool_calls=[tc])
|
|
usage = SimpleNamespace(
|
|
prompt_tokens=10, completion_tokens=5,
|
|
prompt_cache_hit_tokens=0, prompt_tokens_details=None,
|
|
)
|
|
return SimpleNamespace(
|
|
choices=[SimpleNamespace(message=msg, finish_reason="tool_calls")],
|
|
usage=usage,
|
|
)
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.chat = SimpleNamespace(completions=FakeCompletions())
|
|
|
|
from backend.services.llm.deepseek_provider import DeepSeekSession
|
|
session = DeepSeekSession(
|
|
client=FakeClient(),
|
|
model="deepseek-v4-flash",
|
|
system="sys",
|
|
max_tokens=256,
|
|
thinking=True,
|
|
reasoning_effort="medium",
|
|
)
|
|
import asyncio
|
|
from backend.services.llm.types import Message, TextBlock, ToolSchema
|
|
asyncio.run(session.complete(
|
|
messages=[Message("user", [TextBlock("resolve")])],
|
|
tools=[ToolSchema(name="save_resolved_specs", description="x", input_schema={"type": "object"})],
|
|
tool_choice={"name": "save_resolved_specs"},
|
|
))
|
|
assert captured["extra_body"]["thinking"]["type"] == "disabled"
|
|
assert "reasoning_effort" not in captured["extra_body"]
|
|
assert captured["tool_choice"]["function"]["name"] == "save_resolved_specs"
|
|
|
|
|
|
def test_thinking_only_echoes_reasoning_content_on_follow_up():
|
|
"""Live U11/U12/U38: thinking-only then retry without reasoning_content → 400."""
|
|
calls: list[dict] = []
|
|
|
|
class FakeCompletions:
|
|
async def create(self, **kwargs):
|
|
calls.append({
|
|
"thinking": kwargs["extra_body"]["thinking"]["type"],
|
|
"messages": kwargs["messages"],
|
|
})
|
|
usage = SimpleNamespace(
|
|
prompt_tokens=10, completion_tokens=5,
|
|
prompt_cache_hit_tokens=0, prompt_tokens_details=None,
|
|
)
|
|
if kwargs["extra_body"]["thinking"]["type"] == "enabled" and len(calls) == 1:
|
|
msg = SimpleNamespace(
|
|
content=None,
|
|
reasoning_content="pondering the schematic",
|
|
tool_calls=None,
|
|
model_extra=None,
|
|
)
|
|
return SimpleNamespace(
|
|
choices=[SimpleNamespace(message=msg, finish_reason="stop")],
|
|
usage=usage,
|
|
)
|
|
fn = SimpleNamespace(name="get_pintable", arguments='{"ref":"U11"}')
|
|
tc = SimpleNamespace(id="c1", function=fn)
|
|
msg = SimpleNamespace(
|
|
content=None, reasoning_content=None, tool_calls=[tc], model_extra=None,
|
|
)
|
|
return SimpleNamespace(
|
|
choices=[SimpleNamespace(message=msg, finish_reason="tool_calls")],
|
|
usage=usage,
|
|
)
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.chat = SimpleNamespace(completions=FakeCompletions())
|
|
|
|
from backend.services.llm.deepseek_provider import DeepSeekSession
|
|
from backend.services.llm.types import Message, TextBlock, ToolSchema
|
|
import asyncio
|
|
|
|
session = DeepSeekSession(
|
|
client=FakeClient(),
|
|
model="deepseek-flash",
|
|
system="sys",
|
|
max_tokens=256,
|
|
thinking=True,
|
|
reasoning_effort="medium",
|
|
)
|
|
completion = asyncio.run(session.complete(
|
|
messages=[Message("user", [TextBlock("review U11")])],
|
|
tools=[ToolSchema(
|
|
name="get_pintable", description="x",
|
|
input_schema={"type": "object"},
|
|
)],
|
|
tool_choice="auto",
|
|
))
|
|
assert len(calls) == 2
|
|
assert calls[0]["thinking"] == "enabled"
|
|
assert calls[1]["thinking"] == "enabled"
|
|
asst = [m for m in calls[1]["messages"] if m.get("role") == "assistant"]
|
|
assert asst
|
|
assert asst[0]["reasoning_content"] == "pondering the schematic"
|
|
assert asst[0]["content"] == ""
|
|
assert completion.tool_calls[0].name == "get_pintable"
|
|
|
|
|
|
def test_completion_from_openai_reads_model_extra_reasoning():
|
|
fn = SimpleNamespace(name="get_pintable", arguments='{"ref":"U11"}')
|
|
tc = SimpleNamespace(id="c1", function=fn)
|
|
msg = SimpleNamespace(
|
|
content=None,
|
|
reasoning_content=None,
|
|
tool_calls=[tc],
|
|
model_extra={"reasoning_content": "need pin table"},
|
|
)
|
|
usage = SimpleNamespace(
|
|
prompt_tokens=10, completion_tokens=5,
|
|
prompt_cache_hit_tokens=0, prompt_tokens_details=None,
|
|
)
|
|
resp = SimpleNamespace(
|
|
choices=[SimpleNamespace(message=msg, finish_reason="tool_calls")],
|
|
usage=usage,
|
|
)
|
|
completion = completion_from_openai(resp)
|
|
assert completion.raw_assistant_blocks[0].reasoning_content == "need pin table"
|
|
out = messages_to_openai(
|
|
[Message("assistant", list(completion.raw_assistant_blocks))],
|
|
vision=False,
|
|
)
|
|
assert out[0]["reasoning_content"] == "need pin table"
|
|
|
|
|
|
def test_tool_schema_and_choice():
|
|
schema = ToolSchema(
|
|
name="save_pintable",
|
|
description="Save pins",
|
|
input_schema={"type": "object", "properties": {}},
|
|
)
|
|
tool = _to_openai_tool(schema)
|
|
assert tool["type"] == "function"
|
|
assert tool["function"]["name"] == "save_pintable"
|
|
assert _to_openai_tool_choice("auto") == "auto"
|
|
forced = _to_openai_tool_choice({"name": "save_pintable"})
|
|
assert forced["function"]["name"] == "save_pintable"
|
|
|
|
|
|
def test_completion_from_openai_parses_tools_and_cache():
|
|
fn = SimpleNamespace(name="submit_review", arguments='{"findings":[]}')
|
|
tc = SimpleNamespace(id="c1", function=fn)
|
|
msg = SimpleNamespace(
|
|
content="done",
|
|
reasoning_content="step by step",
|
|
tool_calls=[tc],
|
|
)
|
|
usage = SimpleNamespace(
|
|
prompt_tokens=1000,
|
|
completion_tokens=50,
|
|
prompt_cache_hit_tokens=400,
|
|
prompt_tokens_details=None,
|
|
)
|
|
resp = SimpleNamespace(
|
|
choices=[SimpleNamespace(message=msg, finish_reason="tool_calls")],
|
|
usage=usage,
|
|
)
|
|
completion = completion_from_openai(resp)
|
|
assert completion.text == "done"
|
|
assert completion.tool_calls[0].name == "submit_review"
|
|
assert completion.tool_calls[0].input == {"findings": []}
|
|
assert completion.usage.input_tokens == 600
|
|
assert completion.usage.cache_read_tokens == 400
|
|
assert completion.raw_assistant_blocks[0].reasoning_content == "step by step"
|
|
|
|
|
|
def test_local_skills_load():
|
|
md = load_skill_markdown("extract-pintable")
|
|
assert "pin table" in md.lower()
|
|
validate = load_skill_validator("extract-pintable")
|
|
assert validate is not None
|
|
errors = validate({
|
|
"mpn": "MSPM0G3507SPTR",
|
|
"component_subtype": "ic.mcu",
|
|
"component_subtype_description": "MCU",
|
|
"package_info": {"base_family": "MSPM0", "package": "LQFP-48", "pin_count": 2},
|
|
"pintable": [
|
|
{"number": 1, "name": "VCC"},
|
|
{"number": 2, "name": "GND"},
|
|
],
|
|
})
|
|
assert errors == []
|
|
|
|
|
|
def test_wroom_rejects_bare_soc_pin1_ant():
|
|
from backend.services.llm.local_skill import load_skill_validator
|
|
validate = load_skill_validator("extract-pintable")
|
|
errors = validate({
|
|
"mpn": "ESP32-S31-WROOM-3",
|
|
"component_subtype": "ic.mcu",
|
|
"package_info": {"base_family": "ESP32-S31", "package": "module", "pin_count": 2},
|
|
"pintable": [
|
|
{"number": 1, "name": "ANT"},
|
|
{"number": 2, "name": "CHIP_PU"},
|
|
{"number": 78, "name": "XTAL_N"},
|
|
{"number": 79, "name": "XTAL_P"},
|
|
],
|
|
})
|
|
assert any("module" in e.lower() or "pad" in e.lower() or "SoC" in e or "WROOM" in e for e in errors)
|
|
|
|
|
|
def test_factory_routes_deepseek(monkeypatch):
|
|
monkeypatch.setattr(settings, "deepseek_api_key", "sk-test")
|
|
from backend.services.llm.factory import get_provider_by_name
|
|
get_provider_by_name.cache_clear()
|
|
try:
|
|
provider = get_provider_by_name("deepseek")
|
|
assert provider.name == "deepseek"
|
|
finally:
|
|
get_provider_by_name.cache_clear()
|
|
|
|
|
|
def test_config_defaults_are_deepseek():
|
|
from backend.config import Settings
|
|
|
|
assert Settings.model_fields["provider_default"].default == "deepseek"
|
|
assert Settings.model_fields["deepseek_model"].default == "deepseek-flash"
|
|
assert Settings.model_fields["model_pintable_deepseek"].default == "deepseek-flash"
|
|
assert Settings.model_fields["model_validation_deepseek"].default == "deepseek-flash"
|
|
assert settings.provider_for_stage("pintable") == "deepseek"
|
|
|
|
|
|
def test_deepseek_pricing_positive():
|
|
pro = cost_for_entry({
|
|
"provider": "deepseek",
|
|
"model": "deepseek-v4-pro",
|
|
"input_tokens": 1_000_000,
|
|
"output_tokens": 0,
|
|
"cache_creation_input_tokens": 0,
|
|
"cache_read_input_tokens": 0,
|
|
})
|
|
assert pro == pytest.approx(1.32)
|
|
flash = cost_for_entry({
|
|
"provider": "deepseek",
|
|
"model": "deepseek-flash",
|
|
"input_tokens": 1_000_000,
|
|
"output_tokens": 0,
|
|
"cache_creation_input_tokens": 0,
|
|
"cache_read_input_tokens": 0,
|
|
})
|
|
assert flash == pytest.approx(0.30)
|
|
cached = cost_for_entry({
|
|
"provider": "deepseek",
|
|
"model": "deepseek-flash",
|
|
"input_tokens": 0,
|
|
"output_tokens": 0,
|
|
"cache_creation_input_tokens": 0,
|
|
"cache_read_input_tokens": 1_000_000,
|
|
})
|
|
assert cached == pytest.approx(0.006)
|
|
assert "default" in PRICING["deepseek"]
|
|
assert "deepseek-flash" in PRICING["deepseek"]
|