API

The complete public API — every spineagent class, protocol, and factory with its real source signature.

完整公开 API。所有签名 100% 来自真实源码(inspect.signature 核对)。一切都从顶层 import spineagent 可达;spineagent.__all__ 即下列名字的全集。导入名 = spineagent。

共享类型(来自 corespine):provider 的 chat 返回 corespine.llm.provider.ChatCompletion (字段:choices: tuple[Choice, ...]、usage: Usage | None、model: str、id: str、 created: int、object: str)。Choice(index, message, finish_reason="stop"); ResponseMessage(role="assistant", content: str|None=None, tool_calls: tuple[ToolCall,...]|None=None); ToolCall(id, function, type="function");FunctionCall(name, arguments="{}"); Usage(prompt_tokens=0, completion_tokens=0, total_tokens=0)。 MockProvider、InProcessPrivacyTraceSink、TraceSink 也来自 corespine。

agent

`class AgentResult`

AgentResult(agent: str, output: str, usage: dict[str, int] | None = None, error: dict[str, object] | None = None, artifacts: tuple[ArtifactRef, ...] = ())

frozen dataclass。一次 agent 步的结果。agent = provenance(产出它的 agent 名)。
属性 ok -> bool:self.error is None。
artifacts:该步产出的产物引用(tuple[ArtifactRef, ...],默认空;ArtifactRef 是轻量元数据引用, 不含字节,见下「artifact 缝」小节。默认路径不产出产物,由使用 artifact sink 的 agent 自行填充)。
契约:成功路径 error is None;error 仅由编排层弹性模式(Coordinator 的 run_sequential / run_parallel / run_pipeline 传 resilient=True;resilient 是这些方法的关键字参数,不是构造器参数)捕获 step 异常时填充,值是 corespine.errors.error_to_dict(exc) 的归一 dict(含 code / retryable / message / context)。

`class Agent` (Protocol, runtime_checkable)

只读属性 name(声明为 @property def name(self) -> str);方法 step(self, task: str, *, trace: TraceSink | None = None) -> AgentResult。
所有 agent 实现都满足它,故可互换进编排 / 桥接。
name 声明为只读 @property(而非可写变量):具体实现既有 @property(如 LlmAgent.name)也有纯类属性(如 EchoTool.name = "echo"),只读属性对两者都兼容;若协议把 name 声明成可写变量, mypy --strict 会因「只读 property vs 可写变量」的型变冲突而报错。语义上不改任何具体实现。

`class LlmAgent`

LlmAgent(name: str, provider: LLMProvider, *, system: str = "")

step(task, *, trace=None) -> AgentResult:把 task(+ 可选 system)按 OpenAI messages 喂给 provider.chat,取 choices[0].message.content 作 output,带 usage。
离线传 corespine.MockProvider();线上传任意真实 provider 适配器,代码不变。

`class FunctionAgent`

FunctionAgent(name: str, fn: Callable[[str], str])

step(task, *, trace=None) -> AgentResult:output = fn(task)。无需 LLM,做编排 / 测试节点。

`class ToolUsingAgent`

ToolUsingAgent(name: str, policy: ToolPolicy, tools: Iterable[Tool], *, max_steps: int = 8)

step(task, *, trace=None) -> AgentResult:在一次 step() 内循环——policy.decide(...) 决定 ToolCall(按名取 Tool 执行、观测追加进 history)或 Finish(返回最终答案)。
$prev:工具参数里的字面量 $prev 在执行前替换为上一步观测输出(首步无上一步则替换为空串)。
max_steps = 最多调用多少次工具(收尾不占预算);触顶强制收尾,绝不死循环。
实现 Agent 协议,可直接进 Coordinator / ChainAgent / 被 AgentTool 包成工具。

`class FunctionCallingAgent`

FunctionCallingAgent(name: str, model: LLMProvider, tools: Iterable[FunctionTool], *, system: str = "", max_steps: int = 8)

