ChainWeaver

ChainWeaver

找出 Agent 不再需要推理的环节。审查证据。将获认可的路径转化为受治理的确定性能力。

正在验证的产品论点——观察 → 证明 → 审查 → 编译。 ChainWeaver 可以检查重复的工具行为、找出候选路径，并通过类型化合约执行经过审查的确定性路径。确定性执行本身并非护城河：如果你已经了解确切的工作流，那么普通的 Python 函数、LangGraph 节点或服务商原生工具可能更简单。当前正在验证的假设是，从跟踪记录中提取证据、有价值的拒绝、受治理的晋级、安全边界的保留以及漂移检测，能让 ChainWeaver 值得采用。请参阅产品验证与采用门槛 和 #553。

消除推理边界，绝不消除安全边界。 将多个工具调用编译为一个能力，不得在无提示的情况下汇总权限或抹去子调用的审批要求。此不变量已在 #554 中明确跟踪。

确定性工具路径的治理。 每个步骤都采用类型化输入/输出、可序列化为文件的工作流、架构漂移检测、确定性证明、属性模糊测试和结构化审计跟踪，为确实满足确定性晋级条件的路径提供严谨的执行基础。

基准测试是关于执行器的证据，而非产品与市场契合度的证明。 仓库的基准测试报告可复现——你可以运行 python benchmarks/report.py 重新生成——报告显示，在合成对比中，确定性核心避免了由大模型介导的状态转换。但它并不能证明每条重复路径都应该编译，也不能证明 ChainWeaver 优于显而易见的纯 Python 实现。独立验证计划明确要求进行这种手动基准对比。

from chainweaver import Tool, Flow, FlowStep, FlowRegistry, FlowExecutor
# (NumberInput, ValueOutput, double_fn defined in full example below)

# 1. Wrap any function as a schema-validated Tool
double = Tool(name="double", description="Doubles a number.",
              input_schema=NumberInput, output_schema=ValueOutput, fn=double_fn)
# 2. Wire tools into a Flow
flow = Flow(name="calc", description="Double a number.",
            steps=[FlowStep(tool_name="double", input_mapping={"number": "number"})])
# 3. Register and execute — zero LLM calls
registry = FlowRegistry()
registry.register_flow(flow)
executor = FlowExecutor(registry=registry)
executor.register_tool(double)
result = executor.execute_flow("calc", {"number": 5})
# result.final_output → {"number": 5, "value": 10}

请参阅下方的完整示例，或运行 python examples/simple_linear_flow.py

安装 · 为什么选择 ChainWeaver？ · 适合我吗？ · 产品验证 · 快速入门 · 架构 · 文档站点 · 路线图

---

30 秒了解

确定性执行器解决一个简单的问题：一旦证明某条路径不需要中间推理，就不必每次运行时都让模型重新决定相同的流程编排。

之前——由模型介入的路径：

turn 1   ─►  LLM("plan")    ─►  search(query)         ─► 12 results
turn 2   ─►  LLM("next?")   ─►  extract(results)      ─► 8 facts
turn 3   ─►  LLM("next?")   ─►  validate(facts)       ─► 7 facts
turn 4   ─►  LLM("next?")   ─►  format(facts)         ─► answer

审核后——已接受的路径可以确定性地运行：

turn 1   ─►  LLM("plan")    ─►  search_summarize_flow(query)
                                  └─ search ─► extract ─► validate ─► format

Agent 仍会决定调用哪项能力。能力内部的确定性步骤会通过严格的 Pydantic 验证运行，不涉及大模型。

更难的产品问题出现在这张图之前：这条路径是否应该编译？有用的 ChainWeaver 分析必须能够说明候选路径为何具有重复性且在结构上安全，并且拒绝那些仍涉及语义判断、副作用、授权或审批边界的路径。目前正在通过 #553 对独立轨迹验证这一主张。

可直接复制粘贴的执行器路径：

pip install 'chainweaver[yaml]'
python examples/simple_linear_flow.py

下面的摘要是脚本生成的真实 ExecutionResult 的精简视图：

flow=double_add_format success=True
final_output={'number': 5, 'value': 20, 'result': 'Final value: 20'}
step 0 double          {'value': 10}
step 1 add_ten         {'value': 20}
step 2 format_result   {'result': 'Final value: 20'}

---

为什么选择 ChainWeaver？

为什么不直接写一个 Python 函数？

很多时候，你确实应该这么做。

如果你的团队已经知道工作流是固定的，那么普通函数或现有框架中的工作流原语通常是复杂度最低的方案。只有在 ChainWeaver 的生命周期能带来实质价值时，才值得增加这一依赖，例如：

• 从真实轨迹中发现不明显、反复出现的模型介入路径；
• 展示重复性、数据流兼容性和反例的证据；
• 拒绝那些看似合适、但仍需要语义判断的路径；
• 在推广过程中保留审批和授权限制；
• 生成可复现的审核证据和产物标识；
• 在推广后检测架构、安全性或策略漂移；
• 导出或执行已接受的能力，而不假定减少大模型调用就必然提升正确性。

这些优势在真实团队中是否足够显著，是一个可证伪的产品假设，而不是 README 中的既定结论。请参阅 docs/product-validation.md。

当由大模型驱动的 Agent 将工具串联起来——fetch_data → transform → store——一种常见做法是在步骤之间插入大模型调用，由模型决定下一步该做什么。对于一条已被证明并审核为完全确定性的路径，这些中间调用会增加延迟、成本和不确定性，却无法带来有用的判断。

ChainWeaver 的执行器可以在步骤之间不涉及大模型的情况下，运行已接受的确定性路径：

User request
    │
    ▼
FlowExecutor ──► Tool A ──► Tool B ──► Tool C
    │
    ▼
Response

标准	由模型介入的路径	ChainWeaver 确定性路径
确定性步骤之间的大模型调用	可能有一次或多次	0 次
可复现性	取决于模型的决策	确定性路径
架构验证	取决于框架或应用	由 Pydantic 强制执行
可观测性	取决于框架或应用	结构化步骤日志
可复用性	取决于应用	已注册、已版本化的流程

