tps-rck-sdk 0.1.1
Official Python SDK for the Relational Calculate Kernel (RCK)
Requires Python: >=3.8

# Topos Pathless Service RCK Python SDK 完整教程

### 🎯 RCK：Pathless 中的非传统计算引擎

**RCK (Relational Calculate Kernel)** 是 Pathless 服务中的核心组件，专门负责**非传统、非确定性计算任务**的完整生命周期，包括：
- **编译**：将自然语言需求转换为可执行的计算逻辑
- **部署**：自动部署到云原生环境，无需手动干预
- **运行**：执行非确定性计算任务，支持动态调整
- **监控**：实时监控计算过程和结果质量

### 🎯 Pathless 核心理念：目标驱动，路径无关

传统编程思维，开发者需要编写具体的算法步骤和实现细节：
```python
def process_data(input_data):
    step1 = parse(input_data)
    step2 = validate(step1)
    step3 = transform(step2)
    return step3
```

Pathless 编程思维，开发者只需定义目标和约束，系统会自动找到实现路径：
```python
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` 命令：

```bash
pip install pydantic
pip install tps-rck-sdk
```

## 🚀 快速上手

### 1. 导入和初始化客户端

在您的应用启动时，调用 `configure` 函数来设置全局 API 密钥和其他可选参数。

```python
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 模型来定义期望的输出结构。

```python
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 自动为您选择最合适的引擎。

```mermaid
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` 的实例。

- **示例:**
```python
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`。

- **示例:**
```python
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` 的实例。

- **示例:**
```python
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`。

- **示例:**
```python
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]` - 返回类型不确定，需要进行类型检查。

- **示例:**
```python
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会抛出特定类型的异常。

```python
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 许可证授权。