step(task, *, trace=None) -> AgentResult:真 LLM 原生 function-calling 多步循环——把每个 FunctionTool.schema() 喂给 model.chat(messages, tools=...);模型回 tool_calls 则逐个 tool.invoke(json.loads(arguments))、以 OpenAI tool 角色消息喂回、再 chat;无 tool_calls 则出文本收尾。触顶 max_steps 兜底非空。
离线 MockProvider 不回 tool_calls → 直接出文本(诚实:离线不假装会 function-calling)。要真正跑工具循环,需注入会回 tool_calls 的 provider(真实后端,或测试用脚本化 fake)。

`class AgentTool`

AgentTool(agent: Agent, *, name: str | None = None)

实现 Tool 协议。name 默认取 agent.name。
run(arg: str) -> ToolResult:对子 agent 跑一步,把 output 包成带 provenance 的 ToolResult。
用于分层 / 督导式多 agent(可层层嵌套)。最薄桥:只搬运文本,子 agent 的 usage / error 不透传; 子 agent 抛异常照常上抛(错误处理归编排层 / 调用方)。

DeepResearchAgent(name: str = "deep_research", *, provider: LLMProvider | None = None, tools: Iterable[FunctionTool] = (), planner: Callable[[str], list[str]] | None = None, max_subqueries: int = 5, retriever_system: str = "", synthesis_system: str = "")

预置 agent(实现 Agent 协议),纯组合已有原语,不引入新机制。step(task, *, trace=None) 三段:
1. 拆解:用 planner(默认 default_planner,按换行 / ; / ; 结构切分)把任务拆成 ≤ max_subqueries 个子查询;
2. 并行检索:共享一个 FunctionCallingAgent 检索器,每个子查询包成 FunctionAgent,经 Coordinator(...).run_parallel("", resilient=True) 扇出;
3. 综合:一个 LlmAgent 把各路发现拼成 prompt 产出终答。
离线默认 provider=MockProvider()、tools=(),零网络即可端到端跑(输出是确定性 mock 综合)。产出隐私安全 "deep_research" trace(只记子查询数 / 发现数 / 输出长度)。结果 provenance = name。

`default_planner`

default_planner(task: str) -> list[str] — DeepResearchAgent 的默认拆解器:按换行 / ; / ; 做结构化切分(非 LLM)。

tool-policy 缝(会用工具的 agent 的「大脑」)

`class ToolPolicy` (Protocol, runtime_checkable)

decide(self, task: str, *, tools: tuple[str, ...], history: tuple[Observation, ...]) -> Action
给任务 + 可用工具名集 + 历史观测,定下一个动作。约定是无状态纯函数(同输入恒同输出)。

`class ToolCall`

ToolCall(tool: str, arg: str) — frozen dataclass。决定:调一个工具(arg 中字面量 $prev 由 agent 侧替换)。

`class Finish`

Finish(answer: str) — frozen dataclass。决定:收尾给最终答案(约定非空)。

`Action`

Action = ToolCall | Finish(typing.TypeAlias,PEP 604 联合;isinstance 分发)。

`class Observation`

Observation(tool: str, arg: str, output: str) — frozen dataclass。一步执行的观测,喂回循环。

`class SyntaxToolPolicy`

SyntaxToolPolicy() — 离线确定性默认实现。decide(...) 按任务文本里 <tool>: <arg> 显式语法 + 工具名集合确定性路由:游标 = len(history),第 cursor 条工具指令尚存则 ToolCall(该行工具名, 该行参数), 指令耗尽则 Finish(把非指令正文行 + 最后一步观测拼成非空答案)。不假装 LLM 推理。

`tool_policies`

Registry[ToolPolicy](seam 名 tool_policy)。

tool_policies.make(spec, **kwargs) -> ToolPolicy、tool_policies.names() -> list[str]。
已注册:"offline"(→ SyntaxToolPolicy)、"llm"(真实推理式占位,调用即抛 SeamError—— 留待接真 provider 解析 function-calling 后接入)。

tools

`class ToolResult`

ToolResult(tool: str, output: str) — frozen dataclass。tool = provenance(产出它的工具名)。

`class Tool` (Protocol, runtime_checkable)

