mk2878d commited on
Commit
73fdfd2
·
verified ·
1 Parent(s): e1e63a8

Create app.py

Browse files
Files changed (1) hide show
  1. app.py +90 -30
app.py CHANGED
@@ -10,6 +10,7 @@ import zipfile
10
 
11
  from fastapi import Request
12
  from fastapi.responses import StreamingResponse
 
13
  from gradio import Server
14
  import httpx
15
 
@@ -30,11 +31,35 @@ def whygpu():
30
  # being called again (module re-import, a reload, a duplicate call, etc.).
31
  _llama_server_started = False
32
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
33
 
34
  def _is_llama_server_up(timeout: float = 1.0) -> bool:
35
  """Return True if something is already listening on the llama-server port."""
36
  try:
37
- with socket.create_connection(("127.0.0.1", 8000), timeout=timeout):
38
  return True
39
  except OSError:
40
  return False
@@ -124,8 +149,8 @@ def start_llama_server():
124
  binary_path,
125
  "-m", model_path,
126
  "-mm", mmproj_path,
127
- "--port","8000",
128
- "--host", "127.0.0.1",
129
  "-t", "16",
130
  "-tb", "16",
131
  "-fa", "on",
@@ -153,54 +178,89 @@ def start_llama_server():
153
  return process
154
 
155
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
156
  @app.middleware("http")
157
  async def proxy_middleware(request: Request, call_next):
158
  path = request.url.path
159
- if path.startswith("/gradio_api"):
160
  return await call_next(request)
161
 
162
- url = f"http://127.0.0.1:8000{path}"
 
 
 
163
 
164
- headers = {
165
- k: v for k, v in request.headers.items()
166
- if k.lower() not in ("host", "content-length", "accept-encoding")
167
- }
 
168
  body = await request.body()
169
 
170
- client = httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=120.0))
171
  req = client.build_request(
172
  method=request.method,
173
- url=url,
174
  headers=headers,
175
- params=request.query_params,
176
  content=body,
177
  )
178
 
179
  try:
180
- response = await client.send(req, stream=True)
181
- except Exception as e:
182
- await client.aclose()
183
- return StreamingResponse(iter([f"Proxy Error: {e}".encode()]), status_code=502)
 
 
 
 
 
 
 
 
 
 
184
 
185
- async def stream_and_close():
186
  try:
187
- async for chunk in response.aiter_raw():
188
  yield chunk
 
 
 
189
  finally:
190
- await response.aclose()
191
- await client.aclose()
192
-
193
- res_headers = {
194
- k: v for k, v in response.headers.items()
195
- if k.lower() not in ("content-length", "transfer-encoding", "connection")
196
- }
197
-
198
- return StreamingResponse(
199
- stream_and_close(),
200
- status_code=response.status_code,
201
- headers=res_headers
202
  )
203
 
 
 
 
 
 
 
 
 
 
204
 
205
  if __name__ == "__main__":
206
  start_llama_server()
 
10
 
11
  from fastapi import Request
12
  from fastapi.responses import StreamingResponse
13
+ from fastapi.responses import Response
14
  from gradio import Server
15
  import httpx
16
 
 
31
  # being called again (module re-import, a reload, a duplicate call, etc.).
32
  _llama_server_started = False
33
 
34
+ LLAMA_HOST = "127.0.0.1"
35
+ LLAMA_PORT = 8000
36
+ LLAMA_BASE_URL = f"http://{LLAMA_HOST}:{LLAMA_PORT}"
37
+ GRADIO_PREFIX = "/gradio_api"
38
+
39
+ # Hop-by-hop headers (RFC 7230 section 6.1) describe a single connection and
40
+ # must never be forwarded by a proxy in either direction.
41
+ HOP_BY_HOP = {
42
+ "connection",
43
+ "keep-alive",
44
+ "proxy-authenticate",
45
+ "proxy-authorization",
46
+ "te",
47
+ "trailer",
48
+ "trailers",
49
+ "transfer-encoding",
50
+ "upgrade",
51
+ }
52
+ # Request side: also drop Host (httpx sets the right one for 127.0.0.1) and
53
+ # Accept-Encoding / Content-Length (httpx recomputes them).
54
+ REQUEST_DROP = HOP_BY_HOP | {"host", "content-length", "accept-encoding"}
55
+ # Response side: Content-Length is recomputed by StreamingResponse.
56
+ RESPONSE_DROP = HOP_BY_HOP | {"content-length"}
57
+
58
 
