API

The real signature, contract, and one-liner for every corespine export, grouped by seam — verified against inspect.signature and live runs.

corespine.__all__ 每个导出的【真实签名】+ 契约 + 一句话,按缝分节。所有签名经 inspect.signature 核对、所有行为经实跑核对(见 recipes)。心智模型见 overview,易错点见 gotchas。

全部从顶层包导入:from corespine import Registry, MockProvider, ...。 corespine.__version__ == "0.2.0"。__all__ 共 43 项(含 __version__)。

seam —— `corespine.seam.registry`

`class Registry(Generic[T])`

一条缝的 name -> factory 注册表 + spec 解析 + entry-point 自动发现。

Registry(seam: str) -> None          # seam 名既用于报错,也用于 entry-point group "corespine.<seam>"

# 属性(只读 property)
registry.seam   -> str               # 缝名(构造时传入)
registry.group  -> str               # entry-point 发现使用的 group:f"corespine.{seam}"

# 方法
register(name: str, factory: Factory[T]) -> None
    # 登记一个内置实现;名字归一后入表(同名后者覆盖)。factory 签名:(**kwargs) -> T。

make(spec: str, **kwargs: Any) -> T
    # 把 spec 解析到工厂并构造实例。解析顺序:内置注册 -> entry-point 发现 -> 抛 ValueError。
    # spec 归一:去首尾留白 + 转小写 + 连字符/空格统一为下划线("In-Process"=="in_process")。
    # kwargs 原样透传给工厂。未知 spec 抛 ValueError 并列清【当前全部可用名】。

names() -> list[str]
    # 当前全部可用名:内置 + entry-point 发现,去重后按字典序。

契约要点:Factory[T] = Callable[..., T](任意 kwargs → 实例)。内置注册优先于 entry-point (同名时内置胜出,便于测试覆盖)。entry-point 发现是延迟的——只在 make / names 时才扫。

`lazy_extra_import(module: str, *, pkg: str, extra: str) -> Any`

延迟 import 一个可选依赖;缺失时把裸 ImportError 翻译成下面这样的友好提示:

缺少可选依赖 '<module>':请先 `pip install <pkg>[<extra>]` 再重试。

支撑「离线精简默认 + 可选重依赖」范式——核心默认路径零重依赖,真实后端 SDK 仅在选用时才 import。返回成功 import 的模块对象。

observability —— `corespine.observability.trace`

`class TraceSink(Protocol)`(`@runtime_checkable`)

trace 出口的最小结构接口。

emit(self, code: str, **fields: object) -> None    # 发射一条事件(code + 非敏感字段)

`class TraceExporter(Protocol)`(`@runtime_checkable`,0.2.0)

trace 导出面的最小结构接口:接收一条已通过隐私校验的事件,扇出到进程外。

export(self, event: TraceEvent) -> None            # 接收一条已过 FORBIDDEN_KEYS 校验的 TraceEvent

隐私契约(不可放松):exporter 只会收到已在 sink 的 emit 里【先过 FORBIDDEN_KEYS 校验、再扇出】的 TraceEvent——受限正文在校验阶段就被 TraceError 挡下,根本到不了任何 exporter。故导出面天然只承载 code / 计数 / 耗时,与本地记录面等宽。真实导出后端(opentelemetry / langfuse SDK)由各 app 在自己的适配器里经可选 extra 延迟 import,绝不进核心。

`class InProcessTraceExporter`(0.2.0)

进程内默认 TraceExporter:把扇出到的事件收集进内存列表(离线 / 测试用,零依赖)。

InProcessTraceExporter() -> None                    # 无参构造

export(self, event: TraceEvent) -> None             # 追加到内部列表(只可能收到已过隐私校验的事件)

# 属性(只读 property)
events -> list[TraceEvent]                           # 已导出事件的只读副本(外部改动不污染内部)

`class InProcessPrivacyTraceSink`

进程内默认 TraceSink:把事件存进内存列表,且拒绝任何携带受限内容的载荷;可选挂 0..N 个 TraceExporter 做扇出。