只读属性 name(声明为 @property def name(self) -> str);方法 run(self, arg: str) -> ToolResult。
同 Agent.name:声明成只读 @property 是为满足 mypy --strict 型变——纯类属性(EchoTool.name = "echo") 与 @property 实现都兼容。

`class EchoTool`

EchoTool(),name = "echo"。run(arg) -> ToolResult:原样回显 arg。

`class CalcTool`

CalcTool(),name = "calc"。run(arg) -> ToolResult:安全求值算术表达式(白名单 + - * / % ** 与一元 + -,整数结果去掉 .0);非算术节点抛 ValueError,绝不 eval 任意代码。

`tool_registry`

Registry[Tool](seam 名 tool)。

tool_registry.make(spec, **kwargs) -> Tool、tool_registry.names() -> list[str]。
已注册:"echo"、"calc"。支持 entry-point group corespine.tool 第三方工具自动发现。
注:变量名是 tool_registry(不是 tools),以避开与 spineagent.tools 子包同名。

`class FunctionTool`

FunctionTool(name: str, description: str, parameters: dict[str, Any], func: Callable[..., Any]) — dataclass。

schema() -> dict[str, Any]:产出 OpenAI function-tool 形状 {"type":"function","function":{name,description,parameters}},直接喂给 LLMProvider.chat(tools=...)。
invoke(arguments: dict[str, Any]) -> str:用模型给的结构化 dict 调底层函数,str(...) 结果(回填进对话)。
注:FunctionTool 实现的是 invoke(dict 参数),不实现 Tool.run(str 参数);它专给 FunctionCallingAgent 用,不能直接丢进 ToolUsingAgent。

`function_tool`

function_tool(func: Callable[..., Any] | None = None, *, name: str | None = None, description: str | None = None) -> Any

装饰器:把普通函数包成 FunctionTool。name 默认 func.__name__,description 默认其 docstring, parameters 从签名 + 类型注解自动推 JSON-schema(无默认值的参数为 required;str/int/float/bool/list/dict 映射到 JSON 类型,未识别落 string)。
用法:@function_tool 直接装,或 @function_tool(name=..., description=...) 覆盖。

orchestration

`class Coordinator`

Coordinator(agents: Iterable[Agent], *, trace: TraceSink | None = None)

属性 agents -> list[Agent](副本)。
run_sequential(task: str, *, resilient: bool = False) -> list[AgentResult]:逐个跑同一任务,保序。
run_parallel(task: str, *, max_workers: int | None = None, resilient: bool = False) -> list[AgentResult]: 线程池并发跑同一任务,结果仍按 agent 输入顺序返回(max_workers 默认 = agent 数)。
run_pipeline(task: str, *, resilient: bool = False) -> list[AgentResult]:链式——上一个 agent 的 output 作下一个的输入,保序收集每段。
弹性容错 resilient=True:单 agent 异常归一为 error_to_dict(exc) 塞进该步 AgentResult.error, 批次继续(顺序 / 并行跑完其余;流水线在失败处停止)。resilient=False(默认)= fail-fast,异常冒泡。
编排级 trace 只记 mode / agent_count / failures / took_ms,绝不记正文。

`class ChainAgent`

ChainAgent(name: str, agents: Iterable[Agent])

实现 Agent 协议。step(task, *, trace=None) -> AgentResult:复用 Coordinator(...).run_pipeline(task) 把任务逐段传递,返回末端 agent 输出(provenance = chain 名;空链退化为恒等透传)。失败 fail-fast 冒泡。
让流水线成为一等可组合单元:可进 Coordinator / 当 AgentTool 工具 / 套进另一个 chain。

protocol: mcp

`class McpTool`

McpTool(name: str, description: str = "") — frozen dataclass。一个 MCP 工具的最小描述。

`class McpClient` (Protocol, runtime_checkable)

list_tools() -> list[McpTool]、call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]。

`class McpServer` (Protocol, runtime_checkable)

register_tool(tool: McpTool, handler: ToolHandler) -> None、tools() -> list[McpTool]。
ToolHandler = Callable[[dict[str, Any]], dict[str, Any]](模块内类型别名,未导出顶层)。

