""" FastAPI сервер — OpenAI-compatible REST API поверх Ollama Самодостаточный файл: весь код провайдера встроен. Эндпоинты: GET /health GET /v1/models POST /v1/chat/completions (tools, stream) POST /v1/embeddings """ from __future__ import annotations import json import os import time import uuid from dataclasses import dataclass, field from typing import Any, AsyncGenerator, Dict, List, Optional, Union import requests from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse from pydantic import BaseModel, Field # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- # ИСПРАВЛЕНИЕ: Заменен localhost на 127.0.0.1 во избежание зависаний из-за IPv6 OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://127.0.0.1:11434") OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "llama3.2") OLLAMA_TIMEOUT = int(os.getenv("OLLAMA_TIMEOUT", "120")) # --------------------------------------------------------------------------- # Internal dataclasses # --------------------------------------------------------------------------- @dataclass class FunctionCall: name: str arguments: str @dataclass class ToolCall: id: str type: str = "function" function: FunctionCall = None @dataclass class Message: role: str content: Optional[str] = None tool_calls: Optional[List[ToolCall]] = None tool_call_id: Optional[str] = None # --------------------------------------------------------------------------- # Converters # --------------------------------------------------------------------------- def _messages_to_ollama(messages: List[dict]) -> List[dict]: result = [] for msg in messages: role = msg["role"] content = msg.get("content") or "" if role == "tool": result.append({"role": "tool", "content": content}) continue if role == "assistant" and msg.get("tool_calls"): tcs = [] for tc in msg["tool_calls"]: fn = tc["function"] args = fn["arguments"] tcs.append({"function": { "name": fn["name"], "arguments": json.loads(args) if isinstance(args, str) else args, }}) result.append({"role": "assistant", "content": content, "tool_calls": tcs}) continue if isinstance(content, list): texts, images = [], [] for block in content: if block.get("type") == "text": texts.append(block["text"]) elif block.get("type") == "image_url": url = block["image_url"].get("url", "") images.append(url.split(",", 1)[1] if url.startswith("data:") else url) entry: dict = {"role": role, "content": " ".join(texts)} if images: entry["images"] = images result.append(entry) continue result.append({"role": role, "content": content}) return result def _tools_to_ollama(tools: Optional[List[dict]]) -> Optional[List[dict]]: if not tools: return None return [ {"type": "function", "function": { "name": t["function"]["name"], "description": t["function"].get("description", ""), "parameters": t["function"].get("parameters", {}), }} for t in tools if t.get("type") == "function" ] def _parse_message(msg: dict) -> Message: tcs_raw = msg.get("tool_calls", []) tool_calls = None if tcs_raw: tool_calls = [] for tc in tcs_raw: fn = tc.get("function", {}) args = fn.get("arguments", {}) if isinstance(args, dict): args = json.dumps(args, ensure_ascii=False) tool_calls.append(ToolCall( id=f"call_{uuid.uuid4().hex[:8]}", type="function", function=FunctionCall(name=fn.get("name", ""), arguments=args), )) return Message( role=msg.get("role", "assistant"), content=msg.get("content") or None, tool_calls=tool_calls, ) def _finish_reason(msg: dict, done: bool) -> str: return "tool_calls" if msg.get("tool_calls") else ("stop" if done else "length") # --------------------------------------------------------------------------- # HTTP session # --------------------------------------------------------------------------- _session = requests.Session() _session.headers.update({"Content-Type": "application/json"}) def _post(path: str, payload: dict, stream: bool = False): r = _session.post(f"{OLLAMA_BASE_URL}{path}", json=payload, stream=stream, timeout=OLLAMA_TIMEOUT) r.raise_for_status() return r def _get(path: str, timeout: int = 5): # ИСПРАВЛЕНИЕ: Таймаут для обычных GET запросов снижен до 5 сек, чтобы не ждать 2 минуты при неработающей Ollama r = _session.get(f"{OLLAMA_BASE_URL}{path}", timeout=timeout) r.raise_for_status() return r # --------------------------------------------------------------------------- # FastAPI # --------------------------------------------------------------------------- app = FastAPI( title="Ollama OpenAI-Compatible API", description="OpenAI-compatible REST API backed by Ollama", version="1.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # --------------------------------------------------------------------------- # Pydantic schemas # --------------------------------------------------------------------------- class _FunctionDef(BaseModel): name: str description: Optional[str] = None parameters: Optional[Dict[str, Any]] = None class _Tool(BaseModel): type: str = "function" function: _FunctionDef class _Message(BaseModel): role: str content: Optional[Union[str, List[Dict[str, Any]]]] = None tool_calls: Optional[List[Dict[str, Any]]] = None tool_call_id: Optional[str] = None name: Optional[str] = None class ChatRequest(BaseModel): model: str = Field(default=OLLAMA_MODEL) messages: List[_Message] tools: Optional[List[_Tool]] = None tool_choice: Optional[Union[str, Dict[str, Any]]] = None temperature: Optional[float] = None top_p: Optional[float] = None max_tokens: Optional[int] = None stream: bool = False stop: Optional[Union[str, List[str]]] = None seed: Optional[int] = None class EmbeddingRequest(BaseModel): model: str = Field(default=OLLAMA_MODEL) input: Union[str, List[str]] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _msg_to_dict(m: _Message) -> dict: d: dict = {"role": m.role} if m.content is not None: d["content"] = m.content if m.tool_calls: d["tool_calls"] = m.tool_calls if m.tool_call_id: d["tool_call_id"] = m.tool_call_id if m.name: d["name"] = m.name return d def _tool_to_dict(t: _Tool) -> dict: return {"type": t.type, "function": { "name": t.function.name, "description": t.function.description or "", "parameters": t.function.parameters or {}, }} async def _sse_stream(model: str, payload: dict) -> AsyncGenerator[str, None]: cid = f"chatcmpl-{uuid.uuid4().hex}" created = int(time.time()) with _session.post(f"{OLLAMA_BASE_URL}/api/chat", json=payload, stream=True, timeout=OLLAMA_TIMEOUT) as resp: resp.raise_for_status() for line in resp.iter_lines(): if not line: continue try: data = json.loads(line) except json.JSONDecodeError: continue msg_data = data.get("message", {}) done = data.get("done", False) finish_reason = _finish_reason(msg_data, done) if done else None delta: dict = {"role": "assistant", "content": msg_data.get("content") or None} if msg_data.get("tool_calls"): tcs = [] for i, tc in enumerate(msg_data["tool_calls"]): fn = tc.get("function", {}) args = fn.get("arguments", {}) if isinstance(args, dict): args = json.dumps(args, ensure_ascii=False) tcs.append({ "index": i, "id": f"call_{uuid.uuid4().hex[:8]}", "type": "function", "function": {"name": fn.get("name", ""), "arguments": args}, }) delta["tool_calls"] = tcs chunk = { "id": cid, "object": "chat.completion.chunk", "created": created, "model": model, "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}], } yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n" if done: break yield "data: [DONE]\n\n" # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get("/health") def health(): try: data = _get("/api/tags").json() return {"status": "ok", "ollama": OLLAMA_BASE_URL, "models_count": len(data.get("models", []))} except Exception as e: raise HTTPException(status_code=503, detail=f"Ollama недоступен: {e}") @app.get("/v1/models") def list_models(): try: data = _get("/api/tags").json() return {"object": "list", "data": [ {"id": m["name"], "object": "model", "created": int(time.time()), "owned_by": "ollama"} for m in data.get("models", []) ]} except Exception as e: raise HTTPException(status_code=503, detail=str(e)) @app.post("/v1/chat/completions") async def chat_completions(req: ChatRequest): messages = [_msg_to_dict(m) for m in req.messages] tools = [_tool_to_dict(t) for t in req.tools] if req.tools else None ollama_messages = _messages_to_ollama(messages) ollama_tools = _tools_to_ollama(tools) options: dict = {} if req.temperature is not None: options["temperature"] = req.temperature if req.top_p is not None: options["top_p"] = req.top_p if req.max_tokens is not None: options["num_predict"] = req.max_tokens if req.stop is not None: options["stop"] = [req.stop] if isinstance(req.stop, str) else req.stop if req.seed is not None: options["seed"] = req.seed payload: dict = {"model": req.model, "messages": ollama_messages, "stream": req.stream} if options: payload["options"] = options if ollama_tools: payload["tools"] = ollama_tools if req.stream: return StreamingResponse( _sse_stream(req.model, payload), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) try: resp_data = _post("/api/chat", payload).json() except Exception as e: raise HTTPException(status_code=502, detail=str(e)) msg_data = resp_data.get("message", {}) message = _parse_message(msg_data) done = resp_data.get("done", True) finish = _finish_reason(msg_data, done) message_dict: dict = {"role": message.role, "content": message.content} if message.tool_calls: message_dict["tool_calls"] = [ {"id": tc.id, "type": tc.type, "function": {"name": tc.function.name, "arguments": tc.function.arguments}} for tc in message.tool_calls ] result: dict = { "id": f"chatcmpl-{uuid.uuid4().hex}", "object": "chat.completion", "created": int(time.time()), "model": req.model, "choices": [{"index": 0, "message": message_dict, "finish_reason": finish}], } pt = resp_data.get("prompt_eval_count", 0) ct = resp_data.get("eval_count", 0) if pt or ct: result["usage"] = {"prompt_tokens": pt, "completion_tokens": ct, "total_tokens": pt + ct} return JSONResponse(result) @app.post("/v1/embeddings") def embeddings(req: EmbeddingRequest): texts = [req.input] if isinstance(req.input, str) else req.input data_out: list = [] total = 0 try: for i, text in enumerate(texts): r = _post("/api/embed", {"model": req.model, "input": text}).json() vec = r.get("embeddings", [[]])[0] data_out.append({"object": "embedding", "index": i, "embedding": vec}) total += r.get("prompt_eval_count", len(text.split())) except Exception as e: raise HTTPException(status_code=502, detail=str(e)) return {"object": "list", "data": data_out, "model": req.model, "usage": {"prompt_tokens": total, "total_tokens": total}} # --------------------------------------------------------------------------- # /v1/mess — веб-интерфейс чата # --------------------------------------------------------------------------- from fastapi.responses import HTMLResponse CHAT_HTML = """ Ollama Chat

Ollama Chat

OpenAI-compatible API
🦙

Начни диалог с моделью

Поддерживается Markdown и блоки кода
""" @app.get("/v1/mess", response_class=HTMLResponse) def chat_ui(): return HTMLResponse(content=CHAT_HTML)