tps-rck-sdk 0.1.1

Official Python SDK for the Relational Calculate Kernel (RCK)

Topos Pathless Service RCK Python SDK 完整教程

🎯 RCK：Pathless 中的非传统计算引擎

RCK (Relational Calculate Kernel) 是 Pathless 服务中的核心组件，专门负责非传统、非确定性计算任务的完整生命周期，包括：

• 编译：将自然语言需求转换为可执行的计算逻辑
• 部署：自动部署到云原生环境，无需手动干预
• 运行：执行非确定性计算任务，支持动态调整
• 监控：实时监控计算过程和结果质量

🎯 Pathless 核心理念：目标驱动，路径无关

传统编程思维，开发者需要编写具体的算法步骤和实现细节：

def process_data(input_data):
    step1 = parse(input_data)
    step2 = validate(step1)
    step3 = transform(step2)
    return step3

Pathless 编程思维，开发者只需定义目标和约束，系统会自动找到实现路径：

import tps_rck_sdk
from tps_rck_sdk import EndpointModel

class MySchema(EndpointModel):
    # ... define schema fields

result = tps_rck_sdk.structured_transform(
    input_text=input_data,
    function_logic="提取关键信息并结构化",
    output_schema=MySchema
)

---

📦 安装

要将 RCK Python SDK 添加到您的项目中，请使用 pip 命令：

pip install pydantic
pip install tps-rck-sdk

🚀 快速上手

1. 导入和初始化客户端

在您的应用启动时，调用 configure 函数来设置全局 API 密钥和其他可选参数。

import tps_rck_sdk
import os

# 推荐从环境变量或安全配置中读取 API 密钥
api_key = os.getenv("RCK_API_KEY", "YOUR_API_KEY")

# 基本配置
tps_rck_sdk.configure(api_key=api_key)

# 高级配置 - 优化性能和并发
tps_rck_sdk.configure(
    api_key=api_key,
    base_url="https://rck-aehhddpisa.us-west-1.fcapp.run",
    rate_limit_calls=15,          # 每60秒最大15次调用
    rate_limit_period=60,
    max_concurrent_requests=5,    # 最大并发请求数
    default_timeout=60.0          # 默认超时时间(秒)
)

2. 发起你的第一次 API 调用

这是一个从文本中提取结构化信息的简单示例。Pathless 的强大之处在于，您可以使用 Pydantic 模型来定义期望的输出结构。

from tps_rck_sdk import EndpointModel, structured_transform
from pydantic import Field
from typing import List

# 1. 使用 Pydantic 定义输出数据结构
class TechStack(EndpointModel):
    languages: List[str] = Field(description="编程语言")
    frameworks: List[str] = Field(description="框架或库")

# 2. 调用 API，描述你的目标
try:
    result = structured_transform(
        input_text="项目后端需要用 Python 和 Django，前端则使用 React。",
        output_schema=TechStack,
        function_logic="从用户的需求中提取技术栈信息。",
        config_overrides={"speed": "fast"} # 可选：覆盖全局配置
    )

    # 3. 直接获得类型安全的 Pydantic 对象
    print(result.model_dump_json(indent=2))
    # 预期输出:
    # {
    #   "languages": [
    #     "Python"
    #   ],
    #   "frameworks": [
    #     "Django",
    #     "React"
    #   ]
    # }

except tps_rck_sdk.RCKAPIError as e:
    print(f"API 请求失败: {e}")

---

🏛️ SDK 架构与引擎选择

SDK 提供了一系列高级函数，每个函数都对应 RCK 后端的一个专用引擎。您也可以使用 auto 函数让 RCK 自动为您选择最合适的引擎。

graph TD
    A[计算任务] --> B{任务类型?}
    B -->|结构化转换| C[structured_transform<br/>🎯 Standard 引擎]
    B -->|示例学习| D[learn_from_examples<br/>🧲 Attractor 引擎]
    B -->|文本/代码生成| E[generate_text<br/>💫 Pure 引擎]
    B -->|图像生成| F[generate_image<br/>🎨 Image 引擎]
    B -->|不确定?| G[auto<br/>🤖 自动选择引擎]

---

📖 API 详细用法

structured_transform

通用AI计算，将非结构化输入转换为精确的、遵循 Pydantic 模式的结构化输出。

• 参数:

  • input_text (str, 必需): 需要处理的输入文本。
  • output_schema (Type[EndpointModel], 必需): 用于定义输出结构的 Pydantic 模型。
  • function_logic (str, 必需): 对任务的自然语言描述。
  • config_overrides (dict, 可选): 覆盖全局配置，例如 {"speed": "quality", "temperature": 0.5}。

• 返回: EndpointModel 的实例。

• 示例:

class UserInfo(EndpointModel):
    name: str
    age: int

user = structured_transform(
    input_text="用户叫张三，今年25岁。",
    output_schema=UserInfo,
    function_logic="提取姓名和年龄。"
)
print(f"姓名: {user.name}, 年龄: {user.age}")