`class OfflineMcpStub`

OfflineMcpStub() — 离线进程内回环,同时满足 McpClient 与 McpServer。

register_tool(tool, handler)、tools()、list_tools()、call_tool(name, arguments)(未注册抛 KeyError)。

`class McpClientTool`

McpClientTool(name: str, client: McpClient, *, arg_key: str = "input", result_key: str = "result")

实现 Tool 协议。run(arg: str) -> ToolResult:把单串 arg 包成 {arg_key: arg} 调 client.call_tool(name, ...),取结果 dict 的 result_key 转字符串,带 provenance(tool = name)。
最薄阻抗匹配:只做 str 单参 + 单键结果映射。

`mcp_clients`

Registry[McpClient](seam 名 mcp_client)。make(spec, **kw) / names()。

已注册:"offline"(→ OfflineMcpStub)、"real"(走 [mcp] extra 延迟 import 后抛 SeamError,待接入)。

`load_mcp_sdk() -> Any`

延迟 import 真实 MCP SDK(import 名 mcp);未装 [mcp] extra 时给「pip install spineagent[mcp]」友好报错。

protocol: a2a

`class A2ATask`

A2ATask(task_id: str, text: str) — frozen dataclass。一条跨 agent 任务(协议载荷本身)。

`class A2AResult`

A2AResult(task_id: str, output: str, agent: str) — frozen dataclass。agent = provenance。

`class A2AAgent` (Protocol, runtime_checkable)

属性 name: str;card() -> dict[str, Any](能力描述);send(task: A2ATask) -> A2AResult。

`class OfflineA2AStub`

OfflineA2AStub(*, name: str = "offline-a2a", responder: Callable[[str], str] | None = None)

离线回环。responder 默认 lambda text: f"echo:{text}"。
name、card()({"name","transport":"offline-loopback","skills":["echo"]})、send(task) -> A2AResult。

`class A2AAgentAdapter`

A2AAgentAdapter(remote: A2AAgent, *, task_id: str = "task")

实现 Agent 协议。name 取 remote.name。step(task, *, trace=None) -> AgentResult:把 task 包成 A2ATask 交给 remote.send,把 A2AResult 转成 AgentResult。透明桥:输出原样继承自 remote。

`a2a_agents`

Registry[A2AAgent](seam 名 a2a_agent)。make(spec, **kw) / names()。

已注册:"offline"(→ OfflineA2AStub)、"real"(走 [a2a] extra 延迟 import 后抛 SeamError,待接入)。

`load_a2a_sdk() -> Any`

延迟 import 真实 A2A SDK(a2a-sdk,import 名 a2a);未装 [a2a] extra 给友好报错。

sandbox 缝(代码执行的隔离缝)

同家族元模式:Protocol + 离线确定性默认 + Registry 工厂 + 真实后端经可选 extra 延迟 import。模块 spineagent.sandbox.seam。

`class Sandbox` (Protocol, runtime_checkable)

只读属性 name;方法 run(self, code: str, *, timeout: float | None = None, limits: Limits | None = None, env: Mapping[str, object] | None = None) -> SandboxResult。

`class InProcessSandbox`

InProcessSandbox(),name = "in_process"。离线确定性默认:白名单 AST 求值器,把代码当「纯表达式白名单迷你语言」跑。

安全内建白名单仅 len/str/int/float/bool/abs/round/min/max/sum/sorted/repr/list/dict/tuple/set——故意没有 open/eval/exec/__import__/getattr/range;拒绝 import / 属性访问 / 任意调用 / 非白名单名字。
失败不抛异常,归一为非零 SandboxResult(error 原因码 syntax / disallowed / limit_exceeded / error)。

支撑 dataclass

Limits(timeout_seconds: float | None = None, max_output_chars: int | None = None, max_ops: int | None = None) — frozen。
DEFAULT_LIMITS = Limits(timeout_seconds=5.0, max_output_chars=64_000, max_ops=100_000)。
ResourceUsage(ops: int = 0, output_chars: int = 0, wall_seconds: float = 0.0) — frozen。
SandboxResult(sandbox: str, output: str, returncode: int = 0, usage: ResourceUsage = ..., error: str | None = None) — frozen,属性 ok -> bool(returncode == 0)。

