zhaoxi
8f1398c591
新增persona_template表和CRUD API，BaseIndividualModel增加node_affinity和template_origin_id字段， WorkerCluster支持多集群Ray资源调度，环境变量收敛到pydantic-settings统一校验， 数据库异常转换为结构化BusinessError/RetryableError，系统节点支持custom_system_prompt。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
# Copyright 2026 zhaoxi826
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import sys
|
|
|
|
from loguru import logger
|
|
from rich.logging import RichHandler
|
|
from loguru._logger import Logger
|
|
|
|
from kilostar.utils.request_context import get_request_id, get_trace_id
|
|
|
|
|
|
def _is_json_mode() -> bool:
|
|
from kilostar.utils.settings import get_settings
|
|
s = get_settings().log
|
|
if s.kilostar_log_format.lower() == "json":
|
|
return True
|
|
return s.kilostar_log_json.lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _ctx_patcher(record):
|
|
"""日志切面：每条日志写出前，把 contextvars 里的 request_id / trace_id 注入。
|
|
|
|
显式 ``bind(trace_id=...)`` 的 logger 优先（业务代码可以覆盖切面值）；
|
|
没有 bind 时回退到 contextvars，没有 contextvars 时为空串。
|
|
"""
|
|
extra = record["extra"]
|
|
if not extra.get("trace_id"):
|
|
extra["trace_id"] = get_trace_id()
|
|
if not extra.get("request_id"):
|
|
extra["request_id"] = get_request_id()
|
|
|
|
|
|
def setup_logger() -> Logger:
|
|
"""初始化全局 loguru logger。
|
|
|
|
- 默认（开发模式）：``RichHandler`` 彩色输出，格式 ``actor:(...) | request_id:(...) | trace_id:(...) : message``
|
|
- JSON 模式（``KILOSTAR_LOG_FORMAT=json``）：写到 stdout，每行一条 JSON，便于 ELK/Loki 采集
|
|
|
|
request_id / trace_id 来自 ``kilostar.utils.request_context``，由 FastAPI middleware
|
|
或工作流入口绑定到 contextvars，本模块通过 ``patcher`` 透明注入。
|
|
"""
|
|
logger.remove()
|
|
|
|
from kilostar.utils.settings import get_settings
|
|
log_level = get_settings().log.kilostar_log_level.upper()
|
|
|
|
if _is_json_mode():
|
|
logger.configure(
|
|
extra={"actor_name": "System", "trace_id": "", "request_id": ""},
|
|
patcher=_ctx_patcher,
|
|
)
|
|
logger.add(
|
|
sys.stdout,
|
|
serialize=True,
|
|
level=log_level,
|
|
enqueue=True,
|
|
)
|
|
return logger
|
|
|
|
def format_record(record):
|
|
actor = record["extra"].get("actor_name", "System")
|
|
trace_id = record["extra"].get("trace_id", "")
|
|
request_id = record["extra"].get("request_id", "")
|
|
ids = []
|
|
if request_id:
|
|
ids.append(f"request_id:({request_id})")
|
|
if trace_id:
|
|
ids.append(f"trace_id:({trace_id})")
|
|
ids_str = " | " + " | ".join(ids) if ids else ""
|
|
return f"actor:({actor}){ids_str} : {record['message']}"
|
|
|
|
logger.configure(
|
|
extra={"actor_name": "System", "trace_id": "", "request_id": ""},
|
|
patcher=_ctx_patcher,
|
|
)
|
|
|
|
logger.add(
|
|
RichHandler(
|
|
rich_tracebacks=True,
|
|
markup=True,
|
|
show_time=False,
|
|
show_level=False,
|
|
show_path=False,
|
|
),
|
|
format=format_record,
|
|
level=log_level,
|
|
enqueue=True,
|
|
)
|
|
|
|
return logger
|
|
|
|
|
|
global_logger = setup_logger()
|
|
|
|
|
|
def get_logger(actor_name: str, trace_id: str = "") -> Logger:
|
|
"""获取一个绑定了 actor_name 与可选 trace_id 的 logger，便于日志按 Actor/请求归类。
|
|
|
|
若 ``trace_id`` 留空，会回退到 ``contextvars`` 中的当前值（由 middleware 或
|
|
工作流入口设置）。显式传值则会覆盖切面注入。
|
|
"""
|
|
return global_logger.bind(actor_name=actor_name, trace_id=trace_id)
|