cutils-ai 0.2.2

通用工具包 - 基于 vutils 重构，新增 AI/LLM 便捷接口

cutils-ai

通用工具包 — 基于 vutils 重构，新增 AI/LLM 便捷接口 + 后端开发工具链。

安装

# 开发模式安装（推荐）
cd D:\csy_project\python\cutils
pip install -e .

# 或直接安装
pip install cutils-ai

# 可选：Redis 支持
pip install cutils-ai[redis]

模块概览

模块	功能
cutils.utils	通用工具 — 重试、限流、缓存、Schema 校验、哈希、环境变量、时间戳
cutils.ai	AI/Agent — OpenAI 客户端、Prompt 模板、Function Calling、对话管理、Token 预算、结构化输出、文本分块、向量存储
cutils.net	网络 — 代理设置、HTTP 客户端、Webhook 签名验证
cutils.io	数据IO — JSON/JSONL/TXT/CSV/Excel 读写、SQLite、Redis
cutils.log	可观测 — 日志、链路追踪、指标收集
cutils.dl	深度学习模型（通用分类/回归）
cutils.ml	机器学习（Relief 特征选择）
cutils.data	语言识别
cutils.mail	邮件发送
cutils.print_color	彩色终端输出
cutils.timer	代码计时器

---

📦 cutils.utils — 通用工具层

retry — 重试装饰器

from cutils.utils import retry

# 基本用法：失败后最多重试 3 次，指数退避
@retry(max_retries=3, delay=1.0, backoff=2.0)
def call_api():
    ...

# 只对特定异常重试
@retry(max_retries=5, exceptions=(ConnectionError, TimeoutError))
def call_flaky_api():
    ...

# 带回调通知
def on_retry(count, exc):
    print(f"第 {count} 次重试，原因: {exc}")

@retry(max_retries=3, on_retry=on_retry)
def call_api():
    ...

参数	类型	默认值	说明
max_retries	int	3	最大重试次数（不含首次调用）
delay	float	1.0	初始延迟秒数
backoff	float	2.0	退避倍数
max_delay	float	60.0	最大延迟秒数
exceptions	tuple	(Exception,)	触发重试的异常类型
on_retry	callable	None	每次重试回调 (retry_count, exception)

RateLimiter — 令牌桶限流器

from cutils.utils import RateLimiter

# 每秒 10 次请求，突发最多 10 个
limiter = RateLimiter(rate=10, capacity=10)

for i in range(100):
    limiter.acquire()  # 阻塞直到有令牌
    call_api()

# 非阻塞模式
if limiter.acquire(blocking=False):
    call_api()
else:
    print("限流中，跳过")

Cache — 函数缓存

from cutils.utils import Cache

# 基本缓存
cache = Cache(max_size=100, ttl=300)  # 最多 100 条，5 分钟过期
cache.set("key", {"data": [1, 2, 3]})
hit, value = cache.get("key")  # (True, {"data": [1, 2, 3]})

# 装饰器用法
cache = Cache(max_size=50, ttl=60)

@cache.decorator()
def get_user(user_id):
    return db.query(f"SELECT * FROM users WHERE id={user_id}")

user = get_user(123)  # 首次查询，缓存结果
user = get_user(123)  # 直接返回缓存

# 自定义 cache key
@cache.decorator(key_func=lambda user_id, **kw: f"user:{user_id}")
def get_user(user_id, verbose=False):
    return db.query(user_id)

validate_json — JSON Schema 校验

from cutils.utils import validate_json, SchemaError

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"},
        "tags": {"type": "array", "items": {"type": "string"}}
    },
    "required": ["name"]
}

# 校验通过
validate_json({"name": "Alice", "age": 30, "tags": ["dev"]}, schema)

# 校验失败抛 SchemaError
try:
    validate_json({"age": "not a number"}, schema)
except SchemaError as e:
    print(e)  # Schema 校验失败: 缺少必填字段: name

# 严格模式：不允许额外字段
validate_json({"name": "Bob", "extra": 1}, schema, strict=True)  # 抛错

md5 / sha256 / content_hash — 内容哈希

