File size: 11,625 Bytes
e6f80f3 | 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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | """OpenAI-compatible endpoint that speaks the WorldEdge request (scope doc Appendix B, fields in interface.py).
python -m fastfill.serve --model outputs/ff-v1-merged --port 8001
# EmbodiedGen side: WORLDEDGE_FASTFILL_URL=http://<host>:8001/v1 WORLDEDGE_FASTFILL_MODEL=ff-v1-merged
python -m fastfill.serve # self-check (no model)
The last user message is the request JSON (the body's "model" field is not read). The reply's content is
placements_from_text's JSON ({"placements": [{"id", "position_m", "yaw_rad", "support_parent", "support_surface"?}], "raw_placements": [...],
"validation": {"raw", "repaired"}, "error": null, "unsupported": [...]}) plus "model" (--name, default the checkpoint
directory's name) and "checkpoint" (its absolute path): EmbodiedGen caches that content under its own
WORLDEDGE_FASTFILL_MODEL, so set that to the served name, and a cached layout still says which checkpoint made it.
A request with nothing to place (no objects, or all anchored to a wall or ceiling) is answered 200 with no placements
and its unsupported list, without calling the model (training never has an empty room).
HTTP 422 and no layout when the request is rejected (a prompt of --max_len tokens or more included: prompt_too_long),
the answer does not parse (a missing or extra object included), or the repaired layout still fails the canonical
checks: FastFill places every object validly or fails back to the caller, it never drops one. Every error body carries
model and checkpoint; a 422 after generation also the model's text (raw_text) and the unsupported list.
Greedy decoding, one request at a time.
The JSON body requires a positive Content-Length of at most MAX_REQUEST_BYTES (1 MiB). Missing/invalid lengths
and malformed or excessively nested JSON receive 400; oversized bodies receive 413 before any body read.
"""
import argparse
import json
import os
import sys
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from fastfill.interface import placements_from_text, request_to_room
from fastfill.scene import messages, prompt_text
MAX_REQUEST_BYTES = 1024 * 1024
class PromptTooLong(ValueError):
"""Raised by generate() for a prompt of max_len tokens or more (vLLM cannot start on it): answered with 422."""
def answer(req, generate):
"""Request dict -> (HTTP status, body). generate(model messages) -> model text, or raises PromptTooLong."""
try:
room, ctx = request_to_room(req)
except ValueError as e:
return 422, {"error": {"message": str(e), "type": "invalid_request_error"}}
if not room["objects"]: # the empty answer is the only valid one: not generated
return 200, placements_from_text('{"placements":[]}', room, ctx)
try:
text = generate(messages(room, room["constraints"], with_target=False))
except PromptTooLong as e:
return 422, {"error": {"message": str(e), "type": "invalid_request_error", "unsupported": ctx["unsupported"]}}
out = placements_from_text(text, room, ctx)
if out["error"]:
return 422, {"error": {"message": f"model answer rejected: {out['error']}", "type": "invalid_layout",
"raw_text": text, "unsupported": out["unsupported"]}}
if not out["validation"]["repaired"]["ok"]: # every object must be placed validly; else the caller decides
return 422, {"error": {"message": "layout fails validation even after repair", "type": "invalid_layout",
"raw_placements": out["raw_placements"], "validation": out["validation"],
"raw_text": text, "unsupported": out["unsupported"]}}
return 200, out
def make_handler(generate, name, checkpoint):
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
if not self.path.endswith("/chat/completions"):
return self.reply(404, {"error": {"message": "POST .../v1/chat/completions"}})
try:
length = int(self.headers["Content-Length"])
if length <= 0:
raise ValueError("Content-Length must be positive")
if length > MAX_REQUEST_BYTES:
return self.reply(413, {"error": {"message": f"body exceeds {MAX_REQUEST_BYTES} bytes",
"type": "invalid_request_error"}})
body = json.loads(self.rfile.read(length))
req = json.loads(body["messages"][-1]["content"])
except (ValueError, KeyError, IndexError, TypeError, RecursionError) as e:
return self.reply(400, {"error": {"message": f"bad body: {e!r}", "type": "invalid_request_error"}})
try:
status, out = answer(req, generate)
except Exception as e: # a bug or a dead model: say so instead of dropping the connection
return self.reply(500, {"error": {"message": repr(e), "type": "server_error"}})
if status == 200:
content = json.dumps({**out, "model": name, "checkpoint": checkpoint})
out = {"id": f"fastfill-{time.time_ns()}", "object": "chat.completion", "created": int(time.time()),
"model": name, "choices": [{"index": 0, "finish_reason": "stop",
"message": {"role": "assistant", "content": content}}],
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}}
self.reply(status, out)
def reply(self, status, obj):
if status != 200:
obj["error"].update(model=name, checkpoint=checkpoint)
data = json.dumps(obj).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
return Handler
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--model", required=True, help="merged checkpoint dir")
ap.add_argument("--name", help="model name echoed in replies (default: the checkpoint directory's name)")
ap.add_argument("--host", default="0.0.0.0")
ap.add_argument("--port", type=int, default=8001)
ap.add_argument("--max_len", type=int, default=40960, help="model sequence limit; the answer may use what the prompt leaves")
a = ap.parse_args()
from vllm import LLM, SamplingParams
llm = LLM(model=a.model, max_model_len=a.max_len)
tok = llm.get_tokenizer()
def generate(msgs):
p = prompt_text(tok, msgs)
n = len(tok(p)["input_ids"])
if n >= a.max_len: # vLLM raises on such a prompt: refuse it as a bad request instead
raise PromptTooLong(f"prompt_too_long: {n} tokens, limit {a.max_len}")
sp = SamplingParams(temperature=0, max_tokens=a.max_len - n)
return llm.generate([p], sp, use_tqdm=False)[0].outputs[0].text
ckpt = os.path.abspath(a.model) if os.path.isdir(a.model) else a.model # a dir, or a hub id as given
name = a.name or os.path.basename(os.path.normpath(ckpt))
print(f"FastFill endpoint: http://{a.host}:{a.port}/v1 model {name} = {ckpt}", flush=True)
HTTPServer((a.host, a.port), make_handler(generate, name, ckpt)).serve_forever()
if __name__ == "__main__":
if sys.argv[1:]:
main()
else: # self-check: answer() and the HTTP wrapper, a stub model
import threading
import urllib.error
import urllib.request
calls = []
def stub(text):
def gen(msgs):
calls.append(json.loads(msgs[1]["content"]))
if text is None:
raise PromptTooLong("prompt_too_long: 50000 tokens, limit 40960")
return text
return gen
req = {"room": {"boundary_xy": [[0, 0], [4, 0], [4, 4], [0, 4]], "height_m": 2.8,
"fixed_geometry": [{"id": "col_1", "category": "column", "size_xyz_m": [0.3, 0.3, 2.8],
"position_m": [3.5, 3.5, 0]}]},
"objects_to_place": [{"id": "table_1", "category": "table", "size_xyz_m": [1.4, 0.8, 0.75]},
{"id": "cup_1", "category": "cup", "size_xyz_m": [0.08, 0.08, 0.1]},
{"id": "clock_1", "category": "clock", "size_xyz_m": [0.3, 0.05, 0.3], "anchor": "wall"}]}
good = '{"placements":[{"id":"table_1","pos":[2,2,0],"yaw":0},{"id":"cup_1","on":"table_1","pos":[2,2,0.75],"yaw":0}]}'
clock = [{"id": "clock_1", "reason": "unsupported_anchor"}]
st, out = answer(req, stub(good))
assert st == 200 and len(calls) == 1 and out["validation"]["repaired"]["ok"] and out["unsupported"] == clock, out
# nothing placeable: no objects, or only wall / ceiling items -> 200, empty placements, the model never called
for objs, unsup in (([], []), (req["objects_to_place"][2:], clock)):
st, out = answer({**req, "objects_to_place": objs}, stub("never"))
assert st == 200 and len(calls) == 1 and out["placements"] == [] and out["unsupported"] == unsup, out
assert out["error"] is None and out["validation"]["repaired"]["ok"], out
# 422 after generation keeps the model's text and the unsupported list: bad JSON, a missing object, a layout
# that fails after repair (table through the column), a prompt over the limit (no text)
on_col = good.replace("[2,2,0]", "[3.4,3.4,0]").replace("[2,2,0.75]", "[3.4,3.4,0.75]")
for text, kind in (("{not json", "invalid_layout"), (good.split(',{"id":"cup_1"')[0] + "]}", "invalid_layout"),
(on_col, "invalid_layout"), (None, "invalid_request_error")):
st, out = answer(req, stub(text))
e = out["error"]
assert st == 422 and e["type"] == kind and e["unsupported"] == clock and e.get("raw_text") == text, out
assert out["error"]["message"].startswith("prompt_too_long")
assert answer(req, stub(on_col))[1]["error"]["validation"]["repaired"]["fixed_blocking"] == [["table_1", "col_1"]]
# over HTTP: every reply names the served model and checkpoint (200 in the content EmbodiedGen caches)
texts = [good, "{not json"]
srv = HTTPServer(("127.0.0.1", 0), make_handler(lambda m: texts.pop(0), "ff-test", "/ckpt/ff-test"))
threading.Thread(target=srv.serve_forever, daemon=True).start()
def post(obj):
body = json.dumps({"model": "fastfill", "messages": [{"role": "user", "content": json.dumps(obj)}]}).encode()
try:
with urllib.request.urlopen(f"http://127.0.0.1:{srv.server_port}/v1/chat/completions", body) as f:
return f.status, json.load(f)
except urllib.error.HTTPError as f:
return f.code, json.load(f)
st, rep = post(req)
content = json.loads(rep["choices"][0]["message"]["content"])
assert st == 200 and rep["model"] == "ff-test" and (content["model"], content["checkpoint"]) == ("ff-test", "/ckpt/ff-test")
st, rep = post(req)
assert st == 422 and rep["error"]["model"] == "ff-test" and rep["error"]["raw_text"] == "{not json", rep
assert post({"room": {}})[0] == 422 and post({"room": {}})[1]["error"]["checkpoint"] == "/ckpt/ff-test"
srv.shutdown()
print("serve.py self-check ok")
|