Use local or remote Ollama server to generate text.
(self, history, verbose=False)
| 164 | return thought |
| 165 | |
| 166 | def ollama_fn(self, history, verbose=False): |
| 167 | """ |
| 168 | Use local or remote Ollama server to generate text. |
| 169 | """ |
| 170 | thought = "" |
| 171 | if self.is_local: |
| 172 | server_port = self.server_address.split(":")[-1] if ":" in str(self.server_address) else "11434" |
| 173 | host = f"{self.internal_url}:{server_port}" |
| 174 | else: |
| 175 | host = f"http://{self.server_address}" |
| 176 | client = OllamaClient(host=host) |
| 177 | |
| 178 | try: |
| 179 | stream = client.chat( |
| 180 | model=self.model, |
| 181 | messages=history, |
| 182 | stream=True, |
| 183 | ) |
| 184 | for chunk in stream: |
| 185 | if verbose: |
| 186 | print(chunk["message"]["content"], end="", flush=True) |
| 187 | thought += chunk["message"]["content"] |
| 188 | except httpx.ConnectError as e: |
| 189 | raise Exception( |
| 190 | f"\nOllama connection failed at {host}. Check if the server is running." |
| 191 | ) from e |
| 192 | except Exception as e: |
| 193 | if hasattr(e, 'status_code') and e.status_code == 404: |
| 194 | animate_thinking(f"Downloading {self.model}...") |
| 195 | client.pull(self.model) |
| 196 | return self.ollama_fn(history, verbose) |
| 197 | if "refused" in str(e).lower(): |
| 198 | raise Exception( |
| 199 | f"Ollama connection refused at {host}. Is the server running?" |
| 200 | ) from e |
| 201 | raise e |
| 202 | |
| 203 | return thought |
| 204 | |
| 205 | def huggingface_fn(self, history, verbose=False): |
| 206 | """ |
nothing calls this directly
no test coverage detected