Spaces:
Runtime error
Runtime error
File size: 5,669 Bytes
00a4363 66de9c8 00a4363 0649f1b 66de9c8 1e09b5f db8dd62 1aa4a16 db8dd62 66de9c8 00a4363 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """Provider agnostic LLM factory. Any LiteLLM model string works; Z.ai GLM is the default.
Examples:
Z.ai: LLM_MODEL=openai/glm-5.3 LLM_API_BASE=https://api.z.ai/api/paas/v4
Anthropic: LLM_MODEL=anthropic/claude-sonnet-5 LLM_API_BASE=
OpenAI: LLM_MODEL=openai/gpt-5 LLM_API_BASE=
Local: LLM_MODEL=ollama_chat/qwen3 LLM_API_BASE=http://localhost:11434
"""
from contextlib import contextmanager
from dataclasses import dataclass
from functools import lru_cache
import dspy
from app.config import settings
def is_zai() -> bool:
return "z.ai" in settings.llm_api_base or "bigmodel.cn" in settings.llm_api_base
def _extra_kwargs(effort: str | None) -> dict:
if not is_zai():
return {}
# GLM-5.x always reasons; thinking.type only accepts "enabled". Cost and latency are tuned
# with reasoning_effort (low | high | max) instead. extra_body fields land top level in the request.
effort = effort or settings.llm_reasoning_effort
return {"extra_body": {"thinking": {"type": "enabled"}, "reasoning_effort": effort}}
def _build(model: str, *, effort: str | None, temperature: float | None, max_tokens: int | None) -> dspy.LM:
return dspy.LM(
model,
api_base=settings.llm_api_base or None,
api_key=settings.llm_api_key or None,
temperature=settings.llm_temperature if temperature is None else temperature,
max_tokens=max_tokens or settings.llm_max_tokens,
cache=True,
**_extra_kwargs(effort),
)
@lru_cache
def main_lm(effort: str | None = None) -> dspy.LM:
"""effort overrides LLM_REASONING_EFFORT per module, e.g. main_lm("high") for the groundedness judge."""
return _build(settings.llm_model, effort=effort, temperature=None, max_tokens=None)
@lru_cache
def fast_lm() -> dspy.LM:
return _build(settings.llm_fast_model, effort="low", temperature=None, max_tokens=settings.llm_fast_max_tokens)
REPORT_MAX_TOKENS = 16000 # a few pages of prose plus GLM's low-effort reasoning
@lru_cache
def report_lm() -> dspy.LM:
"""Writes reports: the fast model unless LLM_REPORT_MODEL names another, with room for a long answer."""
return _build(settings.llm_report_model or settings.llm_fast_model, effort="low", temperature=None,
max_tokens=REPORT_MAX_TOKENS)
REPORT_PLAN_MAX_TOKENS = 6000 # the blueprint: a storyline and up to eight short section plans plus reasoning
REPORT_SECTION_MAX_TOKENS = 4000 # one section (two to five paragraphs) plus reasoning, a quarter of a whole report's cap
def _report_model() -> str:
return settings.llm_report_model or settings.llm_fast_model
@lru_cache
def report_plan_lm() -> dspy.LM:
"""Plans a themed report (the blueprint). The report model, with a small cap: the answer is a plan, not prose."""
return _build(_report_model(), effort="low", temperature=None, max_tokens=REPORT_PLAN_MAX_TOKENS)
@lru_cache
def report_section_lm(retry: bool = False) -> dspy.LM:
"""Writes one section of a themed report. retry=True is the second try after a failed one: a different temperature,
so it is a fresh call and not the failed one again from the cache."""
return _build(_report_model(), effort="low", temperature=0.7 if retry else None, max_tokens=REPORT_SECTION_MAX_TOKENS)
@lru_cache
def triage_lm() -> dspy.LM:
"""Cheap model that decides whether a chat message needs research at all."""
if not settings.llm_triage_model:
return fast_lm()
return _build(settings.llm_triage_model, effort="low", temperature=0.2, max_tokens=2000)
FOCUS_MAX_TOKENS = 2000 # three titles and two-line descriptions plus GLM's low-effort reasoning
@lru_cache
def focus_lm(retry: bool = False) -> dspy.LM:
"""The video wizard's focus suggestions: a little material in, three short topics out, so a small cap keeps it
quick. Only for a source with no topics to show. retry=True is the second try after an unusable answer: a
different temperature, so it is a fresh call and not that answer again from the cache."""
return _build(settings.llm_fast_model, effort="low", temperature=0.8 if retry else 0.4,
max_tokens=FOCUS_MAX_TOKENS)
SUGGEST_TEMPLATES_MAX_TOKENS = 2000 # four short template ideas plus GLM's low-effort reasoning
@lru_cache
def suggest_templates_lm() -> dspy.LM:
"""The Create report dialog's suggested templates: four short template ideas out, so a tighter cap than the
general fast_lm keeps this quick without cutting into GLM's reasoning headroom."""
return _build(settings.llm_fast_model, effort="low", temperature=None, max_tokens=SUGGEST_TEMPLATES_MAX_TOKENS)
def configure_default() -> None:
dspy.configure(lm=main_lm(), adapter=dspy.JSONAdapter())
@dataclass
class Usage:
prompt_tokens: int = 0
completion_tokens: int = 0
cost_usd: float = 0.0
@property
def total(self) -> int:
return self.prompt_tokens + self.completion_tokens
@contextmanager
def track_usage(lm: dspy.LM):
"""Collect token usage for calls made inside the block (for usage_events metering)."""
usage = Usage()
start = len(lm.history)
try:
yield usage
finally:
for entry in lm.history[start:]:
u = entry.get("usage") or {}
usage.prompt_tokens += int(u.get("prompt_tokens") or 0)
usage.completion_tokens += int(u.get("completion_tokens") or 0)
usage.cost_usd += float(entry.get("cost") or 0.0)
def provider_name(model: str | None = None) -> str:
model = model or settings.llm_model
return "zai" if is_zai() and model.startswith("openai/") else model.split("/", 1)[0]
|