InProcessPrivacyTraceSink(*, exporters: Sequence[TraceExporter] | None = None) -> None
    # exporters 仅关键字参数;不传即无扇出,纯进程内收集。

emit(self, code: str, **fields: object) -> None
    # 先扫 fields 键,命中 FORBIDDEN_KEYS(归一为小写后比对)立即抛 TraceError、不记录、不扇出;
    # 通过校验的事件以 TraceEvent 追加到内部列表,再扇出到每个 exporter。隐私 by construction。
    # 扇出【严格发生在隐私校验之后】——受限载荷一条都到不了 exporter。

add_exporter(self, exporter: TraceExporter) -> None  # 追加一个扇出目标;后续合法事件都会扇出到它(0.2.0)

codes(self) -> list[str]                            # 已记录事件的 code 序列(按记录顺序)

# 属性(只读 property)
events -> list[TraceEvent]                          # 已记录事件的只读副本(外部改动不污染内部)

`class TraceEvent`(frozen dataclass)

TraceEvent(code: str, fields: dict[str, object] = {})   # 一条被记录的 trace(只读)

`class TraceError(CorespineError)`

trace 载荷违反隐私约定时抛出。code = "trace.forbidden",retryable = False。继承 CorespineError。

`FORBIDDEN_KEYS: frozenset[str]`

禁止出现在 trace 载荷里的键(归一为小写后比对,命中任一即拒整条): {answer, body, chunk, chunk_text, completion, content, prompt, text, value}。

llm —— `corespine.llm.provider`

规范 = OpenAI chat-completions 形状。只有 chat,没有 complete(见 gotchas)。

`class LLMProvider(Protocol)`(`@runtime_checkable`)

chat(self, messages: list[dict[str, Any]], *, tools: list[dict[str, Any]] | None = None) -> ChatCompletion
    # 给 OpenAI 形状的 messages(可选 OpenAI function-tool 形状的 tools),拿回 OpenAI ChatCompletion。
    # tools 是关键字参数(* 之后),必须 tools=... 传。

`class StreamingLLMProvider(Protocol)`(`@runtime_checkable`,0.2.0)

流式 provider 的独立叠加协议——不给 LLMProvider 直接加方法(那会让所有只实现 chat 的既有实现瞬间不再满足 @runtime_checkable 协议、破坏兼容)。非流式实现继续是合法 LLMProvider; 支持流式的实现同时满足两者。

stream_chat(self, messages: list[dict[str, Any]], *, tools: list[dict[str, Any]] | None = None) -> Iterator[ChatCompletionChunk]
    # 逐块吐 ChatCompletionChunk;tools 签名与 chat 对齐(便于同一调用点二选一)。
    # 消费者用 isinstance(p, StreamingLLMProvider) 探测能力再决定走哪条路,不必强迫每个 provider 都实现流式。

等价不变量: 同一次对话流式吐出的所有 chunk 的 delta.content 顺序拼接,逐字等于对应 chat().choices[0].message.content。

`class MockProvider`

确定性离线 provider:零网络、零 key,输出由 messages 派生、可复现。同时满足 LLMProvider 与 StreamingLLMProvider。

MockProvider(*, prefix: str = "mock") -> None       # prefix 仅关键字参数

chat(self, messages: list[dict[str, Any]], *, tools: list[dict[str, Any]] | None = None) -> ChatCompletion
    # 回显最后一条 user 消息,附一段由【整段对话】计算的稳定 12 位 hex 指纹。
    # 同一对话恒定产出同一文本(纯函数);finish_reason 恒为 "stop"。
    # 【绝不伪造 tool_calls】——离线默认不假装会 function-calling(返回 message.tool_calls 为 None)。

