feat: allow per-module log level overrides

README.md

@@ -132,6 +132,30 @@ Run `kimi` with `--mcp-config-file` option to connect to the specified MCP serve
 kimi --mcp-config-file /path/to/mcp.json
 ```
 
+### Logging configuration
+
+Kimi CLI writes logs to `~/.kimi/logs/kimi.log`. By default the log level is `INFO` (or `TRACE` when `--debug` is passed). Use `-L/--log-level` to override a specific module or the default level:
+
+```sh
+# Increase verbosity for tools while keeping the rest quiet
+kimi -L kimi_cli.tools=DEBUG -L WARNING
+```
+
+You can also persist the mapping in `~/.kimi/config.json`:
+
+```json
+{
+  "logging": {
+    "levels": {
+      "default": "INFO",
+      "kosong": "WARNING"
+    }
+  }
+}
+```
+
+Module names act as prefixes and are case-insensitive, so `KIMI_CLI.Tools` works the same as `kimi_cli.tools` for all nested modules.
+
 ## Development
 
 To develop Kimi CLI, run:

src/kimi_cli/app.py

@@ -11,7 +11,7 @@
 
 from kimi_cli.agentspec import DEFAULT_AGENT_FILE
 from kimi_cli.cli import InputFormat, OutputFormat
-from kimi_cli.config import LLMModel, LLMProvider, load_config
+from kimi_cli.config import Config, LLMModel, LLMProvider, load_config
 from kimi_cli.llm import augment_provider_with_env_vars, create_llm
 from kimi_cli.session import Session
 from kimi_cli.soul import LLMNotSet, LLMNotSupported
@@ -31,6 +31,7 @@ async def create(
         stream: bool = True,  # TODO: remove this when we have a correct print mode impl
         mcp_configs: list[dict[str, Any]] | None = None,
         config_file: Path | None = None,
+        config: Config | None = None,
         model_name: str | None = None,
         thinking: bool = False,
         agent_file: Path | None = None,
@@ -43,6 +44,7 @@ async def create(
             yolo (bool, optional): Approve all actions without confirmation. Defaults to False.
             stream (bool, optional): Use stream mode when calling LLM API. Defaults to True.
             config_file (Path | None, optional): Path to the configuration file. Defaults to None.
+            config (Config | None, optional): Preloaded configuration. Defaults to None.
             model_name (str | None, optional): Name of the model to use. Defaults to None.
             agent_file (Path | None, optional): Path to the agent file. Defaults to None.
 
@@ -51,7 +53,7 @@ async def create(
             ConfigError(KimiCLIException): When the configuration is invalid.
             AgentSpecError(KimiCLIException): When the agent specification is invalid.
         """
-        config = load_config(config_file)
+        config = config or load_config(config_file)
         logger.info("Loaded config: {config}", config=config)
 
         model: LLMModel | None = None

src/kimi_cli/cli.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import sys
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from pathlib import Path
 from typing import Annotated, Any, Literal
 
@@ -28,6 +28,9 @@ class Reload(Exception):
 InputFormat = Literal["text", "stream-json"]
 OutputFormat = Literal["text", "stream-json"]
 
+_LOG_LEVEL_OPTION = "--log-level"
+_DEFAULT_LOG_LEVEL_KEY = "default"
+
 
 def _version_callback(value: bool) -> None:
     if value:
@@ -61,6 +64,17 @@ def kimi(
             help="Log debug information. Default: no.",
         ),
     ] = False,
+    log_level_override: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--log-level",
+            "-L",
+            help=(
+                "Override log level per module. Use `module=LEVEL` (case-insensitive) or just "
+                "`LEVEL` to change the default level."
+            ),
+        ),
+    ] = None,
     agent_file: Annotated[
         Path | None,
         typer.Option(
@@ -198,9 +212,12 @@ def kimi(
     del version  # handled in the callback
 
     from kimi_cli.app import KimiCLI
+    from kimi_cli.config import load_config
     from kimi_cli.session import Session
     from kimi_cli.share import get_share_dir
-    from kimi_cli.utils.logging import logger
+    from kimi_cli.utils.logging import configure_file_logging, logger
+
+    config = load_config()
 
     def _noop_echo(*args: Any, **kwargs: Any):
         pass
@@ -229,13 +246,18 @@ def _noop_echo(*args: Any, **kwargs: Any):
 
     if debug:
         logger.enable("kosong")
-    logger.add(
-        get_share_dir() / "logs" / "kimi.log",
-        # FIXME: configure level for different modules
-        level="TRACE" if debug else "INFO",
-        rotation="06:00",
-        retention="10 days",
-    )
+    config_levels = dict(config.logging.levels)
+    cli_levels = _parse_log_level_overrides(log_level_override or [])
+    merged_levels = {**config_levels, **cli_levels}
+    base_level = "TRACE" if debug else "INFO"
+    try:
+        configure_file_logging(
+            get_share_dir() / "logs" / "kimi.log",
+            base_level=base_level,
+            module_levels=merged_levels,
+        )
+    except ValueError as exc:
+        raise typer.BadParameter(str(exc), param_hint=_LOG_LEVEL_OPTION) from exc
 
     work_dir = (work_dir or Path.cwd()).absolute()
     if continue_:
@@ -289,6 +311,7 @@ async def _run() -> bool:
             model_name=model_name,
             thinking=thinking,
             agent_file=agent_file,
+            config=config,
         )
         match ui:
             case "shell":
@@ -319,5 +342,46 @@ async def _run() -> bool:
             continue
 
 
+def _parse_log_level_overrides(values: Iterable[str]) -> dict[str, str]:
+    overrides: dict[str, str] = {}
+    for raw in values:
+        entry = raw.strip()
+        if not entry:
+            raise typer.BadParameter(
+                "Log level override cannot be empty",
+                param_hint=_LOG_LEVEL_OPTION,
+            )
+        if "=" in entry:
+            module, level = entry.split("=", 1)
+            module = module.strip()
+            if not module:
+                raise typer.BadParameter(
+                    "Module name is required before '=' when using --log-level",
+                    param_hint=_LOG_LEVEL_OPTION,
+                )
+        else:
+            module = _DEFAULT_LOG_LEVEL_KEY
+            level = entry
+        level = level.strip()
+        if not level:
+            raise typer.BadParameter("Log level cannot be empty", param_hint=_LOG_LEVEL_OPTION)
+        overrides[_normalize_module_key(module)] = level
+    return overrides
+
+
+def _normalize_module_key(module: str) -> str:
+    cleaned = module.strip().rstrip(".")
+    normalized = cleaned.lower()
+    if not normalized:
+        return _DEFAULT_LOG_LEVEL_KEY
+    if normalized == _DEFAULT_LOG_LEVEL_KEY:
+        return _DEFAULT_LOG_LEVEL_KEY
+    return normalized
+
+
+def main() -> None:
+    cli()
+
+
 if __name__ == "__main__":
     cli()

src/kimi_cli/config.py

@@ -73,6 +73,36 @@ class Services(BaseModel):
     """Moonshot Search configuration."""
 
 
+class LoggingConfig(BaseModel):
+    """Logging configuration."""
+
+    levels: dict[str, str] = Field(
+        default_factory=dict,
+        description="Mapping of module prefixes to log levels.",
+    )
+
+    @model_validator(mode="after")
+    def validate_levels(self) -> LoggingConfig:
+        normalized: dict[str, str] = {}
+        for module, level in self.levels.items():
+            module_name = module.strip()
+            if not module_name:
+                raise ValueError("Module name in logging.levels cannot be empty")
+            level_name = level.strip().upper()
+            try:
+                logger.level(level_name)
+            except ValueError as exc:  # pragma: no cover - loguru raises ValueError
+                raise ValueError(
+                    f"Invalid log level '{level_name}' for module '{module_name}'"
+                ) from exc
+            key = module_name.rstrip(".").lower()
+            if not key:
+                key = "default"
+            normalized[key] = level_name
+        self.levels = normalized
+        return self
+
+
 class Config(BaseModel):
     """Main configuration structure."""
 
@@ -83,6 +113,10 @@ class Config(BaseModel):
     )
     loop_control: LoopControl = Field(default_factory=LoopControl, description="Agent loop control")
     services: Services = Field(default_factory=Services, description="Services configuration")
+    logging: LoggingConfig = Field(
+        default_factory=LoggingConfig,
+        description="Logging configuration",
+    )
 
     @model_validator(mode="after")
     def validate_model(self) -> Self:
@@ -106,6 +140,7 @@ def get_default_config() -> Config:
         models={},
         providers={},
         services=Services(),
+        logging=LoggingConfig(),
     )
 
 

src/kimi_cli/utils/logging.py

@@ -1,12 +1,102 @@
 from __future__ import annotations
 
-from typing import IO
+from collections.abc import Mapping
+from pathlib import Path
+from typing import IO, TYPE_CHECKING, Any
 
 from loguru import logger
 
+if TYPE_CHECKING:
+    from loguru import Record
+else:  # pragma: no cover - runtime fallback for typing-only import
+    Record = dict[str, Any]  # type: ignore[assignment]
+
+MODULE_ROOTS = ("kimi_cli", "kosong")
+DEFAULT_LEVEL_KEY = "default"
+
 logger.remove()
 
 
+def configure_file_logging(
+    log_file: Path,
+    *,
+    base_level: str,
+    module_levels: Mapping[str, str] | None = None,
+    rotation: str = "06:00",
+    retention: str = "10 days",
+) -> None:
+    """Configure the global loguru logger with per-module filtering."""
+
+    logger.remove()
+    log_file.parent.mkdir(parents=True, exist_ok=True)
+    normalized_levels = _normalize_levels(module_levels or {}, base_level)
+    module_filter = _ModuleLevelFilter(normalized_levels)
+    logger.add(
+        log_file,
+        level="TRACE",  # capture everything, filter decides what to keep
+        rotation=rotation,
+        retention=retention,
+        filter=module_filter,
+    )
+    logger.debug("Configured log levels: {levels}", levels=normalized_levels)
+
+
+def _normalize_levels(levels: Mapping[str, str], base_level: str) -> dict[str, int]:
+    normalized: dict[str, int] = {}
+    for module, level_name in levels.items():
+        key = module.strip().rstrip(".").lower() or DEFAULT_LEVEL_KEY
+        normalized[key] = _level_to_no(level_name)
+    if DEFAULT_LEVEL_KEY not in normalized:
+        normalized[DEFAULT_LEVEL_KEY] = _level_to_no(base_level)
+    return normalized
+
+
+def _level_to_no(level_name: str) -> int:
+    normalized = level_name.strip().upper()
+    try:
+        return logger.level(normalized).no
+    except ValueError as exc:  # pragma: no cover - loguru raises ValueError
+        raise ValueError(f"Invalid log level '{level_name}'") from exc
+
+
+class _ModuleLevelFilter:
+    """Filter that enforces module-specific log levels."""
+
+    def __init__(self, levels: Mapping[str, int]) -> None:
+        self._levels = dict(levels)
+        self._module_keys = sorted(
+            (key for key in self._levels if key != DEFAULT_LEVEL_KEY),
+            key=len,
+            reverse=True,
+        )
+
+    def __call__(self, record: Record) -> bool:
+        module_path = self._derive_module_path(record)
+        threshold = self._resolve_threshold(module_path)
+        return record["level"].no >= threshold
+
+    def _resolve_threshold(self, module_path: str | None) -> int:
+        if module_path:
+            for key in self._module_keys:
+                if module_path == key or module_path.startswith(f"{key}."):
+                    return self._levels[key]
+        return self._levels[DEFAULT_LEVEL_KEY]
+
+    @staticmethod
+    def _derive_module_path(record: Record) -> str | None:
+        file_info = record.get("file")
+        path_str = getattr(file_info, "path", None)
+        if not path_str:
+            return None
+        path = Path(path_str)
+        module_parts = path.with_suffix("").parts
+        for idx, part in enumerate(module_parts):
+            if part.lower() in MODULE_ROOTS:
+                return ".".join(module_parts[idx:]).lower()
+        module_name = record.get("module")
+        return module_name.lower() if module_name else None
+
+
 class StreamToLogger(IO[str]):
     def __init__(self, level: str = "ERROR"):
         self._level = level

tests/test_config.py

@@ -4,6 +4,7 @@
 
 from kimi_cli.config import (
     Config,
+    LoggingConfig,
     Services,
     get_default_config,
 )
@@ -17,6 +18,7 @@ def test_default_config():
             models={},
             providers={},
             services=Services(),
+            logging=LoggingConfig(levels={}),
         )
     )
 
@@ -33,7 +35,10 @@ def test_default_config_dump():
     "max_steps_per_run": 100,
     "max_retries_per_step": 3
   },
-  "services": {}
+  "services": {},
+  "logging": {
+    "levels": {}
+  }
 }\
 """
     )