mirror of
https://github.com/containers/ramalama.git
synced 2026-02-05 15:47:26 +01:00
@@ -7,11 +7,6 @@ from pathlib import Path
|
||||
from ramalama.benchmarks.errors import MissingStorageFolderError
|
||||
from ramalama.benchmarks.schemas import BenchmarkRecord, DeviceInfoV1, get_benchmark_record
|
||||
from ramalama.benchmarks.utilities import parse_jsonl
|
||||
from ramalama.config import CONFIG
|
||||
from ramalama.log_levels import LogLevel
|
||||
|
||||
logger = logging.getLogger("ramalama.benchmarks")
|
||||
logger.setLevel(CONFIG.log_level or LogLevel.WARNING)
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
BENCHMARKS_FILENAME = "benchmarks.jsonl"
|
||||
|
||||
@@ -1,11 +1,9 @@
|
||||
import base64
|
||||
import os
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Literal, Protocol
|
||||
|
||||
from ramalama.config import get_config
|
||||
from ramalama.console import EMOJI, should_colorize
|
||||
from ramalama.console import should_colorize
|
||||
|
||||
RoleType = Literal["system", "user", "assistant", "tool"]
|
||||
|
||||
@@ -95,25 +93,6 @@ def stream_response(chunks: Iterable[bytes], color: str, provider: StreamParser)
|
||||
return assistant_response
|
||||
|
||||
|
||||
def default_prefix() -> str:
|
||||
config = get_config()
|
||||
|
||||
if not EMOJI:
|
||||
return "> "
|
||||
|
||||
if config.prefix:
|
||||
return config.prefix
|
||||
|
||||
if engine := config.engine:
|
||||
if os.path.basename(engine) == "podman":
|
||||
return "🦭 > "
|
||||
|
||||
if os.path.basename(engine) == "docker":
|
||||
return "🐋 > "
|
||||
|
||||
return "🦙 > "
|
||||
|
||||
|
||||
def serialize_part(part: AttachmentPart) -> dict[str, Any]:
|
||||
if isinstance(part, ImageURLPart):
|
||||
payload: dict[str, Any] = {"url": part.url}
|
||||
@@ -133,7 +112,6 @@ __all__ = [
|
||||
"ToolCall",
|
||||
"ImageURLPart",
|
||||
"ImageBytesPart",
|
||||
"default_prefix",
|
||||
"stream_response",
|
||||
"serialize_part",
|
||||
]
|
||||
|
||||
@@ -549,7 +549,8 @@ def bench_parser(subparsers):
|
||||
|
||||
|
||||
def benchmarks_parser(subparsers):
|
||||
storage_folder = CONFIG.benchmarks.storage_folder
|
||||
config = get_config()
|
||||
storage_folder = config.benchmarks.storage_folder
|
||||
epilog = f"Storage folder: {storage_folder}" if storage_folder else "Storage folder: not configured"
|
||||
parser = subparsers.add_parser(
|
||||
"benchmarks",
|
||||
@@ -584,8 +585,8 @@ def benchmarks_parser(subparsers):
|
||||
|
||||
def benchmarks_list_cli(args):
|
||||
"""Display a list of benchmark results from storage."""
|
||||
|
||||
bench_manager = BenchmarksManager(CONFIG.benchmarks.storage_folder)
|
||||
config = get_config()
|
||||
bench_manager = BenchmarksManager(config.benchmarks.storage_folder)
|
||||
results = bench_manager.list()
|
||||
|
||||
if not results:
|
||||
|
||||
@@ -513,8 +513,9 @@ class Transport(TransportBase):
|
||||
else:
|
||||
print_bench_results(results)
|
||||
|
||||
if not CONFIG.benchmarks.disable:
|
||||
bench_manager = BenchmarksManager(CONFIG.benchmarks.storage_folder)
|
||||
config = get_config()
|
||||
if not config.benchmarks.disable:
|
||||
bench_manager = BenchmarksManager(config.benchmarks.storage_folder)
|
||||
bench_manager.save(results)
|
||||
|
||||
def run(self, args, cmd: list[str]):
|
||||
|
||||
Reference in New Issue
Block a user