|
10 | 10 | export OPENAI_BASE_URL=http://localhost:8080/v1 OPENAI_API_KEY=unused |
11 | 11 | # then plain `openai`, curl, or any SDK just works |
12 | 12 |
|
13 | | -Each request: reserve a session, forward the body, release the session. call_runner does |
14 | | -the 402 payment challenge internally, so the client never sees discovery or payment. |
15 | | -(Release matters: the runner has capacity 1, so an unreleased session would block the |
16 | | -next call.) |
| 13 | +Each request: discover the runner, forward the body. The runner is single-shot, so the |
| 14 | +orchestrator reserves a session for the call and releases it when the response returns -- |
| 15 | +the gateway manages no session at all. call_runner does the 402 payment challenge |
| 16 | +internally, so the client never sees discovery or payment. Pricing is metered, so the |
| 17 | +call keeps paying for as long as it runs, which for a long generation is the point. |
17 | 18 |
|
18 | 19 | Livepeer integration (grep `# Livepeer:`): |
19 | | - 1. reserve_session() — discover the runner, reserve a session |
20 | | - 2. call_runner() — forward the request through the orchestrator (pays 402) |
21 | | - 3. stop_runner_session() — release the session |
| 20 | + 1. runner_selector() — discover the runner advertising this app |
| 21 | + 2. call_runner() — forward the request through the orchestrator (pays 402) |
22 | 22 |
|
23 | | -These three calls are the *entire* Livepeer surface. They live here, and only here, |
| 23 | +These two calls are the *entire* Livepeer surface. They live here, and only here, |
24 | 24 | because an OpenAI client can't do discovery or settle payments itself — so `client.py` |
25 | 25 | (and any OpenAI SDK/curl) stays 100% stock, unaware of Livepeer. |
26 | 26 |
|
|
34 | 34 |
|
35 | 35 | import argparse |
36 | 36 | import logging |
37 | | -from contextlib import suppress |
38 | 37 |
|
39 | 38 | from aiohttp import web |
40 | 39 |
|
41 | | -from livepeer_gateway.live_runner import call_runner, stop_runner_session |
42 | | -from livepeer_gateway.selection import reserve_session |
| 40 | +from livepeer_gateway.errors import LivepeerHTTPError |
| 41 | +from livepeer_gateway.live_runner import call_runner |
| 42 | +from livepeer_gateway.selection import runner_selector |
43 | 43 |
|
44 | 44 | APP_ID = "vllm/qwen2.5-0.5b-instruct" |
45 | 45 |
|
@@ -71,49 +71,59 @@ def main() -> None: |
71 | 71 | async def _forward(request: web.Request) -> web.StreamResponse: |
72 | 72 | payload = await request.json() |
73 | 73 | runner_path = request.path # e.g. /v1/chat/completions |
74 | | - session = await reserve_session( |
| 74 | + cursor = await runner_selector( # Livepeer: 1 |
75 | 75 | discovery_url=args.discovery, # omit if the signer does discovery itself |
76 | 76 | app=APP_ID, |
| 77 | + ) |
| 78 | + runner = cursor.candidates[0] |
| 79 | + runner_url = runner.url.rstrip("/") + runner_path |
| 80 | + |
| 81 | + # When the OpenAI client asks for stream=True the runner replies with |
| 82 | + # text/event-stream; pipe those chunks straight through with stream=True |
| 83 | + # so tokens reach the client as they arrive instead of buffering the blob. |
| 84 | + if payload.get("stream"): |
| 85 | + async with await call_runner( # Livepeer: 2 (streaming) |
| 86 | + runner=runner, # discovery metadata tells call_runner the price unit |
| 87 | + runner_url=runner_url, |
| 88 | + payload=payload, |
| 89 | + signer_url=signer_url, |
| 90 | + stream=True, |
| 91 | + ) as stream: |
| 92 | + resp = web.StreamResponse( |
| 93 | + status=stream.status, |
| 94 | + headers={ |
| 95 | + "Content-Type": stream.content_type or "text/event-stream" |
| 96 | + }, |
| 97 | + ) |
| 98 | + await resp.prepare(request) |
| 99 | + async for ( |
| 100 | + chunk |
| 101 | + ) in stream.aiter_bytes(): # raw bytes -> keep SSE framing |
| 102 | + await resp.write(chunk) |
| 103 | + await resp.write_eof() |
| 104 | + return resp |
| 105 | + |
| 106 | + result = await call_runner( # Livepeer: 2 |
| 107 | + runner=runner, # discovery metadata tells call_runner the price unit |
| 108 | + runner_url=runner_url, |
| 109 | + payload=payload, |
77 | 110 | signer_url=signer_url, |
78 | | - ) # Livepeer: 1 |
| 111 | + ) |
| 112 | + return web.json_response(result.data) |
79 | 113 |
|
| 114 | + async def _forward_or_error(request: web.Request) -> web.StreamResponse: |
| 115 | + # A single-shot call holds a capacity slot for its duration, so a busy runner |
| 116 | + # answers 503. Hand that back as JSON an OpenAI client can read. |
80 | 117 | try: |
81 | | - runner_url = session.app_url.rstrip("/") + runner_path |
82 | | - |
83 | | - # When the OpenAI client asks for stream=True the runner replies with |
84 | | - # text/event-stream; pipe those chunks straight through with stream=True |
85 | | - # so tokens reach the client as they arrive instead of buffering the blob. |
86 | | - if payload.get("stream"): |
87 | | - async with await call_runner( # Livepeer: 2 (streaming) |
88 | | - runner_url=runner_url, |
89 | | - payload=payload, |
90 | | - signer_url=signer_url, |
91 | | - stream=True, |
92 | | - ) as stream: |
93 | | - resp = web.StreamResponse( |
94 | | - status=stream.status, |
95 | | - headers={ |
96 | | - "Content-Type": stream.content_type or "text/event-stream" |
97 | | - }, |
98 | | - ) |
99 | | - await resp.prepare(request) |
100 | | - async for ( |
101 | | - chunk |
102 | | - ) in stream.aiter_bytes(): # raw bytes -> preserve SSE framing |
103 | | - await resp.write(chunk) |
104 | | - await resp.write_eof() |
105 | | - return resp |
106 | | - |
107 | | - result = await call_runner( |
108 | | - runner_url=runner_url, payload=payload, signer_url=signer_url |
109 | | - ) # Livepeer: 2 |
110 | | - return web.json_response(result.data) |
111 | | - finally: |
112 | | - with suppress(Exception): |
113 | | - await stop_runner_session(session) # Livepeer: 3 |
| 118 | + return await _forward(request) |
| 119 | + except LivepeerHTTPError as exc: |
| 120 | + return web.json_response( |
| 121 | + {"error": {"message": str(exc), "type": "livepeer_error"}}, |
| 122 | + status=exc.status_code, |
| 123 | + ) |
114 | 124 |
|
115 | 125 | app = web.Application() |
116 | | - app.router.add_post("/v1/{tail:.*}", _forward) # forward every OpenAI path |
| 126 | + app.router.add_post("/v1/{tail:.*}", _forward_or_error) # forward every OpenAI path |
117 | 127 | log.info( |
118 | 128 | "gateway on http://%s:%d/v1 -> %s (signer=%s)", |
119 | 129 | args.host, |
|
0 commit comments