103 lines
4.3 KiB
Python
103 lines
4.3 KiB
Python
"""Latest DeepSeek adapter preserves screenshots and validates browser actions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
from backend.browser import llm_factory
|
|
from backend.config import MODELS_BY_PROVIDER, Settings
|
|
|
|
|
|
class BrowserAction(BaseModel):
|
|
action: str
|
|
|
|
|
|
@pytest.fixture
|
|
def model_client(monkeypatch):
|
|
import openai
|
|
|
|
calls = []
|
|
reply = {"content": '{"action":"done"}', "finish_reason": "stop"}
|
|
|
|
class Client:
|
|
def __init__(self, **kwargs):
|
|
self.chat = SimpleNamespace(completions=SimpleNamespace(create=self.create))
|
|
self.options = kwargs
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *args):
|
|
return False
|
|
|
|
async def create(self, **kwargs):
|
|
calls.append({"options": self.options, **kwargs})
|
|
message = SimpleNamespace(content=None, tool_calls=[SimpleNamespace(type="function", function=SimpleNamespace(
|
|
name="return_structured_output", arguments=reply["content"]))])
|
|
return SimpleNamespace(
|
|
choices=[SimpleNamespace(message=message, finish_reason=reply["finish_reason"])],
|
|
usage=None,
|
|
)
|
|
|
|
monkeypatch.setattr(openai, "AsyncOpenAI", Client)
|
|
return calls, reply
|
|
|
|
|
|
def test_new_defaults_and_vision_capabilities():
|
|
cfg = Settings(_env_file=None)
|
|
assert cfg.llm_model == "deepseek-flash"
|
|
assert MODELS_BY_PROVIDER["deepseek"] == ["deepseek-flash", "deepseek-v4-pro"]
|
|
assert llm_factory.use_vision_for_model("deepseek-flash", "deepseek")
|
|
assert not llm_factory.use_vision_for_model("deepseek-v4-pro", "deepseek")
|
|
assert not llm_factory.use_vision_for_model("deepseek-chat", "deepseek")
|
|
|
|
|
|
async def test_browser_request_preserves_screenshot_and_forces_one_complete_structured_result(model_client):
|
|
from browser_use.llm.messages import ContentPartImageParam, ContentPartTextParam, ImageURL, UserMessage
|
|
|
|
calls, _reply = model_client
|
|
model = llm_factory._build_model({"provider": "deepseek", "model": "deepseek-flash", "base_url": "https://api.deepseek.com/v1"}, "TEST-KEY")
|
|
assert isinstance(model, llm_factory.DeepSeekBrowserChat)
|
|
image = "data:image/png;base64,aW1hZ2U="
|
|
result = await model.ainvoke([UserMessage(content=[ContentPartTextParam(text="观察页面"), ContentPartImageParam(image_url=ImageURL(url=image))])], BrowserAction)
|
|
assert result.completion.action == "done"
|
|
user = next(row for row in calls[0]["messages"] if row["role"] == "user")
|
|
assert user["content"][1]["image_url"]["url"] == image
|
|
assert "response_format" not in calls[0]
|
|
assert calls[0]["tool_choice"] == {"type": "function", "function": {"name": "return_structured_output"}}
|
|
assert calls[0]["tools"][0]["function"]["parameters"] == BrowserAction.model_json_schema()
|
|
assert calls[0]["extra_body"] == {"thinking": {"type": "disabled"}}
|
|
assert calls[0]["options"]["max_retries"] == 0
|
|
assert "return_structured_output" in calls[0]["messages"][0]["content"]
|
|
assert calls[0]["messages"][-1]["role"] == "user"
|
|
assert model.provider == "deepseek"
|
|
|
|
|
|
async def test_invalid_browser_action_is_not_returned_as_executable(model_client):
|
|
from browser_use.llm.exceptions import ModelProviderError
|
|
from browser_use.llm.messages import UserMessage
|
|
|
|
_calls, reply = model_client
|
|
reply["content"] = '{"unexpected":"no action"}'
|
|
model = llm_factory.DeepSeekBrowserChat("deepseek-flash", "TEST-KEY")
|
|
with pytest.raises(ModelProviderError, match="不符合"):
|
|
await model.ainvoke([UserMessage(content="观察页面")], BrowserAction)
|
|
|
|
|
|
async def test_truncated_browser_action_is_not_executed(model_client):
|
|
from browser_use.llm.exceptions import ModelOutputTruncatedError
|
|
from browser_use.llm.messages import UserMessage
|
|
|
|
_calls, reply = model_client
|
|
reply["finish_reason"] = "length"
|
|
model = llm_factory.DeepSeekBrowserChat("deepseek-flash", "TEST-KEY")
|
|
with pytest.raises(ModelOutputTruncatedError):
|
|
await model.ainvoke([UserMessage(content="观察页面")], BrowserAction)
|
|
|
|
|
|
def test_key_is_not_exposed_in_model_repr():
|
|
assert "TEST-PRIVATE-KEY" not in repr(llm_factory.DeepSeekBrowserChat("deepseek-flash", "TEST-PRIVATE-KEY"))
|