`sandboxes`

Registry[Sandbox](seam 名 sandbox)。make(spec, **kw) / names()。支持 entry-point group corespine.sandbox。

已注册:"in_process"(→ InProcessSandbox)、"subprocess"、"container"。
"subprocess" / "container" 是桩:make(...) 直接抛 corespine.errors.SeamError(留待使用者按平台接入; "container" 先经 load_container_sdk() 延迟 import docker([sandbox] extra)再抛 SeamError)。本壳只提供缝 + InProcessSandbox。

`load_container_sdk() -> Any`

延迟 import docker([sandbox] extra);未装时给友好报错。

skills 缝(manifest 文件夹包 → 桥成 FunctionTool)

模块 spineagent.skills.*。把一个「manifest 文件夹包」当技能,桥进 FunctionTool 供 FunctionCallingAgent 用。

`class SkillSpec`

SkillSpec(name: str, description: str, inputs: dict[str, Any]) — frozen。技能规格(inputs 是 JSON-schema object)。

`class SkillResult`

SkillResult(skill: str, output: str, usage: ResourceUsage | None = None) — frozen。skill = provenance。

`class Skill` (Protocol, runtime_checkable)

属性 spec: SkillSpec;describe() -> dict[str, Any](产出 OpenAI function-tool schema);invoke(args: dict[str, Any]) -> SkillResult。

`class FixtureSkill`

FixtureSkill(spec: SkillSpec, script: str, *, sandbox: Sandbox | None = None) — 离线默认实现。

describe() 回 OpenAI function-tool schema;invoke(args) 把 script 经 sandbox.run(..., env=args) 跑(默认 InProcessSandbox),非 ok 抛 SkillError。

`class SkillBundle`

@staticmethod SkillBundle.load(path: str | Path, *, sandbox: Sandbox | None = None) -> Skill:从文件夹包加载。
- manifest 文件名 manifest.toml(3.11+ 走 tomllib,3.10 回退 tomli);必填字段 name / description / inputs, 可选 script(默认脚本文件名 skill.py)。返回一个 FixtureSkill。

`skill_as_function_tool`

skill_as_function_tool(skill: Skill) -> FunctionTool — 桥:取 skill.describe()["function"] 的 name/description/parameters 建 FunctionTool,func=lambda **kwargs: skill.invoke(kwargs).output。桥出来即可丢进 FunctionCallingAgent。

`class SkillError`

SkillError(skill: str, reason: str, detail: str)(继承 RuntimeError)。

`skill_registry`

Registry[Skill](seam 名 skill)。make(spec, **kw) / names()。支持 entry-point group corespine.skill。

已注册:"fixture"(→ FixtureSkill)、"bundle"(→ SkillBundle.load,需传 path=)。

middleware 缝(agent step 的洋葱环切面)

模块 spineagent.agent.middleware。围绕任意 Agent 的一次 step 装一圈可组合切面(只记元数据,隐私安全)。

`class StepContext`

StepContext(agent: str, task: str, trace: TraceSink | None = None, step: int = 0, tools: list[str] = ..., attachments: list[str] = ..., extras: dict[str, Any] = ...) — dataclass。中间件读写的一步上下文。

`class Middleware` (Protocol, runtime_checkable)

before_step(self, ctx: StepContext) -> None、after_step(self, ctx: StepContext, result: AgentResult) -> AgentResult。
只有 before_step / after_step 两个钩子,没有 wrap。

`class MiddlewareAgent`

MiddlewareAgent(name: str, agent: Agent, middlewares: Iterable[Middleware]) — 实现 Agent 协议。

step(...):洋葱式——正序跑各 before_step → 内层 agent.step → 逆序跑各 after_step,末了把 result.agent 重盖为 name。

四件套(内建 Middleware)