from cutils.utils import md5, sha256, content_hash

md5("hello")           # '5d41402abc4b2a76b9719d911017c592'
sha256("hello")        # '2cf24dba5fb0a30e26e83b2ac5b9e29e...'
content_hash("a", "b") # 组合哈希，适合缓存 key 生成

env — 环境变量加载器

from cutils.utils import env, env_int, env_float, env_bool, env_list, EnvError

# 字符串
api_key = env("OPENAI_API_KEY", required=True)  # 不存在抛 EnvError
base_url = env("BASE_URL", default="https://api.openai.com/v1")

# 类型转换
port = env_int("PORT", default=8080)
timeout = env_float("TIMEOUT", default=30.0)
debug = env_bool("DEBUG", default=False)       # "true"/"1"/"yes" -> True
hosts = env_list("ALLOWED_HOSTS", default=["localhost"])  # "a.com,b.com" -> ["a.com", "b.com"]

时间戳工具

from cutils.utils import now_iso, parse_iso, ts_millis, ts_seconds, format_duration

now_iso()                          # '2026-05-23T12:58:00+08:00'
now_iso(timezone(timedelta(hours=8)))  # 指定时区
parse_iso("2026-05-23T12:58:00+08:00")  # -> datetime 对象
ts_millis()                        # 1747999080000 (毫秒时间戳)
ts_seconds()                       # 1747999080.0 (秒时间戳)
format_duration(3661)              # '1h 1m 1s'
format_duration(0.5)               # '500ms'

---

🤖 cutils.ai — 智能体开发层

OpenAI 客户端（已有）

from cutils.ai import configure, chat, chat_stream, embedding, vision, OpenAIClient

# 全局配置
configure(api_key="sk-xxx", model="gpt-4o")

# 单轮对话
reply = chat("你好")
reply = chat("翻译成英文", system="你是专业翻译", temperature=0.3)

# 流式输出
for chunk in chat_stream("写一首诗"):
    print(chunk, end="", flush=True)

# Embedding
vec = embedding("hello world")  # -> List[float]
vecs = embedding(["hello", "world"])  # 批量

# 图片理解
desc = vision("https://example.com/cat.jpg", "这是什么动物？")

# 独立客户端
client = OpenAIClient(api_key="sk-yyy", base_url="https://custom.api.com")
reply = chat("你好", client=client)

PromptTemplate — Prompt 模板引擎

from cutils.ai import PromptTemplate, render_prompt

# 快捷函数
result = render_prompt("你是{role}，请用{lang}回答。", role="翻译助手", lang="中文")

# 模板类（可复用）
template = PromptTemplate("""
你是一个{role}。

{% if language %}请始终用{language}回复。{% endif %}

可用工具:
{% for tool in tools %}
- {tool.name}: {tool.description}
{% endfor %}
""")

result = template.render(
    role="翻译助手",
    language="中文",
    tools=[
        {"name": "search", "description": "搜索互联网"},
        {"name": " calculator", "description": "数学计算"},
    ]
)

模板语法:

