Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 12 additions & 26 deletions main.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from openai import OpenAI, OpenAIError

from utils import calculate_input_token_budget, get_required_config, read_input
from settings import load_settings
from utils import read_input


def trim_history(messages, keep_count):
Expand All @@ -18,29 +19,14 @@ def trim_history(messages, keep_count):

def main():
try:
api_key = get_required_config("API_KEY", cast=str)
base_url = get_required_config("BASE_URL", cast=str)
model_name = get_required_config("MODEL_NAME", cast=str)
show_usage = get_required_config("SHOW_USAGE", cast=bool)
max_tokens = get_required_config("MAX_TOKENS", cast=int)
temperature = get_required_config("TEMPERATURE", cast=float)
top_p = get_required_config("TOP_P", cast=float)
memory_keep_count = get_required_config("MEMORY_KEEP_COUNT", cast=int)
max_context_tokens = get_required_config(
"MAX_CONTEXT_TOKENS",
cast=int,
)
calculate_input_token_budget(max_context_tokens, max_tokens)

if memory_keep_count < 0:
raise ValueError("MEMORY_KEEP_COUNT must be 0 or greater")
settings = load_settings()
except ValueError as error:
print(f"Configuration error: {error}")
return 1

client = OpenAI(
base_url=base_url,
api_key=api_key,
base_url=settings.base_url,
api_key=settings.api_key,
)

while True:
Expand Down Expand Up @@ -84,13 +70,13 @@ def main():

try:
with client.chat.completions.create(
model=model_name,
model=settings.model_name,
messages=messages,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
max_tokens=settings.max_tokens,
temperature=settings.temperature,
top_p=settings.top_p,
stream=True,
stream_options={"include_usage": show_usage},
stream_options={"include_usage": settings.show_usage},
extra_body={
"reasoning": {
"effort": "low",
Expand Down Expand Up @@ -127,7 +113,7 @@ def main():
if finish_reason == "length":
print("Warning: the answer may be incomplete.")

if show_usage and usage:
if settings.show_usage and usage:
print("\n-----")
print("Prompt tokens:", usage.prompt_tokens)
print("Completion tokens:", usage.completion_tokens)
Expand All @@ -149,7 +135,7 @@ def main():
"content": answer,
}
)
messages = trim_history(messages, memory_keep_count)
messages = trim_history(messages, settings.memory_keep_count)


if __name__ == "__main__":
Expand Down
69 changes: 69 additions & 0 deletions settings.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
from dataclasses import dataclass, field

from decouple import config


def get_required_config(name, cast):
raw_value = config(name, default=None)

if raw_value is None or not str(raw_value).strip():
raise ValueError(f"{name} is missing or empty in .env")

try:
value = config(name, cast=cast)
except (TypeError, ValueError) as error:
raise ValueError(f"{name} has an invalid value: {error}") from error

return value.strip() if isinstance(value, str) else value


@dataclass(frozen=True)
class Settings:
api_key: str = field(repr=False)
base_url: str
model_name: str
show_usage: bool
max_tokens: int
max_context_tokens: int
temperature: float
top_p: float
memory_keep_count: int

def __post_init__(self):
if self.max_tokens <= 0:
raise ValueError("MAX_TOKENS must be greater than 0")

if self.max_context_tokens <= 0:
raise ValueError("MAX_CONTEXT_TOKENS must be greater than 0")

if self.max_tokens >= self.max_context_tokens:
raise ValueError(
"MAX_TOKENS must be smaller than MAX_CONTEXT_TOKENS"
)

if self.memory_keep_count < 0:
raise ValueError("MEMORY_KEEP_COUNT must be 0 or greater")

@property
def input_token_budget(self):
return self.max_context_tokens - self.max_tokens


def load_settings():
return Settings(
api_key=get_required_config("API_KEY", cast=str),
base_url=get_required_config("BASE_URL", cast=str),
model_name=get_required_config("MODEL_NAME", cast=str),
show_usage=get_required_config("SHOW_USAGE", cast=bool),
max_tokens=get_required_config("MAX_TOKENS", cast=int),
max_context_tokens=get_required_config(
"MAX_CONTEXT_TOKENS",
cast=int,
),
temperature=get_required_config("TEMPERATURE", cast=float),
top_p=get_required_config("TOP_P", cast=float),
memory_keep_count=get_required_config("MEMORY_KEEP_COUNT", cast=int),
)


__all__ = ["Settings", "get_required_config", "load_settings"]
50 changes: 50 additions & 0 deletions tests/test_settings.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
import unittest

from settings import Settings


class SettingsTests(unittest.TestCase):
def make_settings(self, **overrides):
values = {
"api_key": "secret",
"base_url": "https://example.com/v1",
"model_name": "example-model",
"show_usage": True,
"max_tokens": 1000,
"max_context_tokens": 8192,
"temperature": 0.2,
"top_p": 1.0,
"memory_keep_count": 10,
}
values.update(overrides)
return Settings(**values)

def test_calculates_input_token_budget(self):
settings = self.make_settings()

self.assertEqual(settings.input_token_budget, 7192)

def test_rejects_non_positive_output_limit(self):
with self.assertRaisesRegex(ValueError, "MAX_TOKENS"):
self.make_settings(max_tokens=0)

def test_rejects_non_positive_context_limit(self):
with self.assertRaisesRegex(ValueError, "MAX_CONTEXT_TOKENS"):
self.make_settings(max_context_tokens=0)

def test_rejects_output_limit_equal_to_context_limit(self):
with self.assertRaisesRegex(ValueError, "must be smaller"):
self.make_settings(max_tokens=8192)

def test_rejects_negative_memory_limit(self):
with self.assertRaisesRegex(ValueError, "must be 0 or greater"):
self.make_settings(memory_keep_count=-1)

def test_api_key_is_hidden_from_representation(self):
settings = self.make_settings()

self.assertNotIn("secret", repr(settings))


if __name__ == "__main__":
unittest.main()
33 changes: 1 addition & 32 deletions utils.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,3 @@
from decouple import config


def get_required_config(name, cast):
raw_value = config(name, default=None)

if raw_value is None or not str(raw_value).strip():
raise ValueError(f"{name} is missing or empty in .env")

try:
value = config(name, cast=cast)
except (TypeError, ValueError) as error:
raise ValueError(f"{name} has an invalid value: {error}") from error

return value.strip() if isinstance(value, str) else value


def read_input(prompt):
try:
return input(prompt).strip()
Expand All @@ -23,18 +6,4 @@ def read_input(prompt):
return None


def calculate_input_token_budget(context_limit, output_limit):
"""Calculate how many tokens are available for input messages."""
if context_limit <= 0:
raise ValueError("MAX_CONTEXT_TOKENS must be greater than 0")

if output_limit <= 0:
raise ValueError("MAX_TOKENS must be greater than 0")

if output_limit >= context_limit:
raise ValueError("MAX_TOKENS must be smaller than MAX_CONTEXT_TOKENS")

return context_limit - output_limit


__all__ = ["calculate_input_token_budget", "get_required_config", "read_input"]
__all__ = ["read_input"]