generate_text

自然语言生成，用于创意写作、代码生成等任务。

• 参数:

  • input_text (str, 必需): 输入的提示或上下文。
  • function_logic (str, 必需): 对生成任务的描述。

• 返回: str。

• 示例:

code = generate_text(
    input_text="一个计算斐波那契数列的 Python 函数",
    function_logic="只生成函数代码，不要任何解释性文字。"
)
print(code)

learn_from_examples

通过少量样本归纳出模式和逻辑，然后处理新数据。

• 参数:

  • input_model (AttractorInputBase, 必需): 需要应用学习到模式的新输入 Pydantic 模型。
  • output_schema (Type[EndpointModel], 必需): 期望输出的 Pydantic 模型。
  • examples (List[Tuple[Input, Output]], 必需): 一个或多个 (输入模型, 输出模型) 的元组。

• 返回: EndpointModel 的实例。

• 示例:

from tps_rck_sdk import AttractorInputBase

class Product(AttractorInputBase):
    description: str

class ProductCategory(EndpointModel):
    category: str

examples = [
    (Product(description="苹果公司的最新智能手机"), ProductCategory(category="电子产品")),
    (Product(description="一本关于中世纪历史的书"), ProductCategory(category="书籍")),
]
new_product = Product(description="新鲜采摘的有机草莓")

result = learn_from_examples(
    input_model=new_product,
    output_schema=ProductCategory,
    examples=examples
)
print(result.category) # -> "食品"

generate_image

根据文本描述生成图像。

• 参数:

  • input_text (str, 必需): 图像核心内容的描述。
  • composition (str, 必需): 构图和镜头描述。
  • lighting (str, 必需): 光照条件描述。
  • style (str, 必需): 艺术风格描述。

• 返回: ImageResponse。

• 示例:

image_response = generate_image(
    input_text="一只雄伟的巨龙飞越水晶城堡",
    composition="从下方拍摄的广角镜头，强调巨龙的巨大",
    lighting="黄金时刻的阳光",
    style="史诗奇幻数字绘画风格"
)

if image_response.images:
    # imageData 是解码后的 bytes，可直接写入文件
    image_bytes = image_response.images[0].imageData
    with open("dragon.png", "wb") as f:
        f.write(image_bytes)
    print("图像已保存为 dragon.png")

auto

自动模式，让 RCK 后端根据您的参数智能地决定调用哪个引擎。当您不确定任务类型时，这是最佳选择。

• 参数: 灵活的参数组合，至少需要 input_text 和一个逻辑定义参数（如 function_logic, examples, 或图像参数）。

• 返回: Union[EndpointModel, ImageResponse, str, dict] - 返回类型不确定，需要进行类型检查。

• 示例:

from tps_rck_sdk import auto, ImageResponse

# 场景1：可能返回结构化数据
result = auto(
    input_text="分析情感：'这太棒了！'",
    function_logic="返回包含 'sentiment' 和 'confidence' 的JSON。",
    output_schema=SentimentSchema 
)
if isinstance(result, SentimentSchema):
    print(f"情感: {result.sentiment}")

# 场景2：可能返回文本
result = auto(
    input_text="用一句话总结相对论",
    function_logic="提供一个简洁的摘要。"
)
if isinstance(result, str):
    print(f"摘要: {result}")

# 场景3：可能返回图像
result = auto(
    input_text="一只可爱的小熊猫",
    composition="特写",
    lighting="柔和的光线",
    style="照片级真实感"
)
if isinstance(result, ImageResponse):
    print(f"生成了 {result.count} 张图片。")

---

📥 处理响应

• 结构化数据: structured_transform, learn_from_examples 和 auto (当提供 output_schema 时) 直接返回您定义的 Pydantic 模型实例，您可以立即访问其属性。
• 文本: generate_text 直接返回一个字符串。
• 图像: generate_image 返回一个 ImageResponse 对象。
  • response.images: List[ImageDetails] - 包含所有图像信息的列表。
  • image.imageData: bytes - 图像的原始字节数据，可直接用于保存或处理。
  • image.data_url: str - 图像的 Base64 Data URL。

---

🚨 错误处理

当API请求失败或参数验证不通过时，SDK会抛出特定类型的异常。

from tps_rck_sdk import RCKAPIError, RCKConfigurationError

try:
    # ... 你的 API 调用
except RCKAPIError as e:
    # 来自服务器的错误
    print(f"API Error: {e}")
    print(f"Status Code: {e.status_code}")
    print(f"Response Body: {e.response_body}")
except RCKConfigurationError as e:
    # SDK 配置错误
    print(f"Configuration Error: {e}")
except Exception as e:
    # 其他网络错误或未知错误
    print(f"An unexpected error occurred: {e}")

• RCKAPIError: 表示服务器返回了一个错误（例如，4xx或5xx状态码）。
• RCKConfigurationError: 表示 SDK 未正确配置（例如，缺少 API 密钥）。

📄 许可证

本项目采用 MIT 许可证授权。