stream_chat(self, messages: list[dict[str, Any]], *, tools: list[dict[str, Any]] | None = None) -> Iterator[ChatCompletionChunk]  # 0.2.0
    # 确定性分块流式:首块吐 role(delta.role="assistant"),随后按 16 字符宽度逐块吐 content 片段,
    # 末块吐空 delta + finish_reason="stop"。所有 content 片段拼接 == chat() 的 message.content(等价不变量)。
    # chat 与 stream_chat 共用同一派生逻辑(同 id / 同文本),这正是等价不变量成立的根。

响应数据类(均 frozen dataclass,字段与 OpenAI 完全一致)

ChatCompletion(
    choices: tuple[Choice, ...],                    # 必填
    usage: Usage | None = None,
    model: str = "", id: str = "", created: int = 0, object: str = "chat.completion",
)
Choice(index: int, message: ResponseMessage, finish_reason: str = "stop")     # index/message 必填
ResponseMessage(role: str = "assistant", content: str | None = None, tool_calls: tuple[ToolCall, ...] | None = None)
ToolCall(id: str, function: FunctionCall, type: str = "function")             # id/function 必填
FunctionCall(name: str, arguments: str = "{}")                                # arguments 是 JSON 字符串(与 OpenAI 一致)
Usage(prompt_tokens: int = 0, completion_tokens: int = 0, total_tokens: int = 0)

取文本的惯用路径:completion.choices[0].message.content。

流式数据类(0.2.0,均 frozen dataclass,字段与 OpenAI 完全一致)

ChatCompletionChunk(
    choices: tuple[ChunkChoice, ...],               # 必填
    model: str = "", id: str = "", created: int = 0, object: str = "chat.completion.chunk",
)
ChunkChoice(index: int, delta: ChoiceDelta, finish_reason: str | None = None)  # index/delta 必填;finish_reason 仅末块非空
ChoiceDelta(role: str | None = None, content: str | None = None)               # 首块给 role,后续块给 content 片段

拼接文本的惯用路径:"".join(c.choices[0].delta.content or "" for c in chunks)(== 非流式的 message.content)。

llm rate limit —— `corespine.llm.rate_limit`

`class RateLimitedProvider`

包装任意 LLMProvider,按每分钟 token 上限主动限流(provider-agnostic)。

RateLimitedProvider(inner: LLMProvider, *, tokens_per_minute: int, window_seconds: float = 60.0) -> None
    # tokens_per_minute / window_seconds 必须为正,否则抛 ValueError。

chat(self, messages: list[dict[str, Any]], *, tools: list[dict[str, Any]] | None = None) -> ChatCompletion
    # 限流后转发给 inner.chat。滑动窗口(过去 window_seconds 秒)内已用 token 达到
    # tokens_per_minute 上限时【阻塞等待】,直到最老一条记录滑出窗口再放行;预算够则直接放行。
    # token 用量从返回的 ChatCompletion.usage.total_tokens 在调用后【事后累计】(usage 为 None 记 0),
    # 故为软限流:并发下可能小幅超额,但长期收敛到上限。线程安全(内部用锁保护窗口状态)。

与 SDK 的 max_retries 互补:SDK 撞 429 后被动退避,RateLimitedProvider 预先主动节流。

blob —— `corespine.blob.store`(0.2.0)

「key → bytes」制品存储的最小面。Protocol 只承诺字节 round-trip,元数据最小化(核心不背 content-type / 大小 / 时间戳;需要元数据的 app 自行在 key 命名空间或外层包装里绑)。

`class BlobStore(Protocol)`(`@runtime_checkable`)

put(self, key: str, data: bytes) -> None            # 写入 / 覆盖
get(self, key: str) -> bytes                         # 读回字节;缺失抛 BlobNotFound
delete(self, key: str) -> None                       # 删除,幂等(删不存在的 key 不报错)
exists(self, key: str) -> bool                        # 是否存在(空字节也算存在,不是缺失)

`class MemoryBlobStore`

进程内 dict 实现:零落地、离线确定性(测试 / 临时缓存用)。

MemoryBlobStore() -> None                            # 无参构造

`class FileSystemBlobStore`

本地文件系统实现:key 经 sha256 摊平成 root 下扁平文件名。