这与 LangChain / LangGraph / Prefect / Dagster / Temporal 有何不同？

这些框架也能执行确定性代码。不应该因为其他地方无法实现确定性执行而选择 ChainWeaver。它目前的产品主张更为明确：从观察到的 Agent／工具行为入手，确定哪些部分不再需要推理，让证据和拒绝理由可供审核，然后将已接受的路径推广为受治理的确定性能力。

执行底层刻意保持精简，并且步骤之间不调用大模型；但本项目正在验证，用户看重的究竟是不是证据与治理生命周期，而不只是多一个工作流运行时。

有关详细的版本化对比，请参阅 docs/comparisons.md；有关可证伪这一定位的标准，请参阅 docs/product-validation.md。

---

这适合我吗？

ChainWeaver 专为一种特定的问题场景而构建。 完整的适用/不适用说明页面介绍了其中的细节；简而言之：

适合使用 ChainWeaver 的情况

• 你有真实的 Agent/工具调用轨迹，并怀疑其中部分路径是重复的流程代码，而非有用的模型判断。
• 你希望通过证据和审查来确定哪些路径值得提升为确定性流程，而不只是为已知工作流提供运行时。
• 路径提升后，确定性、严格的架构、可审计性和漂移检测都很重要。
• 你愿意明确维护安全与审批边界，而不是把宏工具调用视为对所有子操作的全面授权。

不适合使用 ChainWeaver 的情况

• 你已经了解工作流，普通 Python 函数或现有框架就足以清晰地表达它。
• 每一步都需要开放式推理来选择下一步（请使用 Agent 框架：LangGraph、OpenAI / Anthropic SDK 工具调用循环）。
• 你需要一个用于跨时间调度或持久化作业的通用工作流引擎（请使用 Prefect、Dagster 或 Temporal）。
• 你期望执行器调用大模型。它有意不这样做。
• 编译期间，你无法保留有副作用路径的授权/审批语义。

产品主张、验证协议和停止/转向标准已公开于 docs/product-validation.md。

确定性执行设计背后的正确性论证，请参阅 docs/data-integrity.md。

Weaver Stack 的一部分

ChainWeaver 是 Weaver Stack 中的确定性多步骤工具执行层。Weaver Stack 是一组小巧、可组合的 SDK，共享 weaver-spec 的 SelectableItem 路由契约。在请求路径上，路由器会选择要调用的能力，ChainWeaver 负责运行其背后的确定性工具路径，下游层则会对调用进行门控和防护：

flowchart LR
    req([Request]) --> ctx[contextweaver<br/>context assembly]
    ctx --> cw[<b>ChainWeaver</b><br/>deterministic flow execution]
    cw --> ak[agent-kernel<br/>capability gating]
    ak --> af[agentfence<br/>runtime guardrails]
    subgraph adjacent [Adjacent · use any subset]
        vg[vibeguard]
        lw[lessonweaver]
        se[skdr-eval]
    end

可单独使用，也可组合使用。 每一层都能独立运行——ChainWeaver 的基础安装没有任何硬性依赖于其他同系列项目，可完全独立运行。实际互操作通过 chainweaver[weaver-stack] 额外组件实现，它会固定已发布的 weaver-contracts 包版本：ChainWeaver 直接使用其中的 SelectableItem / RoutingDecision / CapabilityToken 类型，因此路由器可以将路由决策直接传给 resolve_flow_from_routing_decision()，以进行确定性执行。请参阅可运行的 Weaver Stack 黄金路径示例（问题 #234）。

层	负责内容	同系列项目
路由 / 能力选择	“哪个具名操作负责处理此请求？”	weaver-spec（#91 — SelectableItem 契约）
上下文组装	“提示词中应包含哪些事实和工具描述？”	contextweaver（#106）
Agent 内核	由模型驱动的工具调用循环本身	agent-kernel（#89）
确定性流程执行	“严格按照架构运行这一确切的工具序列，步骤之间不调用大模型”	ChainWeaver — 本代码库
经验与评估	将轨迹转化为经过审查的操作指南（ChainWeaver 如何为其提供信息）	lessonweaver（#210）

ChainWeaver 不会取代 Agent 框架。它旨在由 Agent 框架调用——请参阅 LangGraph 配方（问题 #205）和 OpenAI Agents SDK 配方（问题 #206），了解规范的集成模式。

有关主机层面的预期（何时调用、如何存储轨迹、有副作用的工具、MCP 一致性），请参阅 运行时职责页面。

---

安装

pip install chainweaver                  # base install — no extras
pip install 'chainweaver[yaml]'          # most common — needed for .flow.yaml files
pip install 'chainweaver[yaml,otel,mcp]' # combine extras with commas

基础安装仅拉取 5 个运行时依赖（deepdiff、packaging、pydantic、tenacity、typer），且不会固定任何传递依赖的大模型 SDK。请根据实际使用的集成选择额外依赖：

Extra	Use when	Pulls in
chainweaver[yaml]	读取／写入 .flow.yaml 流文件（CLI 的 run、validate、check、doctor 命令需要此依赖）	pyyaml
chainweaver[otel]	为每次流运行生成 OpenTelemetry span	opentelemetry-api
chainweaver[mcp]	通过 chainweaver.mcp 适配器经由 MCP 暴露流	mcp
chainweaver[contrib]	导入精选的标准工具库（参见标准工具库）	（目前无需额外依赖）
chainweaver[langchain]	ChainWeaver 与 LangChain BaseTool 之间的双向适配器	langchain-core
chainweaver[llamaindex]	ChainWeaver 与 LlamaIndex FunctionTool 之间的双向适配器	llama-index-core
chainweaver[test]	基于 Hypothesis 的自定义流属性测试	hypothesis、hypothesis-jsonschema
chainweaver[docs]	使用 mkdocs 在本地构建文档站点	mkdocs、mkdocs-material、mkdocstrings
chainweaver[weaver-stack]	与真实 Weaver Stack 互操作——使用共享的路由／能力契约（weaver-spec #91、contextweaver #106、agent-kernel #89、#233）	weaver-contracts
chainweaver[integrations]	一次性安装上述所有集成额外依赖——组合 CI 会进行测试	上述集成行的依赖并集

