108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
import json
|
|
|
|
from mulit_agent_app.infrastructure.openai_compatible_provider import OpenAICompatibleProvider
|
|
|
|
|
|
class _Credentials:
|
|
def load_llm_api_key(self):
|
|
return "test-key"
|
|
|
|
|
|
class _Response:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def read(self):
|
|
return json.dumps({"choices": [{"message": {"content": " 模型结果 "}}]}).encode("utf-8")
|
|
|
|
|
|
def test_provider_posts_a_chat_completion(monkeypatch, tmp_path):
|
|
captured = {}
|
|
|
|
def fake_urlopen(request, timeout):
|
|
captured["url"] = request.full_url
|
|
captured["body"] = json.loads(request.data.decode("utf-8"))
|
|
captured["authorization"] = request.get_header("Authorization")
|
|
captured["timeout"] = timeout
|
|
return _Response()
|
|
|
|
monkeypatch.setattr("mulit_agent_app.infrastructure.openai_compatible_provider.urlopen", fake_urlopen)
|
|
|
|
output = OpenAICompatibleProvider("https://api.example.test/v1", "example-model", _Credentials()).generate("请总结")
|
|
|
|
assert output == "模型结果"
|
|
assert captured == {
|
|
"url": "https://api.example.test/v1/chat/completions",
|
|
"body": {"model": "example-model", "messages": [{"role": "user", "content": "请总结"}], "stream": False},
|
|
"authorization": "Bearer test-key",
|
|
"timeout": 90,
|
|
}
|
|
|
|
|
|
def test_provider_removes_reasoning_tags_before_returning_result(monkeypatch):
|
|
class ThinkingResponse(_Response):
|
|
def read(self):
|
|
return json.dumps({"choices": [{"message": {"content": "<think>内部推理</think>最终答案"}}]}).encode("utf-8")
|
|
|
|
monkeypatch.setattr(
|
|
"mulit_agent_app.infrastructure.openai_compatible_provider.urlopen", lambda request, timeout: ThinkingResponse()
|
|
)
|
|
|
|
output = OpenAICompatibleProvider("https://api.example.test/v1", "example-model", _Credentials()).generate("请总结")
|
|
|
|
assert output == "最终答案"
|
|
|
|
|
|
def test_provider_executes_confirmed_local_tool_call_before_returning_answer(monkeypatch):
|
|
responses = [
|
|
{
|
|
"choices": [
|
|
{
|
|
"message": {
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": "call-1",
|
|
"type": "function",
|
|
"function": {"name": "run_powershell", "arguments": '{"command":"Get-Date"}'},
|
|
}
|
|
],
|
|
}
|
|
}
|
|
]
|
|
},
|
|
{"choices": [{"message": {"content": "已实际完成本机操作。"}}]},
|
|
]
|
|
captured: list[dict[str, object]] = []
|
|
|
|
class Response:
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
|
|
def read(self):
|
|
return json.dumps(self.payload).encode("utf-8")
|
|
|
|
def fake_urlopen(request, timeout):
|
|
captured.append(json.loads(request.data.decode("utf-8")))
|
|
return Response(responses.pop(0))
|
|
|
|
monkeypatch.setattr("mulit_agent_app.infrastructure.openai_compatible_provider.urlopen", fake_urlopen)
|
|
commands: list[str] = []
|
|
output = OpenAICompatibleProvider("https://api.example.test/v1", "example-model", _Credentials()).generate(
|
|
"在桌面创建文件夹", tool_executor=lambda command: commands.append(command) or "已创建。"
|
|
)
|
|
|
|
assert output == "已实际完成本机操作。"
|
|
assert commands == ["Get-Date"]
|
|
assert captured[0]["tools"]
|
|
assert captured[1]["messages"][-1] == {"role": "tool", "tool_call_id": "call-1", "content": "已创建。"}
|