FileSystemBlobStore(root: str | Path) -> None        # 需显式 root;构造时 mkdir(parents=True, exist_ok=True)

用 sha256(key) 的 hex 作文件名(而非把 key 直接当路径):杜绝路径穿越(含 .. / 斜杠的 key 也逃不出 root)、跨平台文件名安全、同 key 跨进程确定映射(同 root 新实例读回同字节)。不追求可逆, 不支持枚举 / 反查 key;需要枚举语义的 app 走真实后端。

`make_blob_store(spec: str, **kwargs: object) -> BlobStore` · `BLOB_REGISTRY`

make_blob_store(spec, **kwargs) -> BlobStore
    # 按 spec 构造(大小写 / 连字符 / 留白不敏感)。内置 "memory"(无参)、"filesystem"(需 root=...);
    # 未知 spec 抛 ValueError 并列清可用名。真实后端(s3 / minio)由第三方经 entry-point 注册后同样以 spec 名选用。

BLOB_REGISTRY: Registry[BlobStore]                   # 缝 "blob" 的注册表(内置 memory / filesystem);group "corespine.blob"

`class BlobError(CorespineError)` · `class BlobNotFound(BlobError)`

class BlobError(CorespineError):    code = "blob.error"       # blob 缝的边界异常基类
class BlobNotFound(BlobError):      code = "blob.not_found"   # get 一个不存在的 key 时抛出

config —— `corespine.config.env`

`load_from_env(cls: type[T], *, prefix: str, env: Mapping[str, str] | None = None) -> T`

按 dataclass 字段从 PREFIX_* 读取并构造实例。

cls 必须是 dataclass(否则 TypeError)。
每个字段从 env_key(prefix, field) 读取;命中则按字段注解类型转换,缺失则用 dataclass 字段默认值。
字段无默认值且 env 缺失 → 抛 ValueError(把缺失的 env 名报清楚)。
支持类型:str / int / float / bool(及其 X | None 可选形式)。bool 解析: {1,true,yes,on} 为真、{0,false,no,off,""} 为假(均大小写不敏感),其余 raise ValueError。
env 可注入(默认读 os.environ)——测试传入自己的 mapping,不碰进程环境。

`env_key(prefix: str, field_name: str) -> str`

字段名 → 环境变量名:f"{prefix.rstrip('_')}_{field_name.upper()}"。例:env_key("APP", "port") == "APP_PORT"。

queue —— `corespine.queue.task_queue`

`class TaskQueue(Protocol)`(`@runtime_checkable`)

enqueue(self, func: JobFunc | str, payload: dict[str, Any], *, job_id: str | None = None) -> str
get(self, job_id: str) -> JobStatus | None

JobFunc = Callable[[dict[str, Any]], dict[str, Any]]——job 函数签名约定为 fn(payload: dict) -> dict。

`class FakeQueue`

同步内存队列:enqueue 时内联执行 job(离线 / 测试用,无需任何外部服务)。

FakeQueue() -> None                                 # 无参构造

enqueue(self, func: JobFunc | str, payload: dict[str, Any], *, job_id: str | None = None) -> str
    # func 可传可调用对象,也可传点路径字符串(如 "pkg.mod.fn",末段为可调用对象)。
    # 内联执行 func(payload):成功记 status="finished" + result;
    # 抛异常则【捕获】记 status="failed" + error={"type","message"}(不外抛)。
    # 幂等:显式 job_id 且已知 -> 直接返回该 id,不重跑。返回 job id(未给则取 12 位 hex)。

get(self, job_id: str) -> JobStatus | None          # 按 id 查状态;未知 id 返回 None

`class JobStatus`(dataclass,非 frozen)

JobStatus(id: str, status: str, result: dict[str, Any] | None = None, error: dict[str, Any] | None = None)
    # status 取值:"finished"(成功,看 result) / "failed"(失败,看 error)。

conformance —— `corespine.conformance.harness`

`class InvariantPack(Generic[T])`(frozen dataclass)