维护者工具（pytest、ruff、mypy、nbmake 等）不是已发布的额外依赖：它们属于 PEP 735 依赖组，可通过 pip install -e ".[integrations]" --group dev (#550) 安装。[dev] 额外依赖已不再存在。

软件包元数据（pyproject.toml）提供了文档、源代码、更新日志和问题跟踪器的 URL，因此 pip show chainweaver 和 PyPI 侧边栏会将用户引导至正确位置。

---

快速开始

定义工具、构建流并执行

from pydantic import BaseModel
from chainweaver import Tool, Flow, FlowStep, FlowRegistry, FlowExecutor

# --- 1. Declare schemas ---

class NumberInput(BaseModel):
    number: int

class ValueOutput(BaseModel):
    value: int

class ValueInput(BaseModel):
    value: int

class FormattedOutput(BaseModel):
    result: str

# --- 2. Implement tool functions ---

def double_fn(inp: NumberInput) -> dict:
    return {"value": inp.number * 2}

def add_ten_fn(inp: ValueInput) -> dict:
    return {"value": inp.value + 10}

def format_result_fn(inp: ValueInput) -> dict:
    return {"result": f"Final value: {inp.value}"}

# --- 3. Wrap as Tool objects ---

double_tool = Tool(
    name="double",
    description="Takes a number and returns its double.",
    input_schema=NumberInput,
    output_schema=ValueOutput,
    fn=double_fn,
)

add_ten_tool = Tool(
    name="add_ten",
    description="Takes a value and returns value + 10.",
    input_schema=ValueInput,
    output_schema=ValueOutput,
    fn=add_ten_fn,
)

format_tool = Tool(
    name="format_result",
    description="Formats a numeric value into a human-readable string.",
    input_schema=ValueInput,
    output_schema=FormattedOutput,
    fn=format_result_fn,
)

# --- 4. Define the flow ---

flow = Flow(
    name="double_add_format",
    description="Doubles a number, adds 10, and formats the result.",
    steps=[
        FlowStep(tool_name="double",        input_mapping={"number": "number"}),
        FlowStep(tool_name="add_ten",       input_mapping={"value": "value"}),
        FlowStep(tool_name="format_result", input_mapping={"value": "value"}),
    ],
)

# --- 5. Execute ---

registry = FlowRegistry()
registry.register_flow(flow)

executor = FlowExecutor(registry=registry)
executor.register_tool(double_tool)
executor.register_tool(add_ten_tool)
executor.register_tool(format_tool)

result = executor.execute_flow("double_add_format", {"number": 5})

print(result.success)       # True
print(result.final_output)  # {'number': 5, 'value': 20, 'result': 'Final value: 20'}

for record in result.execution_log:
    print(record.step_index, record.tool_name, record.outputs)
# 0 double {'value': 10}
# 1 add_ten {'value': 20}
# 2 format_result {'result': 'Final value: 20'}

你也可以直接运行随附的示例：

python examples/simple_linear_flow.py   # simple arithmetic flow
python examples/etl_flow.py             # ETL flow: fetch → validate → normalize → enrich → store
python examples/mcp_search_flow.py      # MCP-style search → extract → format flow
python examples/naive_vs_compiled.py    # timing comparison: naive LLM calls vs ChainWeaver flow
python examples/coding_agent_pr_review.py    # deterministic PR-review checklist
python examples/coding_agent_changelog.py    # changelog generation workflow template
python examples/coding_agent_debug_log.py    # debug-log triage workflow template
python examples/mcp_style_before_after_demo.py        # before/after MCP-style flow demo
python examples/release_readiness_flow/release_readiness.py  # deterministic release-readiness gate
python examples/skdr_policy_eval_flow.py              # offline policy-evaluation workflow template
python examples/integrations/langgraph_node.py        # call a flow from a LangGraph node (needs chainweaver[langgraph])
python examples/integrations/openai_agents_tool.py    # expose a flow as an OpenAI Agents SDK tool (needs chainweaver[openai-agents])

托管文档还包含一份食谱，其中配有 examples/cookbook/ 下的脚本，以及框架示例和工作流模板（LangGraph、OpenAI Agents SDK、发布就绪性、策略评估）。

使用 @tool 装饰器

@tool 装饰器会通过内省类型提示自动生成输入架构，从而省去样板代码：

from pydantic import BaseModel
from chainweaver import tool, Flow, FlowStep, FlowRegistry, FlowExecutor

class ValueOutput(BaseModel):
    value: int

class FormattedOutput(BaseModel):
    result: str

@tool(description="Doubles a number.")
def double(number: int) -> ValueOutput:
    return {"value": number * 2}

@tool(description="Adds ten.")
def add_ten(value: int) -> ValueOutput:
    return {"value": value + 10}

@tool(description="Formats the result.")
def format_result(value: int) -> FormattedOutput:
    return {"result": f"Final value: {value}"}

flow = Flow(
    name="double_add_format",
    description="Doubles a number, adds 10, and formats.",
    steps=[
        FlowStep(tool_name="double",        input_mapping={"number": "number"}),
        FlowStep(tool_name="add_ten",       input_mapping={"value": "value"}),
        FlowStep(tool_name="format_result", input_mapping={"value": "value"}),
    ],
)

registry = FlowRegistry()
registry.register_flow(flow)

executor = FlowExecutor(registry=registry)
executor.register_tool(double)
executor.register_tool(add_ten)
executor.register_tool(format_result)

result = executor.execute_flow("double_add_format", {"number": 5})
print(result.final_output)  # {'number': 5, 'value': 20, 'result': 'Final value: 20'}

装饰后的工具也可以直接调用：

print(double(number=5))  # {'value': 10}

请参阅 examples/decorator_tool.py，查看可运行的前后对比示例。

使用 FlowBuilder

FlowBuilder 提供流畅、可链式调用的 API，是直接构造 Flow 对象的一种更符合 Python 风格的替代方式。它生成的 Flow 完全相同——这只是语法糖，而非替代方案：

from chainweaver import FlowBuilder

flow = (
    FlowBuilder("double_add_format", "Doubles a number, adds 10, and formats.")
    .step("double", number="number")
    .step("add_ten", value="value")
    .step("format_result", value="value")
    .build()
)

• .step(tool_name, **mapping) — 添加一个步骤；字符串值用于查找上下文键，非字符串值是字面常量；不传 kwargs 时会传递完整上下文。
• .step_from(flow_step) — 追加一个预构建的 FlowStep，用于互操作。
• .with_input_schema(Model) / .with_output_schema(Model) — 可选的流级 Pydantic 架构声明。
• .with_trigger(conditions) — 可选的自由格式触发器元数据。
• .build() — 返回经过验证的 Flow；如果缺少 name 或 description，则引发 FlowBuilderError。

---

交互式 Playground

想在本地不安装任何内容就试用 ChainWeaver？playground/ 目录提供了一个 Streamlit 应用，你可以选择预加载的流程、编辑其 JSON 输入、运行流程，并通过 Mermaid 图查看逐步执行轨迹，全程不调用大模型——使用的正是该库提供的同一个 FlowExecutor。

pip install -r playground/requirements.txt
streamlit run playground/app.py

该应用提供三个示例流程（算术、数据流和 MCP 风格的搜索），会生成可分享的 ?share=<token> 链接，让运行结果通过 URL 往返传递；同时完全无状态，因此可部署到 Streamlit Community Cloud，无需后端。有关本地运行和部署说明，请参阅 playground/README.md。

---

架构

chainweaver/
├── __init__.py       # Public API
├── builder.py        # FlowBuilder — fluent API for flow construction
├── compat.py         # schema_fingerprint, check_flow_compatibility
├── compiler.py       # compile_flow — static schema flow validation
├── decorators.py     # @tool decorator for zero-boilerplate tool definition
├── tools.py          # Tool — named callable with Pydantic schemas
├── flow.py           # FlowStep + Flow + FlowStatus — ordered step definitions
├── registry.py       # FlowRegistry — multi-version flow catalogue
├── executor.py       # FlowExecutor — deterministic, LLM-free runner
├── exceptions.py     # Typed exceptions with traceable context
└── log_utils.py      # Structured per-step logging

核心抽象

Tool

Tool(
    name="my_tool",
    description="...",
    input_schema=MyInputModel,   # Pydantic BaseModel
    output_schema=MyOutputModel, # Pydantic BaseModel
    fn=my_callable,
)

工具将普通 Python 可调用对象与 Pydantic 模型封装在一起，以进行严格的输入和输出验证。

FlowStep

FlowStep(
    tool_name="my_tool",
    input_mapping={
        "key_for_tool": "key_from_context",   # flat top-level lookup
        "city": "/user/address/city",         # RFC-6901 pointer into nested context
        "limit": 10,                          # non-string -> literal constant
    },
    output_mapping={"renamed": "value"},      # rename/prune outputs before merge
)

input_mapping 将累积执行上下文中的键映射到工具的输入架构。字符串值会在上下文中查找——普通键用于顶层查找，以 / 开头的字符串则表示指向嵌套上下文的 RFC-6901 JSON 指针（#387）；非字符串值则是字面常量。

output_mapping（#386）可选，用于在工具输出合并到上下文之前对其重命名并筛选：{context_key: output_key} 仅保留所列的输出键，并逐一重命名。不设置该字段时，所有输出键都会原样合并。

要注入绝不能出现在模型可见架构中的单次请求密钥（身份验证令牌、账号），请在执行时传入，而不要放在 initial_input 中：

result = executor.execute_flow(
    "account_overview",
    {"query": "what's my balance?"},          # LLM-visible
    dynamic_params={"billingAccountNumber": "1.60007029"},  # hidden, injected (#316)
)

Flow

Flow(
    name="my_flow",
    version="0.1.0",             # SemVer string; defaults to "0.1.0" if omitted
    description="...",
    steps=[step_a, step_b, step_c],
    deterministic=True,          # metadata annotation; executor is always LLM-free
    trigger_conditions={"intent": "process data"},  # optional metadata
)

由步骤按顺序组成的序列。完整字段表（status、tool_schema_hashes，以及带有已解析属性访问器的 input_schema_ref / output_schema_ref 字符串字段）请参阅 AGENTS.md §5。

FlowStep 只能运行工具（tool_name）或已注册的子流程（flow_name）之一，二者必须且只能选其一。引用子流程可组合可复用的流程（issue #75）：

fetch_validate = Flow(
    name="fetch_validate",
    description="Fetch and validate.",
    steps=[
        FlowStep(tool_name="fetch", input_mapping={"url": "url"}),
        FlowStep(tool_name="validate", input_mapping={"data": "data"}),
    ],
)
fetch_then_transform = Flow(
    name="fetch_then_transform",
    description="Reuse fetch_validate, then transform.",
    steps=[
        FlowStep(flow_name="fetch_validate", input_mapping={"url": "url"}),  # sub-flow
        FlowStep(tool_name="transform", input_mapping={"data": "data"}),
    ],
)

执行器会使用该步骤解析后的输入运行子流程，将其输出合并回父上下文，并将子流程的 ExecutionResult 附加到父流程的 StepRecord.sub_result。执行前会检查子流程引用是否存在循环，以及是否超出可配置的最大嵌套深度（FlowExecutor(max_composition_depth=...)，默认为 10）；否则会引发 FlowCompositionError。

传给 execute_flow 的 deadline 或 CancellationToken 会转发到组合子流程，因此在子流程内部的步骤边界也会检查取消状态和墙钟时间预算，而不只是父流程边界。长时间运行的子流程会在自身步骤之间停止。成本报告中的 steps_executed 统计组合步骤实际驱动的工具调用次数（递归计算），因此 llm_calls_avoided 会反映整个组合过程中运行的所有工具。

FlowRegistry

registry = FlowRegistry()
registry.register_flow(flow)
registry.get_flow("my_flow")
registry.list_flows()
registry.match_flow_by_intent("process data")  # basic substring match

内存中的流程目录。

FlowExecutor

executor = FlowExecutor(registry=registry)
executor.register_tool(tool_a)
result = executor.execute_flow("my_flow", {"key": "value"})

# Version-targeted execution: run an exact registered version instead of the
# latest. Omitting `version` keeps the default (latest) behaviour. The version
# that actually ran is always recorded on `result.flow_version`, so routing,
# audit, and replay can correlate a result with the precise flow definition.
result = executor.execute_flow("my_flow", {"key": "value"}, version="1.2.0")
assert result.flow_version == "1.2.0"

逐步运行流程，并进行完整的架构验证和结构化日志记录。全程不会调用大模型。

ChainAnalyzer

from chainweaver import ChainAnalyzer, ToolChain

analyzer = ChainAnalyzer(tools=[tool_a, tool_b, tool_c])

# All schema-compatible pairs
matrix: dict[str, list[str]] = analyzer.compatibility_matrix()

# All valid tool sequences up to length 3
chains: list[ToolChain] = analyzer.find_chains(max_depth=3)

# Filter by start or end tool
chains = analyzer.find_chains(max_depth=3, start="tool_a", end="tool_c")

# Promote chains to ready-to-register Flow objects
flows = analyzer.suggest_flows(max_depth=3, min_depth=2)

在注册或执行任何流程之前，离线发现架构兼容的工具组合。compatibility_matrix() 会检查消费者工具的每个必需输入字段是否都出现在生产者工具的输出中，且类型匹配。suggest_flows() 会按名称匹配自动配置 input_mapping，并返回可直接传给 FlowRegistry.register_flow() 的 Flow 对象。

数据流

initial_input (dict)
       │
       ▼
 ┌─────────────────────────────────────────────┐
 │  Execution context (cumulative dict)        │
 │                                             │
 │  Step 0: resolve inputs → run tool → merge  │
 │  Step 1: resolve inputs → run tool → merge  │
 │  Step N: resolve inputs → run tool → merge  │
 └─────────────────────────────────────────────┘
       │
       ▼
 ExecutionResult.final_output (merged context)

---

MCP 集成概念

ChainWeaver 可以介于 Agent/工具观察与确定性执行之间：

Agent / tool traces
   │  (observe repeated paths)
   ▼
Candidate analysis + human review
   │  (prove/reject; preserve security boundaries)
   ▼
Governed deterministic capability
   │  (FlowExecutor and/or supported export)
   ▼
MCP / host-framework invocation

MCP 是互操作层，而不是产品类别。当前运行时可以将经过审核的流程作为 MCP 工具公开；与此同时，#555 正在探索是否可以通过可移植输出，让同一项获批能力在其他宿主中执行，而不要求 ChainWeaver 拥有运行时。

ChainWeaver 是供你嵌入的库，而不是负责管理你的跟踪存储、身份系统或企业授权控制面的运行时。宿主作者应阅读 docs/runtime-responsibilities.md。

---

集成

ChainWeaver 可接入 MCP 生态系统和主流 Agent 框架。现有集成仍受支持；相比独立产品验证，拓展新的适配器类型目前优先级较低。

集成	功能	入口
MCP 服务（出站）	将你的流程公开为 MCP 工具 — Agent 可将整个编译后的流程作为一个确定性工具调用	chainweaver serve · 指南 · FlowServer
MCP 适配器（入站）	将 MCP 服务公布的工具封装为 ChainWeaver Tool	chainweaver.mcp.MCPToolAdapter
LangGraph	从 LangGraph 节点调用流程	示例 · examples/integrations/langgraph_node.py
OpenAI Agents SDK	将流程公开为 Agents SDK FunctionTool	示例 · examples/integrations/openai_agents_tool.py
LangChain / LlamaIndex	双向工具桥接	chainweaver.integrations.{langchain,llamaindex}（见下文）
OpenCode	观察工具运行、挖掘宏流程，并将经过审核的流程重新公开为 MCP 工具	示例 · chainweaver opencode
Claude Code	捕获 PostToolUse 钩子跟踪、挖掘宏流程，并将经过审核的流程重新公开为 MCP 工具	示例 · chainweaver claude
VS Code / Copilot	捕获 MCP 工具跟踪（Copilot OTel），并通过 .vscode/mcp.json 公开经过审核的流程	示例 · chainweaver vscode
GitHub Action	在 CI 中验证 .flow.yaml / .flow.json 文件，并以内联批注标注 PR	.github/actions/chainweaver · 指南

安装所需的扩展：pip install 'chainweaver[mcp]'（或 langgraph、openai-agents、langchain、llamaindex）。未安装相应扩展就导入任何集成时，会引发明确的 ImportError。

如果想在 MCP 注册表、awesome 列表或框架目录中发布或收录 ChainWeaver，请参阅 docs/distribution.md。广泛分发有意暂缓，待命名决策（#556）和验证证据（#553）明确后再进行。

---

错误处理

所有错误都有明确类型且可追溯：

异常	引发时机
ToolNotFoundError	某个步骤引用了未注册的工具
FlowNotFoundError	请求的工作流未注册
FlowAlreadyExistsError	注册已存在的工作流（未设置 overwrite=True）
FlowStatusError	执行状态不是 ACTIVE 的工作流（未设置 force=True）
FlowCancelledError	deadline 已过期，或 CancellationToken 在步骤边界处被取消（携带部分结果）
InvalidFlowVersionError	工作流注册时使用了不符合 PEP 440 的版本字符串
FlowSerializationError	工作流文件（YAML/JSON）格式错误、含有未知判别符，或引用了无法解析的类
SchemaValidationError	输入或输出未通过 Pydantic 验证
InputMappingError	映射键不存在于上下文中
FlowExecutionError	工具可调用对象引发了意外异常
ApprovalDeniedError	执行时审批回调拒绝了步骤、引发异常或返回了无效值；或者 strict_safety=True，且需要审批的步骤没有回调
SafetyCeilingError	某个步骤的 ToolSafetyContract.side_effects 超出了执行器配置的 max_side_effect_level
GuardrailViolationError	已注册的 guardrail_callback 在输入阶段阻止了某个步骤（内容安全／注入检查）
ToolDefinitionError	@tool 装饰器无法根据函数构建工具
DAGDefinitionError	DAGFlow 存在环、重复的 step_id 或未知依赖项
FlowCompositionError	组合工作流存在子工作流环、超过 max_composition_depth，或引用了未注册的子工作流
ToolTimeoutError	设置了 timeout_seconds 的 Tool 超出了配置的墙钟时间上限
ToolOutputSizeError	设置了 max_output_size 的 Tool 返回的输出超过了配置上限
FlowBuilderError	调用 FlowBuilder.build() 时未提供名称或描述
AttestationInputError	证明输入生成器无法为某个架构字段合成值
PluginDiscoveryError	严格模式插件发现（discover_tools(strict=True) / discover_flows(strict=True)）遇到异常的入口点加载器
ContribError	chainweaver.contrib.tools 中的工具违反了约定（缺少 JSON 指针键、谓词形状错误、断言不匹配）
FixtureStaleError	record_then_replay 重放调用无法匹配到记录（fixture 缺失或已过期）
FuzzConfigError	基于属性的模糊测试运行配置错误（没有属性、runs < 1、工作流没有 input_schema 且没有基础输入，或输入字段类型不受支持）
CostProfileError	请求估算不在维护的 PROVIDER_PRICES 表中的 (provider, model) 对的成本
MCPMetadataError	服务端提供的 MCP 工具名称不符合适配器的 MetadataPolicy（且 on_invalid_name="error"）
MCPSchemaDriftError	在 MCPToolAdapter(on_drift="error") 下，已固定的 MCP 工具原始架构发生了变化
FlowAuthenticationError	网络公开的 FlowServer 身份验证器返回了 None 或引发异常；调用会在分发前被拒绝
RateLimitExceededError	FlowServer 限流器拒绝了调用
FlowAuthorizationError	FlowServer 授权回调拒绝了调用（仅携带客户端安全的 reason_code）
CheckpointVersionError	恢复的快照的 snapshot_version 与运行中的库存在不兼容的主版本差异

所有异常都继承自 ChainWeaverError，并带有稳定的诊断 code（例如 CW-E006）；CLI 会在错误输出中为其添加前缀，失败的 StepRecord 会将其公开为 error_code。完整代码表请参阅 docs/reference/error-table.md。

---

标准工具库

chainweaver.contrib.tools 提供一组精选的确定性工具，让新用户无需编写任何 Tool 样板代码，就能在第一个下午组合出有意义的流程。

from chainweaver.contrib.tools import (
    assert_equal,
    filter_list,
    json_pluck,
    json_set,
    map_list,
    passthrough,
)

工具	用途
passthrough	恒等操作——原样返回上下文。
json_pluck	按 RFC-6901 JSON 指针提取一个值。
json_set	按 RFC-6901 JSON 指针设置一个值；返回一个新字典。
assert_equal	当两个上下文键的值不同时，抛出 ContribError。
map_list	对列表中的每个元素应用一个已注册的子流程。
filter_list	丢弃谓词子流程返回假值的元素。

该库仅支持确定性操作：不涉及 HTTP、文件 I/O、数据库访问、随机数生成或时钟。任何有状态操作都应放在用户代码中。使用 pip install 'chainweaver[contrib]' 安装。

可运行示例：examples/contrib_pluck_and_set.py、examples/contrib_map_filter.py。

---

成本节省报告

朴素 Agent 委托给大模型的每一次步骤间转移，都是 ChainWeaver 能够消除的一次路由调用。CostProfile / CostReport 可将其换算为美元估算；维护中的 PROVIDER_PRICES 表（带日期的快照，不进行实时 HTTP 查询）可用于按真实模型计算价格：

from chainweaver.cost import compute_cost_report

# Build a profile straight from the maintained price table.
report = compute_cost_report(
    steps_executed=6,                 # a six-tool flow
    actual_execution_ms=4.2,
    provider="anthropic",
    model="claude-opus-4-7",
)
print(report)

Cost Avoided Report (estimate)
──────────────────────────────
Steps executed:          6
LLM calls avoided:       5
Est. latency saved:      1500.0ms
Est. cost saved:         $0.1688
Actual execution time:   4.2ms
Priced against:          anthropic/claude-opus-4-7 (as of 2026-05-01)

根据该表生成的每份报告都会包含快照的 as_of 日期，以便识别过期价格。未知的 (provider, model) 组合会抛出 CostProfileError，而不是进行猜测。如果你有更准确的单次调用数据，可以传入显式的 profile=CostProfile(...)；也可以在 FlowExecutor 上设置 cost_profile=，为每个 ExecutionResult 附加报告。价格通过维护者审阅的 PR 刷新（.github/workflows/update-prices.yml），绝不会自动合并。

除非输入来自观测到的跟踪测量，否则这些报告都只是估算值。不得将其作为应当编译某个候选方案的证据；#377 跟踪对假设的模型中介与实测结果进行校准的工作。

---

导出适配器

通过 chainweaver.export，将已编译的流程交给任何外部 Agent 框架：

from chainweaver.export import (
    flow_to_anthropic_tool,
    flow_to_callable,
    flow_to_openai_function,
)

openai_spec = flow_to_openai_function(flow, executor)
anthropic_spec = flow_to_anthropic_tool(flow, executor)
run = flow_to_callable(flow, executor)  # plain dict → dict callable

flow_to_openai_function 会生成 OpenAI 聊天 / responses API 所需的 {"type": "function", "function": {…}} 格式。flow_to_anthropic_tool 会生成 Anthropic 的 tool_use 格式。flow_to_callable 会将流程包装为 Callable[[dict], dict]，适用于任何接受任意 Python 可调用对象的框架。

这些适配器都不会导入 openai 或 anthropic——它们只会生成字典和可调用对象。与这些客户端的运行时集成由调用方负责。

可运行示例：examples/export_openai_anthropic.py。

---

生态系统桥接（LangChain、LlamaIndex）

chainweaver.integrations.langchain 和 chainweaver.integrations.llamaindex 提供轻量级双向适配器，可将现有的 LangChain BaseTool / LlamaIndex FunctionTool 实例导入 ChainWeaver，也可将 ChainWeaver Tool 实例导出。

from chainweaver.integrations.langchain import (
    from_langchain_tool,
    to_langchain_tool,
)

cw_tool = from_langchain_tool(my_langchain_tool)
lc_tool = to_langchain_tool(my_cw_tool)

使用 pip install 'chainweaver[langchain]' / 'chainweaver[llamaindex]' 安装。未安装相应扩展时导入任一模块，都会引发清晰的 ImportError。

---

插件发现

对于第三方包——chainweaver-aws、chainweaver-stripe 等——ChainWeaver 遵循与 pytest、Sphinx、MkDocs 等相同的入口点约定。

发布方（pyproject.toml）：

[project.entry-points."chainweaver.tools"]
aws = "chainweaver_aws:get_tools"

[project.entry-points."chainweaver.flows"]
aws = "chainweaver_aws:get_flows"

使用方：

from chainweaver import FlowExecutor, FlowRegistry

# Auto-register every tool / flow advertised by an installed plugin.
registry = FlowRegistry(discover_plugins=True)
executor = FlowExecutor(registry=registry, discover_plugins=True)

发现操作默认关闭——导入 chainweaver 不会触发插件导入。行为异常的插件（导入时引发异常或返回错误类型）会记录为 WARNING 并跳过；传入 strict=True 调用 discover_tools() / discover_flows()，即可启用严格模式。

可运行示例：examples/plugin_discovery.py。

---

运行时学习

ChainWeaver 可以观察 Agent 的操作，并为重复路径提出确定性流程候选项。重复序列并不能证明该路径适合或值得编译；候选项需要经过审核，产品验证计划也明确会衡量误报、漏报和有价值的拒绝。

from chainweaver import ChainObserver, FlowRegistry

observer = ChainObserver()

# Record tool calls as the agent makes them.
observer.record("fetch", {"url": "..."}, {"body": "..."})
observer.record("validate", {"body": "..."}, {"valid": True})
observer.record("transform", {"body": "..."}, {"records": [1, 2, 3]})
observer.end_trace()
# ... many traces later ...

registry = FlowRegistry()
for suggestion in observer.suggest_flows(min_occurrences=3):
    # Suggestions are proposals — review; never treat confidence as authorization.
    print(suggestion.flow.name, suggestion.confidence,
          suggestion.estimated_llm_calls_avoided)
    registry.register_flow(suggestion.flow)

• ChainObserver (#78) 从运行时跟踪记录中挖掘重复的工具序列，并生成排名后的 FlowSuggestion — 从不自动注册。
• chainweaver record (#226) 挖掘录制的 JSONL 跟踪记录，并写入候选流程文件，以供明确审核和晋升。
• ChainWeaverService (#101) 将观察、静态分析和可选的离线候选提议整合为 分析 → 提议 → 治理 → 晋升 循环。

在将候选建议视为某条路径应转为确定性流程的证明之前，请参阅产品验证与采用门槛。

---

路线图

当前路线图以验证为先，而非以功能数量为先。最新发布的版本是 v0.14.1；main 上的较新工作尚未发布，直至后续版本发布。

优先级	工作	原因
P0	#553 独立产品证伪	确定基于跟踪记录的发现与治理是否优于人工检查 + 普通函数基准。
P0	#554 保留授权/审批	编译可能移除推理边界，但绝不能悄然移除安全边界。
P0	#522 稳定/支持/实验性 API 分级	使兼容性承诺的范围小于实现范围。
P0	#519 发布一致性	源码、软件包、标签、发行版本、文档和构建产物必须保持一致。
P1	#527 隐私配置	跟踪记录分析必须能使用最少化/本地证据运行。
以 #553 为门槛	#334 规范证据架构	只有用户验证这项工作后，才构建大型生命周期模型。
以 #553 为门槛	#498 生产环境黄金路径	将已验证的需求转化为一种规范的端到端验证。
按需求探索	#555 可移植的编译后能力	如果用户重视分析但不重视 FlowExecutor，则让认可的构建产物具备可移植性。
广泛分发前	#556 命名/搜索决策	趁迁移成本仍低时解决可发现性/歧义问题。

在这些门槛产生证据之前，面向广泛目录的提交、托管式 Playground 投入和额外扩展框架适配器都刻意列为较低优先级。

v1.0.0 也受证据门槛约束：根据 docs/v1-release-criteria.md，必须具备独立工作负载/采用者、人工基准、外部安全审查、重复使用、下游集成，以及为期 30 天的 RC 兼容性浸泡测试。

---

命令行界面

ChainWeaver 附带一个 chainweaver 控制台脚本，包含以下子命令。 读取 .flow.yaml 文件需要安装 YAML 扩展 （pip install 'chainweaver[yaml]' — 也列在安装中）。 下面的 run 示例使用了 examples/ 下附带的流程，因此应从仓库根目录 运行。

# Run a flow from disk — no Python required.
chainweaver run examples/double_add_format.flow.yaml \
    --tools examples.simple_linear_flow \
    --input '{"number": 5}'

# Serve a flow as MCP tools (needs chainweaver[mcp]) — agents call the whole
# compiled flow as one deterministic tool. See docs/mcp-server.md.
chainweaver serve examples/double_add_format.flow.yaml \
    --tools examples.simple_linear_flow

# Validate a flow file (used by CI gates and editor tooling).
chainweaver validate flows/etl.flow.yaml
chainweaver check flows/                  # whole-directory variant

# Scaffold a runnable first flow project (tools + flow file + run script).
chainweaver init my-first-flow --template linear --with-tests

# Render a flow as ASCII, Graphviz DOT, or Mermaid. Discover it from a directory
# of flow files, an installed package's entry points, or the default registry.
chainweaver viz my_flow --discover-dir flows/ --format dot | dot -Tpng -o my_flow.png
chainweaver viz my_flow --discover-dir flows/ --format mermaid
chainweaver viz --result trace.json --format mermaid   # overlay a real run

# Explain a flow deterministically (LLM-free) for review — paste into a PR.
chainweaver explain my_flow --discover-dir flows/ > flow-review.md

# Inspect a flow's structure (table or JSON). `flows list` previews what is
# discoverable so you can see what `inspect`/`viz` can target.
chainweaver inspect my_flow --discover-dir flows/ --format json
chainweaver flows list --discover-dir flows/

# Check that your environment is ready before running anything.
chainweaver doctor flow --profile first-run

# Inspect a coding-agent workspace's MCP / observe setup (read-only).
chainweaver doctor vscode --workspace .

# Install tab-completion for your shell (bash/zsh/fish).
chainweaver --install-completion

# Analyze ExecutionResult traces — bottlenecks, p50/p95/p99 across runs,
# and per-step / per-tool retry / skip / fallback / failure aggregates.
chainweaver profile trace_a.json trace_b.json --format json

# Compare two ExecutionResult JSON files step-by-step.
chainweaver diff baseline.json current.json --perf-tolerance 25

# Observed-determinism attestation: run N inputs × M repeats.
chainweaver attest flows/etl.flow.yaml --tools my_pkg.tools --runs 50 --repeats 3

# Advisory optimization suggestions for a saved flow.
chainweaver suggest flows/etl.flow.yaml --tools my_pkg.tools --trace trace_a.json

# Mine candidate flows from a recorded JSONL tool trace (offline, no LLM).
chainweaver record examples/agent_tool_trace.jsonl --output-dir candidates/
chainweaver flows promote candidates/suggested__fetch__validate.flow.yaml --to reviewed
chainweaver flows promote candidates/suggested__fetch__validate.flow.yaml --to active

# Run one continuous-analysis service pass and report flow proposals.
chainweaver service --tools my_pkg.tools --trace trace.jsonl

# Check saved flows for tool schema drift against the live registry.
chainweaver doctor flow flows/ --check-drift --tools my_pkg.tools

# Property-based fuzzing: generate cases, check invariants, save/minimize failures.
chainweaver fuzz flows/etl.flow.yaml --tools my_pkg.tools \
  --property my_pkg.props:no_unauthorized_action --runs 1000 --seed 42 \
  --minimize --save-failures failures/

run 是从全新安装到看到流程执行的最快途径： 指定 .flow.yaml/.flow.json 文件，传入 --tools <module>（一个 Python 模块的导入路径，该模块在顶层公开 Tool 实例），并以 JSON 格式提供初始输入。手动编写的流程文件必须在顶部声明 type: Flow（或 type: DAGFlow）判别字段 — 请参阅 流程文件格式参考。大多数报告类子命令也接受 --format json，用于机器处理 （inspect、validate、check、run、profile、diff、attest、 suggest、doctor）；例外情况包括使用 --format ascii|dot|mermaid 的 viz、使用 --format md|text 的 explain，以及写入原始 JSON Schema、 没有 --format 标志的 dump-schema。生成结果的命令（inspect、 validate、check、profile、diff、attest）会将 --format json 输出封装在稳定且有版本的信封中 （{"schema_version", "status", "data", "errors"}），以便自动化程序根据 status / 错误代码分支处理 — 请参阅 机器可读输出。 所有子命令遵循相同的退出码约定（0 成功，1 业务逻辑错误，2 文件未找到 / 参数错误），CLI 还提供 制表符补全（chainweaver --install-completion）。

inspect 和 viz 可从磁盘或注册表解析流程。 传入 --file <path>、--discover-dir <dir> 或 --discover-entry-points， 即可无需编写任何 Python 来解析流程（issue #381）；chainweaver flows list 可预览可发现的内容。如果没有发现标志，它们会回退到以进程为作用域、 通过编程方式安装的内存注册表 — 在没有标志且未配置注册表的情况下运行 chainweaver inspect my_flow，会以退出码 1 退出，并显示 No registry configured. Call chainweaver.cli.set_default_registry(...) before invoking the CLI.。要接入默认注册表路径，请使用一个小型入口脚本：

# my_cli_entry.py
from chainweaver import FlowRegistry
from chainweaver.cli import main, set_default_registry
from my_app import build_registry  # returns a populated FlowRegistry

set_default_registry(build_registry())
main()

完整模式请参阅 docs/cli.md § 编程式注册， 其中也说明了为何采用这种拆分（面向文件的命令保持零配置， 面向注册表的命令则便于自省）。

---

开发

新贡献者：请参阅 CONTRIBUTING.md 中的你的首次贡献， 了解 good-first-issue / good-first-ai-issue 入门指南，以及首次提交 PR 的 分步流程。

# Install with the integration extras and the maintainer tooling group
pip install --upgrade pip          # --group needs pip >= 25.1
pip install -e ".[integrations]" --group dev

# Run tests
python -m pytest tests/ -v

# Run the examples
python examples/simple_linear_flow.py   # simple arithmetic flow
python examples/etl_flow.py             # ETL flow
python examples/mcp_search_flow.py      # MCP-style search & summarize flow
python examples/naive_vs_compiled.py    # naive vs compiled timing comparison
python examples/coding_agent_pr_review.py
python examples/coding_agent_changelog.py
python examples/coding_agent_debug_log.py

---

许可证

本项目采用 Apache License 2.0 许可 - 详情请参阅 LICENSE 文件。