| 255 | |
| 256 | |
| 257 | class OllamaLLM: |
| 258 | name = "ollama" |
| 259 | |
| 260 | def __init__(self, model: str, base_url: str = "http://localhost:11434", timeout: float = 120.0) -> None: |
| 261 | self.model = model |
| 262 | self._base_url = base_url.rstrip("/") |
| 263 | self._timeout = timeout |
| 264 | |
| 265 | def generate( |
| 266 | self, |
| 267 | prompt: str, |
| 268 | *, |
| 269 | system: Optional[str] = None, |
| 270 | temperature: float = 0.0, |
| 271 | max_tokens: int = 1024, |
| 272 | ) -> LLMResponse: |
| 273 | payload: dict[str, object] = { |
| 274 | "model": self.model, |
| 275 | "prompt": prompt, |
| 276 | "stream": False, |
| 277 | "options": {"temperature": temperature, "num_predict": max_tokens}, |
| 278 | } |
| 279 | if system: |
| 280 | payload["system"] = system |
| 281 | |
| 282 | response = requests.post( |
| 283 | f"{self._base_url}/api/generate", |
| 284 | json=payload, |
| 285 | timeout=self._timeout, |
| 286 | ) |
| 287 | response.raise_for_status() |
| 288 | data = response.json() |
| 289 | return LLMResponse( |
| 290 | text=str(data.get("response") or ""), |
| 291 | model=self.model, |
| 292 | provider=self.name, |
| 293 | prompt_tokens=int(data.get("prompt_eval_count") or 0), |
| 294 | completion_tokens=int(data.get("eval_count") or 0), |
| 295 | ) |
| 296 | |
| 297 | def stream_generate( |
| 298 | self, |
| 299 | prompt: str, |
| 300 | *, |
| 301 | system: Optional[str] = None, |
| 302 | temperature: float = 0.0, |
| 303 | max_tokens: int = 1024, |
| 304 | ) -> Iterator[str]: |
| 305 | payload: dict[str, object] = { |
| 306 | "model": self.model, |
| 307 | "prompt": prompt, |
| 308 | "stream": True, |
| 309 | "options": {"temperature": temperature, "num_predict": max_tokens}, |
| 310 | } |
| 311 | if system: |
| 312 | payload["system"] = system |
| 313 | |
| 314 | with requests.post( |
no outgoing calls
no test coverage detected