TokenUsageMiddleware(tokenizer: Callable[[str], int] | None = None) — 数入(ctx.task)/出(result.output)token 累进 self.totals,发 "mw_token_usage" trace(默认 tokenizer = 空白切分)。
SummaryMiddleware(provider: LLMProvider | None = None, *, max_chars: int = 2000) — ctx.task 超 max_chars 时调 provider(默认 MockProvider)摘要并替换 ctx.task,发 "mw_summary"。
DynamicToolMiddleware(tools_by_step: Mapping[int, Sequence[str]] | None = None, *, default: Sequence[str] = ()) — 按当前步号设 ctx.tools,发 "mw_dynamic_tools"。
AttachmentMiddleware(attachments: Mapping[str, str] | None = None) — 把附件正文前置进 ctx.task、填 ctx.attachments,发 "mw_attachment"。

`middlewares`

Registry[Middleware](seam 名 middleware)。make(spec, **kw) / names()。支持 entry-point group corespine.middleware。

已注册:"token_usage"、"summary"、"dynamic_tool"、"attachment"。

artifact 缝(agent 产物落地缝)

模块 spineagent.agent.artifact。让一步产出的二进制 / 文本产物有统一的 store / fetch 缝,引用回填进 AgentResult.artifacts。

`class Artifact`

Artifact(name: str, data: bytes, mime: str = "application/octet-stream", producer: str = "") — frozen。带字节的产物。

classmethod Artifact.from_text(name, text, *, mime="text/plain", producer="") -> Artifact;属性 text(UTF-8 解码)。

`class ArtifactRef`

ArtifactRef(key: str, sink: str, name: str, mime: str, producer: str, size: int) — frozen。轻量引用(不含字节),即 AgentResult.artifacts 的元素。元数据随引用走。

`class ArtifactSink` (Protocol, runtime_checkable)

只读属性 name;store(self, artifact: Artifact) -> ArtifactRef、fetch(self, ref: ArtifactRef) -> Artifact。

`class InProcessArtifactSink`

InProcessArtifactSink(),name = "in_process"。进程内 dict 落地,离线确定性默认。

`class BlobArtifactSink`

BlobArtifactSink(store: BlobStore, *, name: str = "blob") — 组合 corespine 的 corespine.blob.store.BlobStore。

store(artifact):算内容寻址 key sha256(data).hexdigest(),调 store.put(key, artifact.data),回带元数据的 ArtifactRef。
fetch(ref):data = store.get(ref.key),用 ref 上的元数据重建 Artifact。
跨仓关系:本类只注入并包裹一个 corespine BlobStore 实例(内存 / 文件系统 / 第三方 S3 皆可);元数据全放 ArtifactRef,BlobStore 只负责字节往返(put / get),不存元数据。方向 spineagent → corespine。

`artifact_sinks`

Registry[ArtifactSink](seam 名 artifact_sink)。make(spec, **kw) / names()。支持 entry-point group corespine.artifact_sink。

已注册:"in_process"(→ InProcessArtifactSink)、"blob"(→ BlobArtifactSink,需注入 store=)。

llm provider 适配器(对外统一 OpenAI ChatCompletion 形状)

所有适配器都实现 corespine 的 LLMProvider 协议: chat(self, messages: list[dict[str, Any]], *, tools: list[dict[str, Any]] | None = None) -> ChatCompletion。传 OpenAI 形状的 messages / function-tools,拿回 OpenAI 形状的 ChatCompletion。client= 可注入 fake / 真实 client 做离线单测;不注入则在构造时经对应 load_*_sdk() 延迟 import 真实 SDK。

`class OpenAICompatProvider`

OpenAICompatProvider(model: str, *, max_tokens: int = 4096, client: Any = None, base_url: str | None = None, extra: dict[str, Any] | None = None, **client_kwargs)

走官方 openai SDK 的 chat.completions.create。model 必填(各兼容端点模型名不同)。base_url 指向兼容端点(留空 = 官方 OpenAI)。messages / tools 直传(本就是 OpenAI 形状)。extra [openai]。
一个适配器覆盖一切「OpenAI 兼容」端点(OpenAI / Azure / Together / Groq / DeepSeek / Mistral / xAI / Qwen / Moonshot / Ollama / vLLM / OpenRouter / LiteLLM …)。

