diff --git a/src/calibre/ai/github/__init__.py b/src/calibre/ai/github/__init__.py new file mode 100644 index 0000000000..0b56b5d3ae --- /dev/null +++ b/src/calibre/ai/github/__init__.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python +# License: GPLv3 Copyright: 2025, Kovid Goyal + +from calibre.customize import AIProviderPlugin + + +class GitHubAI(AIProviderPlugin): + name = 'GitHubAI' + version = (1, 0, 0) + description = _('AI services from GitHub, with access to many different AI models') + author = 'Kovid Goyal' + builtin_live_module_name = 'calibre.ai.github.backend' + + @property + def capabilities(self): + from calibre.ai import AICapabilities + return ( + AICapabilities.text_to_text | AICapabilities.text_to_image | AICapabilities.text_and_image_to_image | + AICapabilities.embedding + ) diff --git a/src/calibre/ai/github/backend.py b/src/calibre/ai/github/backend.py new file mode 100644 index 0000000000..bcfdce94ca --- /dev/null +++ b/src/calibre/ai/github/backend.py @@ -0,0 +1,263 @@ +#!/usr/bin/env python +# License: GPLv3 Copyright: 2025, Kovid Goyal + +import json +import os +from collections.abc import Iterable, Iterator +from functools import lru_cache +from typing import Any, NamedTuple +from urllib.request import Request + +from calibre.ai import AICapabilities, ChatMessage, ChatMessageType, ChatResponse, Citation, NoAPIKey, PromptBlocked, ResultBlocked, WebLink +from calibre.ai.github import GitHubAI +from calibre.ai.prefs import decode_secret, pref_for_provider +from calibre.ai.utils import chat_with_error_handler, develop_text_chat, get_cached_resource, read_streaming_response +from calibre.constants import cache_dir + +module_version = 1 # needed for live updates +MODELS_URL = 'https://models.github.ai/catalog/models' +API_VERSION = '2022-11-28' + + +def pref(key: str, defval: Any = None) -> Any: + return pref_for_provider(GitHubAI.name, key, defval) + + +def api_key() -> str: + return pref('api_key') + + +def is_ready_for_use() -> bool: + return bool(api_key()) + + +def decoded_api_key() -> str: + ans = api_key() + if not ans: + raise NoAPIKey('Personal access token required for GitHub AI') + return decode_secret(ans) + + +@lru_cache(2) +def headers() -> tuple[tuple[str, str]]: + api_key = decoded_api_key() + return ( + ('Authorization', f'Bearer {api_key}'), + ('X-GitHub-Api-Version', API_VERSION), + ('Accept', 'application/vnd.github+json'), + ('Content-Type', 'application/json'), + ) + + +class Model(NamedTuple): + # See https://ai.google.dev/api/models#Model + name: str + id: str + slug: str + description: str + version: str + context_length: int + output_token_limit: int + capabilities: AICapabilities + family: str + family_version: float + name_parts: tuple[str, ...] + thinking: bool + + @classmethod + def from_dict(cls, x: dict[str, object]) -> 'Model': + caps = AICapabilities.text_to_text + mid = x['name'] + if 'embedContent' in x['supportedGenerationMethods']: + caps |= AICapabilities.embedding + family, family_version = '', 0 + name_parts = mid.rpartition('/')[-1].split('-') + if len(name_parts) > 1: + family, fv = name_parts[:2] + try: + family_version = float(fv) + except Exception: + family = '' + match family: + case 'imagen': + caps |= AICapabilities.text_to_image + case 'gemini': + if family_version >= 2.5: + caps |= AICapabilities.text_and_image_to_image + if 'tts' in name_parts: + caps |= AICapabilities.tts + return Model( + name=x['displayName'], id=mid, description=x.get('description', ''), version=x['version'], + context_length=int(x['inputTokenLimit']), output_token_limit=int(x['outputTokenLimit']), + capabilities=caps, family=family, family_version=family_version, name_parts=tuple(name_parts), + slug=mid, thinking=x.get('thinking', False) + ) + + +def parse_models_list(entries: list[dict[str, Any]]) -> dict[str, Model]: + ans = {} + for entry in entries['models']: + e = Model.from_dict(entry) + ans[e.id] = e + return ans + + +@lru_cache(2) +def get_available_models() -> dict[str, 'Model']: + cache_loc = os.path.join(cache_dir(), 'github-ai', 'models-v1.json') + data = get_cached_resource(cache_loc, MODELS_URL, headers=headers()) + return parse_models_list(json.loads(data)) + + +def config_widget(): + from calibre.ai.github.config import ConfigWidget + return ConfigWidget() + + +def save_settings(config_widget): + config_widget.save_settings() + + +def human_readable_model_name(model_id: str) -> str: + if m := get_available_models().get(model_id): + model_id = m.name + return model_id + + +def model_choice_for_text() -> Model: + m = gemini_models() + return m.get(pref('model_strategy', 'medium')) or m['medium'] + + +def chat_request(data: dict[str, Any], model: Model, streaming: bool = True) -> Request: + url = f'{API_BASE_URL}/{model.slug}' + if streaming: + url += ':streamGenerateContent?alt=sse' + else: + url += ':generateContent' + return Request(url, data=json.dumps(data).encode('utf-8'), headers=dict(headers()), method='POST') + + +def thinking_budget(m: Model) -> int | None: + # https://ai.google.dev/gemini-api/docs/thinking#set-budget + if not m.thinking: + return None + limits = 0, 24576 + if 'pro' in m.name_parts: + limits = 128, 32768 + elif 'lite' in m.name_parts: + limits = 512, 24576 + match pref('reasoning_strategy', 'auto'): + case 'auto': + return -1 + case 'none': + return limits[0] if 'pro' in m.name_parts else 0 + case 'low': + return max(limits[0], int(0.2 * limits[1])) + case 'medium': + return max(limits[0], int(0.5 * limits[1])) + case 'high': + return max(limits[0], int(0.8 * limits[1])) + return None + + +def for_assistant(self: ChatMessage) -> dict[str, Any]: + return {'text': self.query} + + +def as_chat_responses(d: dict[str, Any], model: Model) -> Iterator[ChatResponse]: + # See https://ai.google.dev/api/generate-content#generatecontentresponse + if pf := d.get('promptFeedback'): + if br := pf.get('blockReason'): + yield ChatResponse(exception=PromptBlocked(block_reason(br))) + return + grounding_chunks, grounding_supports = [], [] + for c in d['candidates']: + has_metadata = False + cost, currency = 0, '' + if fr := c.get('finishReason'): + if fr == 'STOP': + has_metadata = True + cost, currency = model.get_cost(d['usageMetadata']) + else: + yield ChatResponse(exception=ResultBlocked(result_block_reason(fr))) + return + content = c['content'] + if gm := c.get('groundingMetadata'): + grounding_chunks.extend(gm['groundingChunks']) + grounding_supports.extend(gm['groundingSupports']) + citations, web_links = [], [] + if has_metadata: + for x in grounding_chunks: + if w := x.get('web'): + web_links.append(WebLink(**w)) + else: + web_links.append(WebLink()) + + for s in grounding_supports: + if links := tuple(i for i in s['groundingChunkIndices'] if web_links[i]): + seg = s['segment'] + citations.append(Citation( + links, start_offset=seg.get('startIndex', 0), end_offset=seg.get('endIndex', 0), text=seg.get('text', ''))) + role = ChatMessageType.user if 'user' == content.get('role') else ChatMessageType.assistant + content_parts = [] + reasoning_parts = [] + reasoning_details = [] + for part in content['parts']: + if text := part.get('text'): + (reasoning_parts if part.get('thought') else content_parts).append(text) + if ts := part.get('thoughtSignature'): + reasoning_details.append({'signature': ts}) + yield ChatResponse( + type=role, content=''.join(content_parts), reasoning=''.join(reasoning_parts), + reasoning_details=tuple(reasoning_details), has_metadata=has_metadata, model=model.id, + cost=cost, plugin_name=GitHubAI.name, currency=currency, citations=citations, web_links=web_links, + ) + + +def text_chat_implementation(messages: Iterable[ChatMessage], use_model: str = '') -> Iterator[ChatResponse]: + # See https://ai.google.dev/gemini-api/docs/text-generation + if use_model: + model = get_available_models()[use_model] + else: + model = model_choice_for_text() + contents = [] + system_instructions = [] + for m in messages: + d = system_instructions if m.type is ChatMessageType.system else contents + d.append(for_assistant(m)) + data = { + # See https://ai.google.dev/api/generate-content#v1beta.GenerationConfig + 'generationConfig': { + 'thinkingConfig': { + 'includeThoughts': True, + }, + }, + } + if (tb := thinking_budget(model)) is not None: + data['generationConfig']['thinkingConfig']['thinkingBudget'] = tb + if system_instructions: + data['system_instruction'] = {'parts': system_instructions} + if contents: + data['contents'] = [{'parts': contents}] + if pref('allow_web_searches', True): + data['tools'] = [{'google_search': {}}] + rq = chat_request(data, model) + + for datum in read_streaming_response(rq, GitHubAI.name): + yield from as_chat_responses(datum, model) + + +def text_chat(messages: Iterable[ChatMessage], use_model: str = '') -> Iterator[ChatResponse]: + yield from chat_with_error_handler(text_chat_implementation(messages, use_model)) + + +def develop(use_model: str = '', msg: str = '') -> None: + # calibre-debug -c 'from calibre.ai.github.backend import develop; develop()' + print('\n'.join(f'{k}:{m.id}' for k, m in gemini_models().items())) + m = (ChatMessage(msg),) if msg else () + develop_text_chat(text_chat, ('models/' + use_model) if use_model else '', messages=m) + + +if __name__ == '__main__': + develop() diff --git a/src/calibre/ai/github/config.py b/src/calibre/ai/github/config.py new file mode 100644 index 0000000000..b5827b5fcb --- /dev/null +++ b/src/calibre/ai/github/config.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python +# License: GPLv3 Copyright: 2025, Kovid Goyal + + +from functools import partial + +from qt.core import QComboBox, QFormLayout, QLabel, QLineEdit, QWidget + +from calibre.ai.github import GitHubAI +from calibre.ai.prefs import decode_secret, encode_secret, pref_for_provider, set_prefs_for_provider +from calibre.ai.utils import configure +from calibre.gui2 import error_dialog + +pref = partial(pref_for_provider, GitHubAI.name) + + +class ConfigWidget(QWidget): + + def __init__(self, parent: QWidget | None = None): + super().__init__(parent) + l = QFormLayout(self) + l.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.ExpandingFieldsGrow) + la = QLabel('