59
  def _is_llama_server_up(timeout: float = 1.0) -> bool:
60
  """Return True if something is already listening on the llama-server port."""
61
  try:
62
+ with socket.create_connection((LLAMA_HOST, LLAMA_PORT), timeout=timeout):
63
  return True
64
  except OSError:
65
  return False
 
149
  binary_path,
150
  "-m", model_path,
151
  "-mm", mmproj_path,
152
+ "--port", str(LLAMA_PORT),
153
+ "--host", LLAMA_HOST,
154
  "-t", "16",
155
  "-tb", "16",
156
  "-fa", "on",
 
178
  return process
179
 
180
 
181
+ # One shared client for the whole process: reuses keep-alive connections to
182
+ # llama-server instead of opening a new client (and TCP pool) per request.
183
+ _proxy_client: httpx.AsyncClient | None = None
184
+
185
+
186
+ def _get_proxy_client() -> httpx.AsyncClient:
187
+ global _proxy_client
188
+ if _proxy_client is None or _proxy_client.is_closed:
189
+ _proxy_client = httpx.AsyncClient(
190
+ base_url=LLAMA_BASE_URL,
191
+ timeout=httpx.Timeout(600.0, connect=10.0),
192
+ limits=httpx.Limits(max_connections=64, max_keepalive_connections=16),
193
+ follow_redirects=False,
194
+ )
195
+ return _proxy_client
196
+
197
+
198
  @app.middleware("http")
199
  async def proxy_middleware(request: Request, call_next):
200
  path = request.url.path
201
+ if path.startswith(GRADIO_PREFIX):
202
  return await call_next(request)
203
 
204
+ # Keep the raw query string so encoding and repeated keys are preserved.
205
+ target = path
206
+ if request.url.query:
207
+ target = f"{path}?{request.url.query}"
208
 
209
+ # Use a list of tuples so duplicate request headers are not collapsed.
210
+ headers = [
211
+ (k, v) for k, v in request.headers.items()
212
+ if k.lower() not in REQUEST_DROP
213
+ ]
214
  body = await request.body()
215
 
216
+ client = _get_proxy_client()
217
  req = client.build_request(
218
  method=request.method,
219
+ url=target,
220
  headers=headers,
 
221
  content=body,
222
  )
223
 
224
  try:
225
+ upstream = await client.send(req, stream=True)
226
+ except httpx.ConnectError:
227
+ # llama-server is not running / not listening yet.
228
+ return Response(
229
+ content=f"Proxy Error: llama-server is not reachable at {LLAMA_BASE_URL}",
230
+ status_code=503,
231
+ media_type="text/plain",
232
+ )
233
+ except httpx.HTTPError as e:
234
+ return Response(
235
+ content=f"Proxy Error: {e}",
236
+ status_code=502,
237
+ media_type="text/plain",
238
+ )
239
 
240
+ async def stream_body():
241
  try:
242
+ async for chunk in upstream.aiter_raw():
243
  yield chunk
244
+ except httpx.HTTPError as e:
245
+ # Upstream dropped mid-stream; we cannot change the status now.
246
+ print(f"[proxy] upstream stream error for {target}: {e}")
247
  finally:
248
+ await upstream.aclose()
249
+
250
+ response = StreamingResponse(
251
+ stream_body(),
252
+ status_code=upstream.status_code,
 
 
 
 
 
 
 
253
  )
254
 
255
+ # Append headers one by one (not via a dict) so multiple values, such as
256
+ # several Set-Cookie headers, are all forwarded.
257
+ for k, v in upstream.headers.multi_items():
258
+ if k.lower() in RESPONSE_DROP:
259
+ continue
260
+ response.raw_headers.append((k.lower().encode("latin-1"), v.encode("latin-1")))
261
+
262
+ return response
263
+
264
 
265
  if __name__ == "__main__":
266
  start_llama_server()