content_pipeline / core /config.py
AI Engineer
Deploy Streamlit Content Agent
0b29030
Raw
History Blame Contribute Delete
3.29 kB
"""Runtime configuration for the content generation agent.
All settings can come from environment variables (for Docker / server deploys)
or be overridden at runtime from the Streamlit UI (so a user can paste their
own API key and choose a model before running a command).
"""
from __future__ import annotations
import os
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
from dataclasses import dataclass, field, asdict
from typing import Optional
# OpenAI-compatible chat-completion endpoints for the supported providers.
PROVIDER_ENDPOINTS = {
"groq": "https://api.groq.com/openai/v1/chat/completions",
"openrouter": "https://openrouter.ai/api/v1/chat/completions",
}
# Sensible default models per provider.
DEFAULT_MODELS = {
"groq": "llama-3.3-70b-versatile",
"openrouter": "meta-llama/llama-3.3-70b-instruct",
}
# A short menu the UI can present per provider.
MODEL_CHOICES = {
"groq": [
"llama-3.3-70b-versatile",
"llama-3.1-8b-instant",
"openai/gpt-oss-20b",
],
"openrouter": [
"meta-llama/llama-3.3-70b-instruct",
"openai/gpt-4o-mini",
"anthropic/claude-3.5-sonnet",
"google/gemini-flash-1.5",
],
}
def _env_bool(name: str, default: bool) -> bool:
val = os.getenv(name)
if val is None:
return default
return val.strip().lower() in {"1", "true", "yes", "on"}
@dataclass
class Settings:
"""Resolved settings used by the pipeline."""
provider: str = field(default_factory=lambda: os.getenv("LLM_PROVIDER", "groq"))
api_key: Optional[str] = field(default_factory=lambda: os.getenv("LLM_API_KEY"))
model: Optional[str] = field(default_factory=lambda: os.getenv("LLM_MODEL"))
# Mock mode lets the whole pipeline run with no network / no API key.
mock_mode: bool = field(default_factory=lambda: _env_bool("MOCK_MODE", False))
temperature: float = field(default_factory=lambda: float(os.getenv("LLM_TEMPERATURE", "0.7")))
max_tokens: int = field(default_factory=lambda: int(os.getenv("LLM_MAX_TOKENS", "1200")))
# Crawl controls
crawl_max_pages: int = field(default_factory=lambda: int(os.getenv("CRAWL_MAX_PAGES", "40")))
crawl_max_depth: int = field(default_factory=lambda: int(os.getenv("CRAWL_MAX_DEPTH", "2")))
crawl_timeout: int = field(default_factory=lambda: int(os.getenv("CRAWL_TIMEOUT", "10")))
def resolved_model(self) -> str:
if self.model:
return self.model
return DEFAULT_MODELS.get(self.provider, DEFAULT_MODELS["groq"])
def endpoint(self) -> str:
return PROVIDER_ENDPOINTS.get(self.provider, PROVIDER_ENDPOINTS["groq"])
def is_live(self) -> bool:
"""True when we should make real API calls."""
return (not self.mock_mode) and bool(self.api_key)
def to_dict(self) -> dict:
d = asdict(self)
# Never echo the full key back.
if d.get("api_key"):
d["api_key"] = d["api_key"][:4] + "…"
return d
def get_settings(**overrides) -> Settings:
"""Build settings from env, then apply any UI overrides (non-None only)."""
s = Settings()
for k, v in overrides.items():
if v is not None and hasattr(s, k):
setattr(s, k, v)
return s