feat/uploader thread split

Makefile

@@ -1,4 +1,4 @@
-.PHONY:  init sync format proto unit bench
+.PHONY:  init sync format proto unit bench clean
 
 init:
 	uv sync --all-extras
@@ -18,4 +18,7 @@ unit:
 	uv run pytest tests/unit
 
 bench:
-	uv run pytest tests/benchmark
+	uv run pytest tests/benchmark
+
+clean:
+	@bash scripts/clean_pycache.sh .

scripts/clean_pycache.sh

@@ -0,0 +1,39 @@
+#!/bin/bash
+# Description: 递归清除指定目录下的 __pycache__ 和 .pyc 文件
+# 如果删除 __pycache__ 后导致其父目录为空，则连同由于删除产生的空父目录一起递归向上删除
+
+# 目标目录，默认当前目录
+TARGET_DIR="${1:-.}"
+
+echo "开始清理 $TARGET_DIR 目录下的 Python 缓存文件..."
+
+# 1. 寻找所有的 __pycache__ 文件夹，并循环处理
+# 注意：find 结果通过 while 读取，处理带有空格等特殊符号的路径
+find "$TARGET_DIR" -type d -name "__pycache__" | while read -r cache_dir; do
+    # 获取 __pycache__ 所在的父目录
+    parent_dir="$(dirname "$cache_dir")"
+
+    # 删除该 __pycache__ 目录及里面的缓存文件
+    rm -rf "$cache_dir"
+
+    echo "已清理: $cache_dir"
+
+    # 尝试递归往上删除空的父目录
+    # rmdir 的特性是：只有在此目录没有任何文件/文件夹时才会删除并返回成功
+    # 因此我们循环往上 rmdir，如果哪个父层级除了 __pycache__ 还有其它文件，自然会失败并跳出循环
+    while [ "$parent_dir" != "." ] && [ "$parent_dir" != "/" ] && [ "$parent_dir" != "$TARGET_DIR" ]; do
+        if rmdir "$parent_dir" 2>/dev/null; then
+            echo "已删除原本仅含 __pycache__ 的空父目录: $parent_dir"
+            # 指向更上层父目录继续检查
+            parent_dir="$(dirname "$parent_dir")"
+        else
+            # 遇到了非空目录（包含了其它文件），终止向上删除
+            break
+        fi
+    done
+done
+
+# 2. 清理散落在其它地方的独立 .pyc 后缀缓存文件
+find "$TARGET_DIR" -type f -name "*.pyc" -delete 2>/dev/null
+
+echo "清理完成！"

swanlab/sdk/internal/core.py

@@ -6,6 +6,7 @@
 """
 
 from abc import ABC, abstractmethod
+from enum import Enum
 from typing import List
 
 from swanlab.proto.swanlab.record.v1.record_pb2 import Record
@@ -14,6 +15,11 @@
 __all__ = ["CoreProtocol"]
 
 
+class CoreEnum(str, Enum):
+    CORE_PYTHON = "CorePython"
+    CORE = "Core"
+
+
 class CoreProtocol(ABC):
     """
     SwanLab Core 的抽象基类。
@@ -47,3 +53,13 @@ def handle_records(self, records: List[Record]) -> None:
     def shutdown(self) -> None:
         """刷盘、触发收尾回调、释放所有资源。调用后此实例不应再被使用。"""
         ...
+
+
+def create_core(ctx: RunContext, core_enum: CoreEnum = CoreEnum.CORE_PYTHON) -> CoreProtocol:
+    if core_enum == CoreEnum.CORE_PYTHON:
+        from swanlab.sdk.internal.core_python import CorePython
+
+        return CorePython(ctx)
+    else:
+        # TODO: Core 微服务无感接入
+        raise NotImplementedError

swanlab/sdk/internal/core_python/__init__.py

@@ -2,6 +2,7 @@
 @author: cunyue
 @file: __init__.py
 @time: 2026/3/13
+
 @description: SwanLab Core Python 版本，封装SwanLab云端版核心业务，包括：
 1. 提供http客户端，用于与SwanLab云端API进行交互。
 2. 提供rpc封装函数，以rpc方式调用SwanLab云端API。