• {variable} — 变量插值（支持点号 obj.attr）
• {% if condition %}...{% endif %} — 条件块
• {% for item in list %}...{% endfor %} — 循环块
• {# comment #} — 注释（渲染时移除）

tool — Function Calling 工具定义

from cutils.ai import tool, get_tool_schema, get_tools

@tool(description="获取指定城市的天气信息")
def get_weather(city: str, unit: str = "celsius") -> str:
    """获取天气。"""
    return f"{city} 晴天 25°"

@tool(description="搜索互联网信息")
def web_search(query: str, max_results: int = 5) -> str:
    """搜索。"""
    return f"搜索结果: {query}"

# 获取单个 schema
schema = get_tool_schema(get_weather)
# {
#   "type": "function",
#   "function": {
#     "name": "get_weather",
#     "description": "获取指定城市的天气信息",
#     "parameters": {
#       "type": "object",
#       "properties": {
#         "city": {"type": "string"},
#         "unit": {"type": "string", "default": "celsius"}
#       },
#       "required": ["city"]
#     }
#   }
# }

# 批量获取，直接传给 chat()
tools = get_tools(get_weather, web_search)
reply = chat("北京天气怎么样", tools=tools)

类型映射: str→string, int→integer, float→number, bool→boolean, list→array, dict→object

Conversation — 对话管理器

from cutils.ai import Conversation

# 创建对话
conv = Conversation(system="你是一个翻译助手", max_chars=8000)

# 添加消息
conv.add_user("你好")
conv.add_assistant("你好！有什么我可以帮你的吗？")
conv.add_user("翻译这句话成英文")

# 获取消息列表（含 system）
messages = conv.get_messages()
# [
#   {"role": "system", "content": "你是一个翻译助手"},
#   {"role": "user", "content": "你好"},
#   ...
# ]

# 直接传给 chat()
reply = chat(
    conv.get_messages()[-1]["content"],
    history=conv.get_messages()[:-1],
    system=conv.system,
)

# 获取最近 n 条历史
recent = conv.get_history(n=5)

# 统计信息
print(conv.summary())  # "对话统计: 2 条用户消息, 1 条回复, 共 3 条, 45 字符"

# 从已有消息列表恢复
conv = Conversation.from_messages(messages)

# 导出
data = conv.export()

自动裁剪: 当对话总字符数超过 max_chars 时，自动裁剪旧消息（至少保留 2 条）。

TokenBudget — Token 预算追踪

from cutils.ai import TokenBudget

budget = TokenBudget(model="gpt-4o")  # 自动识别 128k 限制
budget.add_system("你是一个翻译助手")  # ~10 tokens
budget.add_user("你好")               # ~5 tokens

print(budget.summary())
# "Token 预算: 使用 ~15/128000 tokens (0.01%), 剩余 ~127485 tokens"

# 检查是否超预算
if budget.is_within_budget(extra_tokens=2000):
    reply = chat("长文本...")

# 自定义限制
budget = TokenBudget(model="custom", max_tokens=32000, reserved=1000)

extract_json / parse_structured — 结构化输出解析

from cutils.ai import extract_json, parse_structured

# 从 LLM 输出提取 JSON
text = '根据分析，结果如下：\n```json\n{"score": 0.95, "label": "positive"}\n```'
data = extract_json(text)  # {"score": 0.95, "label": "positive"}

# 带 Schema 校验的解析
schema = {
    "type": "object",
    "properties": {
        "score": {"type": "number"},
        "label": {"type": "string"}
    },
    "required": ["score", "label"]
}
result = parse_structured(text, schema=schema)

# 校验失败时自动重试（需要传入重试函数）
def retry_func(error_msg):
    return chat(f"输出格式错误: {error_msg}\n请重新输出 JSON。")

result = parse_structured(
    bad_output,
    schema=schema,
    retry_func=retry_func,
    max_retries=2,
)

# 支持 Pydantic model（如果安装了 pydantic）
from pydantic import BaseModel
class AnalysisResult(BaseModel):
    score: float
    label: str

result = parse_structured(text, model_class=AnalysisResult)

chunker — 文本分块器（RAG 必备）

from cutils.ai import chunk_by_sentences, chunk_by_paragraphs, chunk_by_chars, chunk_by_tokens

# 按句子分块（中文友好）
text = "第一句话。第二句话。第三句话。第四句话。"
chunks = chunk_by_sentences(text, max_chunk_size=100, overlap=20)

# 按段落分块（双换行分割）
text = """段落一内容...

段落二内容...

段落三内容..."""
chunks = chunk_by_paragraphs(text, max_chunk_size=500, overlap=50)

# 按固定字符数分块
chunks = chunk_by_chars("a" * 1000, chunk_size=300, overlap=50)

# 按 token 数分块（需自定义 tokenizer）
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
chunks = chunk_by_tokens(
    long_text,
    max_tokens=500,
    overlap_tokens=50,
    tokenizer=enc.encode,
)

EmbeddingStore — 简易向量存储

from cutils.ai import EmbeddingStore
from cutils.ai import embedding as get_embedding

# 创建存储
store = EmbeddingStore(dimension=1536)

# 添加文档
texts = ["Python 是编程语言", "Java 是编程语言", "猫是动物"]
vectors = get_embedding(texts)  # 批量获取 embedding
store.add_batch(texts, vectors, metadata=[{"source": "wiki"}] * 3)

# 语义检索
query_vec = get_embedding("什么编程语言最流行")
results = store.search(query_vec, top_k=2)
for text, score, meta in results:
    print(f"{score:.4f}: {text}")

# 保存/加载
store.save("docs_store.json")
store = EmbeddingStore.load("docs_store.json")

---

🌐 cutils.net — 网络层

HttpClient — HTTP 客户端

from cutils.net import HttpClient

# 基本用法
client = HttpClient(
    base_url="https://api.example.com",
    timeout=30.0,
    max_retries=3,
    headers={"Authorization": "Bearer token"},
)

# GET / POST
resp = client.get("/users/1")
data = client.get_json("/users/1")

resp = client.post("/users", json={"name": "Alice"})

# 上下文管理器
with HttpClient("https://api.example.com") as client:
    data = client.get_json("/data")

# 带代理
client = HttpClient("https://api.example.com", proxy="http://127.0.0.1:7890")

Webhook 签名验证

from cutils.net import (
    verify_github_signature,
    verify_hmac_signature,
    verify_feishu_signature,
    verify_dingtalk_signature,
)

# GitHub
is_valid = verify_github_signature(
    payload=request.body,
    signature=request.headers["X-Hub-Signature-256"],
    secret="my_webhook_secret",
)

# 通用 HMAC
is_valid = verify_hmac_signature(
    payload=request.body,
    signature=request.headers["X-Signature"],
    secret="my_secret",
    algorithm="sha256",
)

# 飞书
is_valid = verify_feishu_signature(
    timestamp=request.headers["X-Lark-Request-Timestamp"],
    nonce=request.headers["X-Lark-Request-Nonce"],
    body=request.body.decode(),
    signature=request.headers["X-Lark-Signature"],
    encrypt_key="your_encrypt_key",
)

# 钉钉
is_valid = verify_dingtalk_signature(
    timestamp=timestamp,
    sign=sign,
    secret=secret,
)

代理设置（已有）

from cutils.net import set_proxy, unset_proxy

set_proxy(7890)   # 设置 HTTP 代理
unset_proxy()     # 清除代理

---

📂 cutils.io — 数据 IO

文件读写（已有）

from cutils.io import (
    jsonload, jsondump,       # JSON
    jsonlload, jsonldump,     # JSONL
    txtload, txtdump,         # TXT
    csvload, csvdump,         # CSV
    add_tail, get_extension,  # 路径工具
)

# JSON
data = jsonload("config.json")
jsondump({"key": "value"}, "output.json")

# JSONL（每行一条 JSON，适合大数据集）
data = jsonlload("data.jsonl")
jsonldump([{"id": 1}, {"id": 2}], "output.jsonl")

# 路径
add_tail("data.json", "_v2")  # "data_v2.json"
get_extension("data.json")    # "json"

SQLiteClient — SQLite 轻量封装

from cutils.io import SQLiteClient

# 上下文管理器用法
with SQLiteClient("app.db") as db:
    # 建表
    db.create_table("users", {
        "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
        "name": "TEXT NOT NULL",
        "age": "INTEGER DEFAULT 0",
    })

    # 插入
    db.insert("users", {"name": "Alice", "age": 30})
    db.insert("users", {"name": "Bob", "age": 25})

    # 查询
    user = db.query_one("SELECT * FROM users WHERE name = ?", ("Alice",))
    # {"id": 1, "name": "Alice", "age": 30}

    users = db.query_all("SELECT * FROM users WHERE age > ?", (20,))
    # [{"id": 1, ...}, {"id": 2, ...}]

    # 更新
    db.update("users", {"age": 31}, "name = ?", ("Alice",))

    # 删除
    db.delete("users", "name = ?", ("Bob",))

    # 检查表是否存在
    if db.table_exists("users"):
        ...

# 内存数据库（测试用）
with SQLiteClient(":memory:") as db:
    ...

RedisClient — Redis 封装

from cutils.io import RedisClient

# 基本用法
with RedisClient(host="localhost", port=6379) as r:
    # String
    r.set("key", "value", ex=3600)       # 设置，1 小时过期
    r.set("config", {"debug": True})      # dict 自动 JSON 序列化
    value = r.get("key")                  # -> "value"
    config = r.get_json("config")         # -> {"debug": True}
    r.delete("key")
    r.expire("key", 60)                   # 设置过期
    ttl = r.ttl("key")                    # 剩余秒数

    # Hash
    r.hset("user:1", "name", "Alice")
    name = r.hget("user:1", "name")
    all_fields = r.hgetall("user:1")

    # Pub/Sub
    r.publish("channel", {"event": "hello"})
    pubsub = r.subscribe("channel")
    for message in pubsub.listen():
        print(message)

    # Pipeline 批量操作
    pipe = r.pipeline()
    for i in range(100):
        pipe.set(f"item:{i}", i)
    pipe.execute()

ExcelHandler — Excel 读写（已有）

from cutils.io import ExcelHandler

# 从文件读取
handler = ExcelHandler("data.xlsx")
print(handler.getColumns())  # ["name", "age"]
print(handler.getRowSize())  # 100

# 创建空表格
handler = ExcelHandler(columns=["name", "age", "score"])
handler.setValue(0, "name", "Alice")
handler.setValue(0, "age", 30)

# 合并单元格填充
handler.fillNanFromMergeUnit("category")

# 保存
handler.saveAs("output.xlsx")
handler.saveAs("output.csv")

---

📊 cutils.log — 可观测性层

trace — 链路追踪

from cutils.log import trace_context, trace_log, push_span, pop_span, get_trace_id

# 上下文管理器：自动生成 trace_id
with trace_context() as ctx:
    print(ctx.trace_id)  # 'a1b2c3d4e5f6'

    # 标记子步骤
    push_span("llm_call")
    trace_log("开始调用 OpenAI API")
    # [trace:a1b2c3d4e5f6] [span:llm_call] 开始调用 OpenAI API

    # 嵌套 span
    push_span("retry_1")
    trace_log("重试中", attempt=1)
    pop_span()

    pop_span()

# 自定义 trace_id（便于跨服务关联）
with trace_context(trace_id="external-request-123") as ctx:
    trace_log("处理请求")

Metrics — 指标收集

from cutils.log import Metrics

metrics = Metrics()

# 手动记录
metrics.record("llm_call", time=1.5, tokens=200)
metrics.record("llm_call", time=0.8, tokens=150)
metrics.record_error("llm_call")

# 装饰器用法（自动追踪耗时和异常）
@metrics.track("api_call")
def call_api():
    ...

# 查看指标
m = metrics.get("llm_call")
print(f"调用次数: {m.count}")          # 2
print(f"平均耗时: {m.avg_time:.3f}s")  # 1.150
print(f"成功率: {m.success_rate:.1%}")  # 50.0%
print(f"Token 消耗: {m.tokens}")        # 350

# 生成报告
print(metrics.report())
# === Metrics Report ===
#   api_call: count=1, avg=0.000s, errors=0, rate=100.0%, tokens=0
#   llm_call: count=2, avg=1.150s, errors=1, rate=50.0%, tokens=350

logger — 日志（已有）

from cutils.log import logger

logger.info("应用启动")
logger.warning("配置缺失，使用默认值")
logger.error("API 调用失败")

---

环境变量

export OPENAI_API_KEY="sk-xxx"
export OPENAI_BASE_URL="https://api.openai.com/v1"  # 可选，自定义 endpoint
export CUTILS_MODEL="gpt-4o"                         # 可选，默认模型

CLI

ccli ping --name World
ccli web_data_view --file_path data.json
ccli compare_two_file --file_path1 a.txt --file_path2 b.txt

依赖

核心依赖: openai, httpx, pandas, numpy, matplotlib, tqdm, scikit-learn, torch, gradio, packaging

可选依赖: redis（Redis 客户端支持）