'+_( + 'You have to create an account at {0}, then generate a Personal access token' + ' with the models:read permission.' + ' After that, you can use the GitHub AI services a limited number of times a day for free.' + ' For more extensive use, you will need to setup GitHub models billing.' + ).format( + 'GitHub', + 'https://docs.github.com/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens', + 'https://docs.github.com/billing/concepts/product-billing/github-models', + )) + la.setWordWrap(True) + la.setOpenExternalLinks(True) + l.addRow(la) + + self.api_key_edit = a = QLineEdit(self) + a.setPlaceholderText(_('A personal access token is required')) + l.addRow(_('Access &token:'), a) + if key := pref('api_key'): + a.setText(decode_secret(key)) + self.model_strategy = ms = QComboBox(self) + l.addRow(_('Model &choice strategy:'), ms) + ms.addItem(_('Cheap and fastest'), 'low') + ms.addItem(_('Medium'), 'medium') + ms.addItem(_('High quality, expensive and slower'), 'high') + if strat := pref('model_choice_strategy', 'medium'): + ms.setCurrentIndex(max(0, ms.findData(strat))) + ms.setToolTip('

' + _( + 'The model choice strategy controls how a model to query is chosen. Cheaper and faster models give lower' + ' quality results.' + )) + + @property + def api_key(self) -> str: + return self.api_key_edit.text().strip() + + @property + def model_choice_strategy(self) -> str: + return self.model_strategy.currentData() + + @property + def settings(self) -> dict[str, str]: + return { + 'api_key': encode_secret(self.api_key), 'model_choice_strategy': self.model_choice_strategy, + } + + @property + def is_ready_for_use(self) -> bool: + return bool(self.api_key) + + def validate(self) -> bool: + if self.is_ready_for_use: + return True + error_dialog(self, _('No API key'), _('You must supply a Personal access token to use GitHub AI.'), show=True) + return False + + def save_settings(self): + set_prefs_for_provider(GitHubAI.name, self.settings) + + +if __name__ == '__main__': + configure(GitHubAI.name) diff --git a/src/calibre/ai/google/config.py b/src/calibre/ai/google/config.py index ecc38e312e..c2d91a5bf5 100644 --- a/src/calibre/ai/google/config.py +++ b/src/calibre/ai/google/config.py @@ -8,6 +8,7 @@ from qt.core import QCheckBox, QComboBox, QFormLayout, QLabel, QLineEdit, QWidge from calibre.ai.google import GoogleAI from calibre.ai.prefs import decode_secret, encode_secret, pref_for_provider, set_prefs_for_provider +from calibre.ai.utils import configure from calibre.gui2 import error_dialog pref = partial(pref_for_provider, GoogleAI.name) @@ -100,3 +101,7 @@ class ConfigWidget(QWidget): def save_settings(self): set_prefs_for_provider(GoogleAI.name, self.settings) + + +if __name__ == '__main__': + configure(GoogleAI.name) diff --git a/src/calibre/ai/open_router/config.py b/src/calibre/ai/open_router/config.py index 43603717bd..d6619e84b0 100644 --- a/src/calibre/ai/open_router/config.py +++ b/src/calibre/ai/open_router/config.py @@ -36,9 +36,10 @@ from qt.core import ( from calibre.ai import AICapabilities from calibre.ai.open_router import OpenRouterAI from calibre.ai.prefs import decode_secret, encode_secret, pref_for_provider, set_prefs_for_provider +from calibre.ai.utils import configure from calibre.customize.ui import available_ai_provider_plugins from calibre.ebooks.txt.processor import create_markdown_object -from calibre.gui2 import Application, error_dialog, gprefs, safe_open_url +from calibre.gui2 import error_dialog, gprefs, safe_open_url from calibre.gui2.widgets2 import Dialog from calibre.utils.date import qt_from_dt from calibre.utils.icu import primary_sort_key @@ -478,7 +479,4 @@ class ConfigWidget(QWidget): if __name__ == '__main__': - app = Application([]) - d = ChooseModel() - d.exec() - print(d.model_id) + configure(OpenRouterAI.name) diff --git a/src/calibre/ai/utils.py b/src/calibre/ai/utils.py index c8ad8831e3..3898fa0712 100644 --- a/src/calibre/ai/utils.py +++ b/src/calibre/ai/utils.py @@ -19,6 +19,7 @@ from urllib.request import ProxyHandler, Request, build_opener from calibre import get_proxies from calibre.ai import ChatMessage, ChatMessageType, ChatResponse, Citation, WebLink from calibre.constants import __version__ +from calibre.customize.ui import available_ai_provider_plugins def atomic_write(path, data): @@ -285,6 +286,35 @@ def develop_text_chat( pprint(msg) +def configure(plugin_name: str, parent: Any = None) -> None: + from qt.core import QDialog, QDialogButtonBox, QVBoxLayout + + from calibre.gui2 import ensure_app + ensure_app(headless=False) + for plugin in available_ai_provider_plugins(): + if plugin.name == plugin_name: + cw = plugin.config_widget() + break + else: + raise KeyError(f'No plugin named: {plugin_name}') + class D(QDialog): + def accept(self): + if not cw.validate(): + return + super().accept() + + d = D(parent=parent) + l = QVBoxLayout(d) + l.addWidget(cw) + bb = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel) + bb.accepted.connect(d.accept) + bb.rejected.connect(d.reject) + l.addWidget(bb) + d.resize(d.sizeHint()) + if d.exec() == QDialog.DialogCode.Accepted: + plugin.save_settings(cw) + + def find_tests() -> None: import unittest class TestAIUtils(unittest.TestCase): diff --git a/src/calibre/customize/builtins.py b/src/calibre/customize/builtins.py index bc268a6d09..c6cd8d732a 100644 --- a/src/calibre/customize/builtins.py +++ b/src/calibre/customize/builtins.py @@ -4,6 +4,7 @@ __copyright__ = '2008, Kovid Goyal ' import glob import os +from calibre.ai.github import GitHubAI from calibre.ai.google import GoogleAI from calibre.ai.open_router import OpenRouterAI from calibre.constants import numeric_version @@ -1978,7 +1979,7 @@ plugins += [ # }}} -plugins.extend((OpenRouterAI, GoogleAI)) +plugins.extend((OpenRouterAI, GoogleAI, GitHubAI)) if __name__ == '__main__': # Test load speed diff --git a/src/calibre/gui2/__init__.py b/src/calibre/gui2/__init__.py index c039c4d7a9..8388563533 100644 --- a/src/calibre/gui2/__init__.py +++ b/src/calibre/gui2/__init__.py @@ -1621,6 +1621,10 @@ def ensure_app(headless=True): with _ea_lock: if _store_app is None and QApplication.instance() is None: args = sys.argv[:1] + if not headless: + _store_app = Application([]) + sys.excepthook = simple_excepthook + return has_headless = ismacos or islinux or isbsd if headless and has_headless: args += ['-platformpluginpath', plugins_loc, '-platform', os.environ.get('CALIBRE_HEADLESS_PLATFORM', 'headless')]