mrleo1nidandClaude Opus 5
84d8aaa848
- Факты о пользователе в SQLite, в системном промпте с номерами - Инструменты remember / update_memory / forget, автоматическое запоминание отключается - Цикл вызова инструментов со стримингом (до 5 кругов), проверка и приведение аргументов - Откат без инструментов для моделей, которые их не поддерживают, с одним предупреждением - Действия в журнале, вкладка «Память» в настройках - Фейковый OpenAI-совместимый сервер для тестов, тесты полного цикла через Assistant Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
91 lines
3.4 KiB
Python
91 lines
3.4 KiB
Python
"""Фейковый OpenAI-совместимый сервер: отдаёт заранее заданные ответы стримом."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class Reply:
|
|
text: str = ""
|
|
# (имя, аргументы JSON); аргументы отдаются несколькими кусками, как у настоящих серверов
|
|
tool_calls: list[tuple[str, str]] = field(default_factory=list)
|
|
status: int = 200
|
|
error: str = ""
|
|
|
|
|
|
class FakeLLMServer:
|
|
def __init__(self) -> None:
|
|
self.replies: list[Reply] = []
|
|
self.requests: list[dict[str, Any]] = []
|
|
server = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def log_message(self, *args: object) -> None:
|
|
pass
|
|
|
|
def do_POST(self) -> None:
|
|
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
|
|
server.requests.append(body)
|
|
reply = server.replies.pop(0) if server.replies else Reply(text="")
|
|
if reply.status != 200:
|
|
payload = json.dumps({"error": {"message": reply.error}}).encode()
|
|
self.send_response(reply.status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(payload)))
|
|
self.end_headers()
|
|
self.wfile.write(payload)
|
|
return
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/event-stream")
|
|
self.end_headers()
|
|
for delta in _deltas(reply):
|
|
chunk = {
|
|
"id": "fake",
|
|
"object": "chat.completion.chunk",
|
|
"created": 0,
|
|
"model": body["model"],
|
|
"choices": [{"index": 0, "delta": delta, "finish_reason": None}],
|
|
}
|
|
self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
|
|
self.wfile.write(b"data: [DONE]\n\n")
|
|
|
|
self._server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
threading.Thread(target=self._server.serve_forever, daemon=True).start()
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
return f"http://127.0.0.1:{self._server.server_port}/v1"
|
|
|
|
def close(self) -> None:
|
|
self._server.shutdown()
|
|
self._server.server_close()
|
|
|
|
|
|
def _deltas(reply: Reply) -> list[dict[str, Any]]:
|
|
deltas: list[dict[str, Any]] = []
|
|
for start in range(0, len(reply.text), 5):
|
|
deltas.append({"content": reply.text[start : start + 5]})
|
|
for index, (name, arguments) in enumerate(reply.tool_calls):
|
|
middle = len(arguments) // 2
|
|
deltas.append(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": index,
|
|
"id": f"call_{index}",
|
|
"type": "function",
|
|
"function": {"name": name, "arguments": arguments[:middle]},
|
|
}
|
|
]
|
|
}
|
|
)
|
|
deltas.append(
|
|
{"tool_calls": [{"index": index, "function": {"arguments": arguments[middle:]}}]}
|
|
)
|
|
return deltas
|