Skip to content

README.zh.md

Latest commit

 

History

History
281 lines (206 loc) · 11.2 KB

README.zh.md

File metadata and controls

281 lines (206 loc) · 11.2 KB

KWeaver Python SDK

KWeaver 平台的 Python SDK — 支持 BKN（Business Knowledge Network）、Vega 数据管理和 Decision Agent。

注意： CLI 由 TypeScript 包 提供。本包仅为 SDK。

English

安装

pip install kweaver-sdk

需要 Python >= 3.10。

开发者指南

使用说明（安装、认证、KWeaverClient、分页与排障）：

从源码 docstring 生成 API HTML（在仓库根执行）：

make -C packages/python docs-python

在浏览器打开 docs/reference/python-api-html/index.html。

若只用 pdoc kweaver 启动文档，像 kweaver.resources.dataflow_v2.DataflowV2Resource 这类注解会变成纯文本、无法跳转 —— pdoc 需要把各子模块都写在命令行里。make docs-python 已通过 scripts/list_pdoc_modules.py 自动带上全部包内模块。

与上面相同的模块列表起本地服务（在 packages/python 下执行）：

PYTHONPATH=src uv run --extra docs sh -c 'exec python -m pdoc -d google -h 127.0.0.1 -p 8765 $(PYTHONPATH=src uv run python scripts/list_pdoc_modules.py)'

快速上手

搜索与对话（最简路径）

import kweaver

kweaver.configure(
    url="https://kweaver.example.com",
    token="my-token",
    bkn_id="supply-chain-bkn-id",
    agent_id="supply-chain-agent-id",
)

# 语义搜索
results = kweaver.search("供应链有哪些关键风险？")
for concept in results.concepts:
    print(concept.concept_name, concept.rerank_score)

# 与 Agent 对话
reply = kweaver.chat("帮我分析一下今年的库存风险")
print(reply.content)

# 流式输出
for chunk in kweaver.chat("给我生成一份风险报告", stream=True):
    print(chunk.delta, end="", flush=True)

无鉴权实例

若平台无 API 鉴权，使用 configure(..., auth=False) 或 KWeaverClient(..., auth=NoAuth())，与 TS CLI kweaver auth <url> --no-auth 写入的 ~/.kweaver/ 哨兵一致（ConfigAuth 在保存的 access token 为 __NO_AUTH__ 时不发送 Authorization）。

from kweaver import KWeaverClient, NoAuth

client = KWeaverClient(base_url="http://localhost:8080", auth=NoAuth())
# 或: kweaver.configure("http://localhost:8080", auth=False)

纯 Python 登录

与 TypeScript CLI 的 --http-signin 一致：RSA 加密密码、/oauth2/signin、换票，无需 Playwright 与 Node CLI。

import kweaver

kweaver.login(
    "https://kweaver.example.com",
    username="alice@example.com",
    password="secret",
    # new_password="..."  # 若服务端返回 401001017（首次改密）
)

kweaver.configure(url=..., username=..., password=...) 使用 HttpSigninAuth，在首次调用 API 时再完成登录。

Client API（完全控制）

from kweaver import KWeaverClient, ConfigAuth

client = KWeaverClient(
    auth=ConfigAuth(),           # 读取 ~/.kweaver/ 凭证
    debug=True,                  # 打印请求/响应诊断
    vega_url="http://vega:13014", # 可选：连接 Vega
)

# BKN — 知识网络
kns = client.knowledge_networks.list()
report = client.knowledge_networks.inspect("kn-123")  # 一站式诊断

# BKN — Schema 管理
ots = client.object_types.list("kn-123")
ot = client.object_types.get("kn-123", "ot-456")      # 含完整 data_properties
cgs = client.concept_groups.list("kn-123")
jobs = client.jobs.list("kn-123")

# Vega — 数据平台
catalogs = client.vega.catalogs.list()
resources = client.vega.resources.list(catalog_id="cat-1", category="table")
models = client.vega.metric_models.list()

# Vega — 查询
result = client.vega.query.dsl(body={"query": {"match_all": {}}, "size": 10})
result = client.vega.query.execute(tables=[...], output_fields=["*"], limit=20)
# 直连 SQL 或 OpenSearch DSL — POST /api/vega-backend/v1/resources/query
# 使用 {{resource_id}} 占位符以路由到正确的 catalog connector
rows = client.vega.query.sql_query({"query": "SELECT * FROM {{<resource-id>}} LIMIT 5", "resource_type": "mysql"})

# Vega — 诊断
info = client.vega.health()
report = client.vega.inspect()

可观测能力

# Debug 模式 — 打印完整 HTTP 诊断 + curl 命令到 stderr
client = KWeaverClient(auth=ConfigAuth(), debug=True)

# Dry-run — 拦截写操作，不实际发送到服务端
client = KWeaverClient(auth=ConfigAuth(), dry_run=True)

SDK 资源一览

BKN（知识网络）