`class AnthropicProvider`

AnthropicProvider(*, model: str = "claude-opus-4-8", max_tokens: int = 4096, client: Any = None, extra: dict[str, Any] | None = None, **client_kwargs)

走官方 anthropic SDK 的 messages.create。内部把 OpenAI messages/tools 转成 Anthropic 原生形状, 再把响应(text / tool_use / stop_reason / usage)转回 OpenAI ChatCompletion。extra [anthropic]。

`class CohereProvider`

CohereProvider(*, model: str = "command-r-plus", client: Any = None, extra: dict[str, Any] | None = None, **client_kwargs)

走官方 cohere SDK 的 ClientV2.chat。Cohere v2 native → OpenAI ChatCompletion。extra [cohere]。

`class GeminiProvider`

GeminiProvider(*, model: str = "gemini-2.5-flash", client: Any = None, extra: dict[str, Any] | None = None, **client_kwargs)

走官方 google-genai SDK 的 models.generate_content。Gemini native → OpenAI ChatCompletion (自造 tool_call id、args→JSON 串、role model→assistant)。同覆盖 AI Studio 与 Vertex。extra [gemini]。

`class BedrockConverseProvider`

BedrockConverseProvider(model: str, *, client: Any = None, region_name: str | None = None, extra: dict[str, Any] | None = None, **client_kwargs)

走 boto3 的 bedrock-runtime Converse API(跨模型同形)。model 必填(= Bedrock modelId)。 Converse native → OpenAI ChatCompletion。extra [bedrock]。

流式:`StreamingLLMProvider`(增量输出的可选加成协议)

除 chat 外,部分适配器额外实现 corespine 的 StreamingLLMProvider 协议: stream_chat(self, messages: list[dict[str, Any]], *, tools: list[dict[str, Any]] | None = None) -> Iterator[ChatCompletionChunk]。

产出 corespine.llm.provider.ChatCompletionChunk(ChunkChoice / ChoiceDelta)。契约:按序拼接每块的 delta.content 恒等于非流式 chat() 的 content(由 conformance 钉死)。
消费方经 isinstance(provider, StreamingLLMProvider) 检测支持与否。
实现流式的:OpenAICompatProvider(stream=True,SDK 块 ~1:1 转 ChatCompletionChunk)、 AnthropicProvider(把 Anthropic 事件转成 OpenAI 形状块:message_start→role、content_block_delta/text_delta→content、 message_delta 的 stop_reason→finish_reason;仅流式文本,tool_use 流式未实现)。
诚实不实现:CohereProvider / GeminiProvider / BedrockConverseProvider 不实现 stream_chat(其 native 流式形状差异大); isinstance(..., StreamingLLMProvider) 对它们如实回 False。

`llm_providers`

Registry[LLMProvider](seam 名 llm)。make(spec, **kw) / names()。

已注册:"mock"(→ corespine.MockProvider)、"openai"、"anthropic"、"cohere"、"gemini"、"bedrock"。

`load_*_sdk()`

load_anthropic_sdk() / load_openai_sdk() / load_cohere_sdk() / load_gemini_sdk() / load_boto3_sdk() → Any。各延迟 import 对应真实 SDK,缺对应 extra 时给「pip install spineagent[<extra>]」友好报错。

conformance(本包绑定的不变量)

供 corespine.ConformanceSuite(implementations, pack) 消费;pack 即下列 InvariantPack。

AGENT_INVARIANTS: InvariantPack[Agent](名 agent_step):step_returns_output、 result_carries_agent_provenance、step_traces_are_privacy_safe。
TOOL_INVARIANTS: InvariantPack[Tool](名 tool_call):result_carries_tool_provenance、run_returns_output。
POLICY_INVARIANTS: InvariantPack[ToolPolicy](名 tool_policy):action_is_a_known_variant、 never_calls_an_unavailable_tool、empty_tools_yields_nonempty_finish、decide_is_pure。

`version`

spineagent.__version__ -> str(当前 "0.1.0")。

On this page