@@ -13,12 +14,13 @@
 BackgroundConsumer 等调用方无需修改。
 """
 
-from typing import List
+from typing import List, Optional
 
 from swanlab.proto.swanlab.record.v1.record_pb2 import Record
 from swanlab.sdk.internal.context import RunContext
 from swanlab.sdk.internal.core import CoreProtocol
 from swanlab.sdk.internal.core_python.store import DataStoreWriter
+from swanlab.sdk.internal.core_python.uploader import Uploader
 from swanlab.sdk.internal.pkg import console
 from swanlab.sdk.utils.helper.env import DEBUG
 
@@ -33,26 +35,34 @@ class CorePython(CoreProtocol):
 
     def __init__(self, ctx: RunContext):
         super().__init__(ctx)
-        self._store: DataStoreWriter | None = None
+        self._store: Optional[DataStoreWriter] = None
+        self._uploader: Optional[Uploader] = None
 
     def startup(self, cloud: bool, persistence: bool) -> None:
-        if self._store is not None:
+        if self._store is not None or self._uploader is not None:
             raise RuntimeError("CorePython has already been started.")
         if persistence:
             self._store = DataStoreWriter()
             self._store.open(str(self._ctx.run_file))
+        if cloud:
+            self._uploader = Uploader()
 
     def handle_records(self, records: List[Record]) -> None:
-        if self._store is None:
+        if self._store is None and self._uploader is None:
             console.warning("CorePython is not started, skipping record handling.")
             return
-        for record in records:
-            self._store.write(record.SerializeToString())
-            if DEBUG:
-                console.debug("Write record:", record.WhichOneof("record_type"))
+        if self._store is not None:
+            for record in records:
+                self._store.write(record.SerializeToString())
+                if DEBUG:
+                    console.debug("Write record:", record.WhichOneof("record_type"))
+        if self._uploader is not None:
+            self._uploader.put(records)
 
     def shutdown(self) -> None:
-        if self._store is None:
-            return
-        self._store.close()
-        self._store = None
+        if self._uploader is not None:
+            self._uploader.finish()
+            self._uploader = None
+        if self._store is not None:
+            self._store.close()
+            self._store = None

swanlab/sdk/internal/core_python/uploader/__init__.py

@@ -0,0 +1,24 @@
+"""
+@author: caddiesnew
+@file: __init__.py
+@time: 2026/3/19
+@description: 指标上传线程，基于 protobuf Record
+"""
+
+from .collector import Collector
+from .sender import (
+    HttpRecordTransport,
+    create_record_transport,
+    trace_records,
+    upload_records,
+)
+from .thread import Uploader
+
+__all__ = [
+    "Collector",
+    "HttpRecordTransport",
+    "Uploader",
+    "create_record_transport",
+    "trace_records",
+    "upload_records",
+]

swanlab/sdk/internal/core_python/uploader/collector.py

@@ -0,0 +1,84 @@
+"""
+@author: caddiesnew
+@file: collector.py
+@time: 2026/3/21
+@description: 上传线程采集器
+"""
+
+import threading
+import time
+from typing import Callable, List, Optional
+
+from swanlab.proto.swanlab.record.v1.record_pb2 import Record
+from swanlab.sdk.internal.core_python.uploader.sender import upload_records
+from swanlab.sdk.internal.pkg import console
+
+
+class Collector:
+    """
+    日志采集器，负责从队列中收集 Records 并批量上传。
+    """
+
+    def __init__(
+        self,
+        upload_interval: float = 1.0,
+        upload_callback: Optional[Callable] = None,
+    ):
+        self.container: List[Record] = []
+        self._lock = threading.Lock()
+        self._upload_interval = upload_interval
+        self._upload_callback = upload_callback
+        self._last_upload_at = time.time()
+
+    def upload(self, pending: List[Record]) -> None:
+        """
+        核心上传逻辑：直接批量上送 protobuf Record。
+        由于网络请求耗时，应在无锁状态下调用。
+        """
+        if not pending:
+            return
+        upload_records(pending, upload_callback=self._upload_callback)
+
+    def submit(self, records: List[Record]) -> None:
+        """提交记录，由定时线程以 upload_interval 为周期进行调用"""
+        with self._lock:
+            self.container.extend(records)
+            if not self.container:
+                return
+
+            pending = self.container
+            if self._upload_interval <= 0:
+                self.container = []
+            else:
+                now = time.time()
+                if now - self._last_upload_at <= self._upload_interval:
+                    return
+
+                self._last_upload_at = now
+                self.container = []
+
+        try:
+            self.upload(pending)
+        except Exception as exc:
+            console.error(f"upload error: {exc}")
+            with self._lock:
+                self.container = pending + self.container
+
+    def flush(self, records: List[Record]) -> None:
+        """刷新所有累积的 records 进行上传"""
+        with self._lock:
+            self.container.extend(records)
+            pending = self.container
+            self.container = []
+
+        try:
+            self.upload(pending)
+        except Exception as exc:
+            console.error(f"upload error: {exc}")
+            with self._lock:
+                self.container = pending + self.container
+
+
+__all__ = [
+    "Collector",
+]

swanlab/sdk/internal/core_python/uploader/helper.py

@@ -0,0 +1,47 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+r"""
+@author: caddiesnew
+@file: helper.py
+@time: 2026/3/30 15:33
+@description: uploader 辅助函数
+"""
+
+from collections import OrderedDict
+from typing import Dict, Iterator, List, Sequence, Tuple
+
+from swanlab.proto.swanlab.record.v1.record_pb2 import Record
+
+
+def generate_chunks(records: Sequence[Record], per_request_len: int) -> Iterator[Tuple[Sequence[Record], int]]:
+    """
+    按固定大小对 Record 序列做切片。
+    yield: (分片数据, 分片长度)
+    """
+    if per_request_len == -1:
+        yield records, len(records)
+        return
+    if per_request_len <= 0:
+        raise ValueError(f"per_request_len must be -1 or > 0, got {per_request_len}")
+
+    total = len(records)
+    if total <= per_request_len:
+        yield records, total
+        return
+
+    for index in range(0, total, per_request_len):
+        chunk = records[index : index + per_request_len]
+        yield chunk, len(chunk)
+
+
+def group_records_by_type(records: Sequence[Record]) -> Dict[str, List[Record]]:
+    """按照 record 的类型进行分组"""
+    grouped: Dict[str, List[Record]] = OrderedDict()
+    for record in records:
+        if not isinstance(record, Record):
+            raise TypeError(f"group_records_by_type only accepts Record instances, got {type(record).__name__}")
+        record_type = record.WhichOneof("record_type")
+        if record_type is None:
+            raise ValueError("Record has no active record_type and cannot be uploaded.")
+        grouped.setdefault(record_type, []).append(record)
+    return grouped

swanlab/sdk/internal/core_python/uploader/sender.py

@@ -0,0 +1,78 @@
+"""
+@author: caddiesnew
+@file: sender.py
+@time: 2026/3/19
+@description: Record 上传入口与 Core sidecar transport
+"""
+
+from typing import Callable, Optional, Sequence
+
+from swanlab.proto.swanlab.record.v1.record_pb2 import Record
+from swanlab.sdk.internal.pkg import console
+
+from .helper import generate_chunks, group_records_by_type
+
+
+class HttpRecordTransport:
+    """PLACEHOLDER: 基于现有 HTTP client 的上传 transport 。"""
+
+    def upload_record_group(self, record_type: str, records: Sequence[Record]) -> None:
+        if len(records) == 0:
+            return
+        console.debug(f"HTTP upload skeleton is ready for record_type={record_type!r}, but request mapping is pending.")
+
+    def close(self) -> None:
+        pass
+
+
+def create_record_transport() -> HttpRecordTransport:
+    """创建 Record transport。"""
+    return HttpRecordTransport()
+
+
+def trace_records(
+    records: Optional[Sequence[Record]] = None,
+    per_request_len: int = 10_000,
+    upload_callback: Optional[Callable[[int], None]] = None,
+    transport: Optional[HttpRecordTransport] = None,
+) -> None:
+    """
+    分片上传 protobuf Record。
+
+    线程层负责缓冲、聚合与重试；此层只做 Record 分片、按 record_type 分组和 transport 调用。
+    """
+    if records is None or len(records) == 0:
+        return
+
+    owns_transport = transport is None
+    active_transport = transport or create_record_transport()
+
+    try:
+        for chunk, chunk_len in generate_chunks(records, per_request_len):
+            for record_type, grouped_records in group_records_by_type(chunk).items():
+                active_transport.upload_record_group(record_type, grouped_records)
+            if upload_callback:
+                upload_callback(chunk_len)
+    finally:
+        if owns_transport:
+            active_transport.close()
+
+
+def upload_records(
+    records: Sequence[Record],
+    upload_callback: Optional[Callable[[int], None]] = None,
+    per_request_len: int = 10_000,
+) -> None:
+    """上传一组 protobuf Record，不在此层做任何 JSON 化。"""
+    if len(records) == 0:
+        return console.debug("No records to upload.")
+    trace_records(records, per_request_len=per_request_len, upload_callback=upload_callback)
+    return None
+
+
+__all__ = [
+    "HttpRecordTransport",
+    "create_record_transport",
+    "trace_records",
+    "upload_records",
+]