1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
#!/usr/bin/env python3
"""A fake OpenAI-compatible model for tests/pi-live.sh: streams a short reply
to /v1/chat/completions, after a delay taken from the last user message
("slow 3" waits 3 seconds), or fails as a provider does ("fail 401": an
authentication error). Asked to name a session, it answers "About <the
latest user message>". Prints its port, then serves until killed."""
import json
import re
import sys
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
def last_user_text(body):
for message in reversed(body.get("messages", [])):
if message.get("role") == "user":
content = message.get("content")
if isinstance(content, list):
return " ".join(part.get("text", "") for part in content if isinstance(part, dict))
return content or ""
return ""
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_GET(self):
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps({"object": "list", "data": [{"id": "fake-model", "object": "model"}]}).encode())
def do_POST(self):
body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}")
text = last_user_text(body)
if "Name this coding-agent session" in text:
said = re.findall(r"^User: (.*)$", text, re.M)
self.reply("About " + (said[-1][:40] if said else "nothing"))
return
match = re.search(r"slow (\d+(?:\.\d+)?)", text)
time.sleep(float(match.group(1)) if match else 0.2)
failure = re.search(r"fail (\d{3})", text)
if failure:
error = json.dumps({"type": "error", "error": {"type": "authentication_error", "message": "invalid x-api-key"}})
self.send_response(int(failure.group(1)))
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(error.encode())
return
self.reply("Done: fake reply.")
def reply(self, text):
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.end_headers()
def chunk(delta, finish=None, usage=None):
data = {"id": "fake", "object": "chat.completion.chunk", "created": int(time.time()), "model": "fake-model",
"choices": [{"index": 0, "delta": delta, "finish_reason": finish}]}
if usage:
data["usage"] = usage
self.wfile.write(f"data: {json.dumps(data)}\n\n".encode())
self.wfile.flush()
chunk({"role": "assistant", "content": text})
chunk({}, "stop", {"prompt_tokens": 10, "completion_tokens": 3, "total_tokens": 13})
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
print(server.server_address[1], flush=True)
server.serve_forever()
|