一组具名不变量。app 用它装自己的保证;harness 只负责跑。

InvariantPack(name: str, invariants: dict[str, Invariant[T]] = {})

add(self, name: str, check: Invariant[T]) -> InvariantPack[T]    # 登记一个不变量;返回自身(可链式)
names(self) -> list[str]                                         # 全部不变量名

Invariant[T] = Callable[[T], None]——拿到一个【新构造的实现实例】,通过则正常返回、违反则抛异常。检查应只读、只验外部可观测行为。

`class ConformanceSuite(Generic[T])`

把实现注册表 × 不变量包绑成可执行 / 可参数化的套件。

ConformanceSuite(implementations: dict[str, Factory[T]], pack: InvariantPack[T]) -> None
    # implementations: 名字 -> 无参工厂 Callable[[], T](每格各新建实例,杜绝状态串味)。
    # implementations 为空 -> 抛 ValueError。

cases(self) -> list[tuple[str, str]]                # 全部 (实现名, 不变量名) 组合
ids(self) -> list[str]                              # 与 cases() 对齐的可读 id,形如 "impl/invariant"
check(self, impl: str, invariant: str) -> None      # 跑单格:新建该实现实例,执行该不变量(失败原样抛)
run(self) -> list[CaseResult]                       # 跑全部格子并收集结果(不抛;用于离线自检/报告)
passed(self) -> bool                                # 便捷:全部格子是否通过

parametrize_kwargs(self) -> dict[str, object]
    # 产出可直接喂 pytest.mark.parametrize(**...) 的 kwargs(core 不 import pytest,只返回纯数据)。
    # 三键:argnames="case"(单形参,值是已绑定该格的零参 thunk);
    #       argvalues=每格一个 thunk(调用即跑 check,违反则原样抛);
    #       ids=与 argvalues 对齐的可读 id,形如 "impl-invariant"。
    # 消费者侧:
    #     @pytest.mark.parametrize(**suite.parametrize_kwargs())
    #     def test_conformance(case): case()

`class CaseResult`(frozen dataclass)

CaseResult(impl: str, invariant: str, passed: bool, error: str | None = None)
    # 单个 (实现 × 不变量) 格子的执行结果;error 形如 "AssertionError: <msg>"(passed=False 时)。

errors —— `corespine.errors`

`class CorespineError(Exception)`

Spine 家族统一异常基类。

CorespineError(message: str = "", *, code: str | None = None, retryable: bool | None = None, **context: object)
    # code/retryable:既可在子类作类属性覆盖,也可构造时实例级覆盖(仅显式传入才覆盖)。
    # 任意 kwargs 收进 self.context(普通 dict),携带可序列化诊断上下文。

# 类属性默认
code: str = "error"
retryable: bool = False

to_dict(self) -> dict[str, object]
    # 归一为可被 json.dumps 序列化的 dict:
    # {"type": 类名, "code": code, "message": str(self), "retryable": bool, "context": dict}

`error_to_dict(exc: BaseException) -> dict[str, object]`

把【任意】异常归一为可序列化 dict。CorespineError 走其 to_dict();其余异常给同形状的保守默认 (code="error"、retryable=False、context={})。

内置子类

class ConfigError(CorespineError):   code = "config.invalid"   # 配置非法(缺必填、类型不符等)
class SeamError(CorespineError):     code = "seam.unknown"     # 缝相关错误(未知实现、工厂构造失败等)
class ProviderError(CorespineError): code = "provider.error"   # provider/外部服务调用失败的统一边界异常

ProviderError 只包裹 SDK / 外部服务抛出的网络 / 超时 / API 异常;程序错误(KeyError / TypeError 等)不归此类,照常向上抛出,避免韧性兜底掩盖逻辑 bug。

注:Registry.make 的未知 spec 抛的是标准库 ValueError(非 SeamError);load_from_env 的缺失/ 非法抛标准库 ValueError / TypeError。SeamError / ConfigError / ProviderError 是供各 app 自用的语义子类示范。

On this page