资源 访问方式 方法
知识网络 client.knowledge_networks list, get, create, update, delete, build, export, inspect
对象类 client.object_types list, get, create, update, delete
BKN 指标（定义，bkn-backend） client.metrics list, create, search, validate, get, update, delete
BKN 指标数据 / 试算（ontology-query） client.metric_query data, dry_run
关系类 client.relation_types list, get, create, update, delete
Action 类 client.action_types list, execute, cancel
概念组 client.concept_groups list, get, create, update, delete, add_members, remove_members
任务 client.jobs list, get_tasks, delete, wait
查询 client.query semantic_search, instances, instances_iter, kn_search（Schema 发现已不推荐）, kn_schema_search（deprecated）, subgraph
Agent client.agents list, get, get_by_key, create, update, delete, publish, unpublish
对话 client.conversations send_message, list_messages
Dataflow（旧生命周期接口） client.dataflows create, run, poll, delete, execute
Dataflow v2 client.dataflow_v2 list_dataflows, run_dataflow_with_file, run_dataflow_with_remote_url, list_dataflow_runs, get_dataflow_logs_page
Resources（vega-backend） client.resources create, list, get, delete, find_by_table, query
Skill client.skills list, market, get, get_market, register_content, register_zip, update_status, update_metadata, update_package_content, update_package_zip, history, republish_history, publish_history, content, read_file, download, install
模型工厂（mf-model-manager / mf-model-api） client.models llm / small: list, get, add, edit, delete, test；invocation: chat, embedding, embeddings, rerank

大模型列表 client.models.llm.list(model_type=...) 会将 model_type 传给 mf-model-manager，取值仅 llm（文本）、rlm（推理）、vu（视觉/多模态）。

Context Loader MCP 可直接使用 ContextLoaderResource：

from kweaver.resources import ContextLoaderResource

cl = ContextLoaderResource(base_url, token, kn_id="kn_01")
schema = cl.search_schema("利润率", search_scope={"concept_groups": ["finance"]})
raw = cl.call_tool("search_schema", {"query": "利润率"})

search_schema() 是 Context Loader MCP search_schema 的类型化封装，默认 response_format 为 json，支持 query、response_format、search_scope、max_concepts、schema_brief、enable_rerank。search_scope.concept_groups 用于按 BKN 概念分组 ID 限定 Schema 发现范围，不是实例数据过滤条件。解析后的返回结果可能包含 object_types、relation_types、action_types、metric_types。

client.query.kn_search() / client.query.kn_schema_search() 是 Schema 发现的 deprecated 兼容/legacy 入口。新接入请使用 ContextLoaderResource.search_schema()。

需要直接使用 MCP 原生 tools/call 时，使用 call_tool(name, args)。这适用于服务端新增工具但 SDK 尚未提供类型化封装的场景，参数会原样透传。

自签名证书的开发环境可传入 tls_insecure=True。

Vega（数据平台）

资源 访问方式 方法
Catalog client.vega.catalogs list, get, health_status, health_report, test_connection, discover, resources
数据资源 client.vega.resources list, get, data, preview
连接器类型 client.vega.connector_types list, get
指标模型 client.vega.metric_models list, get
事件模型 client.vega.event_models list, get
调用链模型 client.vega.trace_models list, get
数据视图 client.vega.data_views list, get
数据字典 client.vega.data_dicts list, get
目标模型 client.vega.objective_models list, get
查询 client.vega.query execute, sql_query, dsl, dsl_count, promql, promql_instant, events
任务 client.vega.tasks list_discover, get_discover, wait_discover, get_metric
命名空间 client.vega health, stats, inspect

配置

KWeaverClient(
    base_url="https://...",          # KWeaver 平台 URL
    auth=ConfigAuth(),               # 或 TokenAuth("...") 或 HttpSigninAuth(url, username=..., password=...)
    vega_url="http://vega:13014",    # 可选：Vega 数据平台 URL
    mf_model_manager_base_url=None,  # 可选：mf-model-manager 的 origin（未设则用环境变量或平台 base_url）
    mf_model_api_base_url=None,      # 可选：mf-model-api 的 origin（未设则用环境变量或平台 base_url）
    debug=False,                     # 打印请求/响应诊断
    dry_run=False,                   # 拦截写操作
    tls_insecure=False,              # 跳过 TLS 校验（仅开发/自签名；生产请用受信任证书）
)
环境变量 说明
KWEAVER_BASE_URL 平台 URL
KWEAVER_TOKEN Bearer Token
KWEAVER_MF_MODEL_MANAGER_URL 可选：mf-model-manager 的根地址（仍追加路径 /api/mf-model-manager/v1）
KWEAVER_MF_MODEL_API_URL 可选：mf-model-api 的根地址（仍追加路径 /api/mf-model-api/v1）
KWEAVER_VEGA_URL Vega 后端 URL
KWEAVER_DEBUG 启用 debug 模式（true）
KWEAVER_FORMAT CLI 输出格式（md/json/yaml）
KWEAVER_TLS_INSECURE 设为 1 或 true 时跳过 TLS 校验（仅开发；kweaver auth … --insecure 会按平台写入 token.json）

Dataflow v2 示例

from kweaver import KWeaverClient, ConfigAuth

client = KWeaverClient(auth=ConfigAuth())

flows = client.dataflow_v2.list_dataflows()

file_run = client.dataflow_v2.run_dataflow_with_file(
    "dag-id",
    file_path="./demo.pdf",
)

remote_run = client.dataflow_v2.run_dataflow_with_remote_url(
    "dag-id",
    url="https://example.com/demo.pdf",
    name="demo.pdf",
)

runs = client.dataflow_v2.list_dataflow_runs(
    "dag-id",
    limit=20,
    sort_by="started_at",
    order="desc",
)

logs = client.dataflow_v2.get_dataflow_logs_page(
    "dag-id",
    remote_run["dag_instance_id"],
    page=0,
    limit=10,
)

Python 包当前仍然是 SDK-only；dataflow CLI 命令由 TypeScript 包提供。

相关链接

许可证

MIT