Load Codex model options dynamically

This commit is contained in:
2026-07-15 20:16:46 +02:00
parent 02481c3de6
commit 1c9fefc089
4 changed files with 145 additions and 28 deletions

View File

@@ -17,7 +17,8 @@ OPENAI_API_KEY=
# Optional: set a default model. Leave empty to use the Codex runtime default.
CODEX_DEFAULT_MODEL=
# Optional: model buttons in /model. Format: "Label=model,Label=model"
# Optional: fallback model buttons for /model if the live Codex model list is unavailable.
# Format: "Label=model,Label=model"
CODEX_MODEL_CHOICES=GPT-5.5=gpt-5.5,GPT-5.4=gpt-5.4,GPT-5.4 mini=gpt-5.4-mini
# Optional: full-access, workspace-write, or read-only. Server-admin mode defaults to full-access.

View File

@@ -45,7 +45,7 @@ journalctl -u codex-tg-bot -f
| `WORK_DIR` | Рабочая директория Codex, по умолчанию `/` |
| `OPENAI_API_KEY` | Опционально; если пусто, SDK переиспользует существующий Codex login |
| `CODEX_DEFAULT_MODEL` | Опциональная модель по умолчанию |
| `CODEX_MODEL_CHOICES` | Кнопки `/model`, формат `Label=model,Label=model` |
| `CODEX_MODEL_CHOICES` | Fallback-кнопки `/model`, если live-список Codex недоступен; формат `Label=model,Label=model` |
| `CODEX_SANDBOX` | `full-access`, `workspace-write` или `read-only`; по умолчанию `full-access` |
| `CODEX_APPROVAL_MODE` | `auto_review` или `deny_all`; по умолчанию `auto_review` |
| `CODEX_BIN` | Опциональный путь к конкретному `codex` executable |

View File

@@ -33,6 +33,47 @@ class DoneEvent:
usage: ThreadTokenUsage | None
@dataclass(frozen=True)
class ModelChoice:
label: str
value: str
efforts: tuple[str, ...]
is_default: bool = False
@dataclass(frozen=True)
class ModelOptions:
models: tuple[ModelChoice, ...]
fallback: bool = False
error: str | None = None
def by_value(self, value: str | None) -> ModelChoice | None:
if value is None:
return None
return next((model for model in self.models if model.value == value), None)
def default_model(self) -> ModelChoice | None:
if config.CODEX_DEFAULT_MODEL:
configured = self.by_value(config.CODEX_DEFAULT_MODEL)
if configured is not None:
return configured
return next((model for model in self.models if model.is_default), None)
def efforts_for(self, model_value: str | None) -> tuple[str, ...]:
model = self.by_value(model_value) or self.default_model()
if model is not None and model.efforts:
return model.efforts
seen: set[str] = set()
efforts: list[str] = []
for choice in self.models:
for effort in choice.efforts:
if effort not in seen:
seen.add(effort)
efforts.append(effort)
return tuple(efforts)
def _shorten(text: str, limit: int = TOOL_INPUT_PREVIEW_LEN) -> str:
text = " ".join(text.split())
if len(text) > limit:
@@ -167,6 +208,54 @@ def _codex_config() -> CodexConfig:
return CodexConfig(codex_bin=config.CODEX_BIN)
def _fallback_model_options(error: str | None = None) -> ModelOptions:
all_efforts = tuple(effort.value for effort in ReasoningEffort)
models = tuple(
ModelChoice(label=label, value=value, efforts=all_efforts)
for label, value in config.MODEL_CHOICES
)
return ModelOptions(models=models, fallback=True, error=error)
def _reasoning_efforts(raw_options: Any) -> tuple[str, ...]:
efforts: list[str] = []
for option in raw_options or []:
value = getattr(getattr(option, "reasoning_effort", None), "value", None)
if value and value not in efforts:
efforts.append(value)
return tuple(efforts)
async def list_model_options() -> ModelOptions:
"""Fetch visible Codex models and their supported reasoning efforts."""
try:
async with AsyncCodex(config=_codex_config()) as codex:
if config.OPENAI_API_KEY:
await codex.login_api_key(config.OPENAI_API_KEY)
response = await codex.models()
except Exception as exc:
return _fallback_model_options(error=str(exc))
models: list[ModelChoice] = []
for model in response.data:
if getattr(model, "hidden", False):
continue
models.append(
ModelChoice(
label=getattr(model, "display_name", None) or model.id,
value=model.model,
efforts=_reasoning_efforts(model.supported_reasoning_efforts),
is_default=bool(getattr(model, "is_default", False)),
)
)
if not models:
return _fallback_model_options(error="Codex returned no visible models")
return ModelOptions(models=tuple(models))
def _thread_kwargs(model: str | None, effort: str | None) -> dict[str, Any]:
selected_model = model or config.CODEX_DEFAULT_MODEL
thread_config = {"model_reasoning_effort": effort} if effort else None

View File

@@ -37,19 +37,14 @@ running_tasks: dict[int, asyncio.Task] = {}
admin = filters.user(config.ADMIN_IDS)
MODELS: list[tuple[str, str | None]] = [
*config.MODEL_CHOICES,
("По умолчанию", None),
]
EFFORTS: list[tuple[str, str | None]] = [
("minimal", "minimal"),
("low", "low"),
("medium", "medium"),
("high", "high"),
("xhigh", "xhigh"),
("По умолчанию", None),
]
REASONING_ORDER = {
"none": 0,
"minimal": 1,
"low": 2,
"medium": 3,
"high": 4,
"xhigh": 5,
}
HELP_TEXT = (
"**Codex Server Admin Bot**\n\n"
@@ -67,33 +62,56 @@ HELP_TEXT = (
)
def _model_keyboard(chat_id: int) -> InlineKeyboardMarkup:
async def _model_options() -> codex_runner.ModelOptions:
return await codex_runner.list_model_options()
def _model_keyboard(
chat_id: int,
options: codex_runner.ModelOptions,
) -> InlineKeyboardMarkup:
current_model = chats.get(chat_id, "model")
current_effort = chats.get(chat_id, "effort")
def label(name: str, value: str | None, current: str | None) -> str:
return f"* {name}" if value == current else name
models: list[tuple[str, str | None]] = [
(choice.label, choice.value) for choice in options.models
]
if current_model and options.by_value(current_model) is None:
models.insert(0, (f"Сохраненная: {current_model}", current_model))
models.append(("По умолчанию", None))
efforts = sorted(
options.efforts_for(current_model),
key=lambda effort: REASONING_ORDER.get(effort, 99),
)
effort_choices: list[tuple[str, str | None]] = [
(effort, effort) for effort in efforts
]
effort_choices.append(("По умолчанию", None))
rows = [[InlineKeyboardButton("- Модель -", callback_data="noop")]]
for i in range(0, len(MODELS), 2):
for i in range(0, len(models), 2):
rows.append(
[
InlineKeyboardButton(
label(name, value, current_model),
callback_data=f"model:{value or ''}",
)
for name, value in MODELS[i : i + 2]
for name, value in models[i : i + 2]
]
)
rows.append([InlineKeyboardButton("- Reasoning effort -", callback_data="noop")])
for i in range(0, len(EFFORTS), 3):
for i in range(0, len(effort_choices), 3):
rows.append(
[
InlineKeyboardButton(
label(name, value, current_effort),
callback_data=f"effort:{value or ''}",
)
for name, value in EFFORTS[i : i + 3]
for name, value in effort_choices[i : i + 3]
]
)
return InlineKeyboardMarkup(rows)
@@ -101,8 +119,8 @@ def _model_keyboard(chat_id: int) -> InlineKeyboardMarkup:
MODEL_TEXT = (
"**Настройки Codex**\n\n"
"Выберите модель и reasoning effort для задач в этом чате.\n"
"\"По умолчанию\" использует настройки Codex runtime или CODEX_DEFAULT_MODEL."
"Выберите модель и reasoning effort для задач в этом чате. "
"Список моделей и уровней reasoning загружается из Codex."
)
@@ -154,10 +172,15 @@ async def cmd_compact(_, message: Message):
@app.on_message(filters.command("model") & admin)
async def cmd_model(_, message: Message):
await message.reply_text(
MODEL_TEXT,
loading = await message.reply_text("Загружаю список моделей Codex...")
options = await _model_options()
text = MODEL_TEXT
if options.fallback:
text += "\n\nНе удалось получить live-список, показываю fallback из env."
await loading.edit_text(
text,
parse_mode=ParseMode.MARKDOWN,
reply_markup=_model_keyboard(message.chat.id),
reply_markup=_model_keyboard(message.chat.id, options),
)
@@ -171,13 +194,17 @@ async def on_settings_callback(_, query: CallbackQuery):
if key not in ("model", "effort"):
await query.answer()
return
await query.answer("Обновляю...")
value = raw or None
chat_id = query.message.chat.id
options = await _model_options()
chats.set(chat_id, key, value)
shown = value or "по умолчанию"
await query.answer(f"{'Модель' if key == 'model' else 'Reasoning'}: {shown}")
if key == "model":
current_effort = chats.get(chat_id, "effort")
if current_effort and current_effort not in options.efforts_for(value):
chats.set(chat_id, "effort", None)
try:
await query.message.edit_reply_markup(_model_keyboard(chat_id))
await query.message.edit_reply_markup(_model_keyboard(chat_id, options))
except RPCError:
pass