octen 0.1.1b4

Official Python SDK for Octen API - Web Search and Text Embeddings

Octen Python SDK

官方 Python SDK，用于访问 Octen API - 提供强大的 Web 搜索和文本嵌入功能。

✨ 特性

• 🔍 Web 搜索 - 搜索排名的网页结果，支持过滤、高亮和完整内容
• 🧮 文本嵌入 - 将文本转换为高质量的向量表示
• ⚡ 高性能 - 支持 HTTP/2、连接池和长连接复用
• 🔄 自动重试 - 智能重试机制，处理临时错误
• 🛡️ 类型安全 - 完整的类型注解，支持 IDE 自动补全
• 📦 简单易用 - 几行代码即可开始使用

📦 安装

pip install octen

需要 Python 3.8 或更高版本。

开发版本

pip install octen[dev]

异步支持

pip install octen[async]

🚀 快速开始

基础用法

from octen import Octen

# 创建客户端
with Octen(api_key="your-api-key") as client:
    # Web 搜索
    response = client.search.search(query="Python programming", count=5)

    for result in response.results:
        print(f"标题: {result['title']}")
        print(f"URL: {result['url']}")
        print(f"摘要: {result.get('highlight', '')}")
        print("-" * 80)

    # 创建文本嵌入
    embedding = client.embedding.create(
        input=["Hello, world!"],
        model="octen-embedding-4b"
    )
    vector = embedding.get_first_embedding()
    print(f"向量维度: {len(vector)}")

高级搜索

from octen import Octen, HighlightOptions, FullContentOptions

with Octen(api_key="your-api-key") as client:
    response = client.search.search(
        query="机器学习最佳实践",
        count=10,
        search_type="semantic",  # 语义搜索
        include_domains=["github.com", "arxiv.org"],  # 只搜索这些域名
        start_time="2024-01-01T00:00:00Z",  # 时间过滤
        highlight=HighlightOptions(
            enable=True,
            max_tokens=500
        ),
        full_content=FullContentOptions(
            enable=True,
            max_tokens=2000
        ),
        timeout=60.0  # 自定义超时
    )

    print(f"找到 {len(response.results)} 个结果")
    print(f"实际搜索类型: {response.search_type}")
    print(f"Token 使用量: {response.usage}")

批量嵌入

from octen import Octen

with Octen(api_key="your-api-key") as client:
    # 批量处理多个文本
    texts = [
        "人工智能改变世界",
        "深度学习的应用",
        "自然语言处理技术"
    ]

    response = client.embedding.create(
        input=texts,
        model="octen-embedding-8b",
        input_type="document"
    )

    vectors = response.get_embeddings()
    print(f"生成了 {len(vectors)} 个向量")

    # 或者使用便捷方法
    query_vector = client.embedding.embed_query("搜索查询")
    doc_vectors = client.embedding.embed_documents(["文档1", "文档2"])

自定义配置

from octen import Octen

client = Octen(
    api_key="your-api-key",
    base_url="https://api.octen.ai",  # 自定义 API 端点
    timeout=10.0,  # 全局默认超时（秒）
    max_retries=3,  # 最大重试次数
    http2=True  # 启用 HTTP/2
)

try:
    # 这个请求使用全局超时（10秒）
    response1 = client.search.search("query 1")

    # 这个请求覆盖超时为 30 秒
    response2 = client.search.search("complex query", timeout=30.0)
finally:
    client.close()  # 释放连接池资源

📚 API 文档

Search API

client.search.search()

执行 Web 搜索。

参数：

• query (str, 必填): 搜索查询字符串，最多 500 字符
• count (int, 可选): 返回结果数量，范围 1-100，默认 5
• search_type (str, 可选): 搜索类型，可选值：
  • "auto" - 自动选择（默认）
  • "keyword" - 关键词搜索
  • "semantic" - 语义搜索
• include_domains (List[str], 可选): 只包含这些域名的结果
• exclude_domains (List[str], 可选): 排除这些域名的结果
• include_text (List[str], 可选): 结果必须包含这些文本
• exclude_text (List[str], 可选): 结果必须排除这些文本
• time_basis (str, 可选): 时间基准，可选 "auto"、"published"、"crawled"
• start_time (str, 可选): 开始时间，ISO 8601 格式
• end_time (str, 可选): 结束时间，ISO 8601 格式
• highlight (HighlightOptions, 可选): 高亮选项配置
• format (str, 可选): 内容格式，可选 "text"、"markdown"
• safesearch (str, 可选): 安全搜索，可选 "off"、"strict"（默认）
• full_content (FullContentOptions, 可选): 完整内容选项配置
• timeout (float, 可选): 请求超时时间（秒）

返回： SearchResponse 对象

响应属性：

• results - 搜索结果列表
• query - 实际使用的查询
• search_type - 实际使用的搜索类型
• usage - Token 使用量信息
• latency - 延迟信息

Embedding API

client.embedding.create()

创建文本嵌入向量。

参数：

• input (str | List[str], 必填): 输入文本或文本列表
• model (str, 可选): 模型名称，可选：
  • "octen-embedding-0.6b" - 轻量级模型
  • "octen-embedding-4b" - 平衡性能
  • "octen-embedding-8b" - 最高质量
• dimension (int, 可选): 向量维度
• input_type (str, 可选): 输入类型，可选 "query" 或 "document"
• truncation (bool, 可选): 是否截断超长输入，默认 True
• timeout (float, 可选): 请求超时时间（秒）

返回： EmbeddingResponse 对象

响应方法：

• get_embeddings() - 获取所有向量
• get_first_embedding() - 获取第一个向量（单输入时使用）

便捷方法：

• embed_query(text) - 嵌入单个查询文本
• embed_documents(texts) - 批量嵌入文档文本

🔧 异步支持

import asyncio
from octen import AsyncOcten

async def main():
    async with AsyncOcten(api_key="your-api-key") as client:
        # 并发执行多个请求
        search_task = client.search.search(query="AI")
        embedding_task = client.embedding.create(
            input=["Hello"],
            model="octen-embedding-4b"
        )

        results, embeddings = await asyncio.gather(
            search_task,
            embedding_task
        )

        print(f"搜索结果: {len(results.results)}")
        print(f"向量维度: {len(embeddings.get_first_embedding())}")

asyncio.run(main())

⚠️ 错误处理

from octen import (
    Octen,
    OctenAPIError,
    OctenTimeoutError,
    OctenConnectionError,
    OctenRateLimitError,
    OctenAuthenticationError,
)

with Octen(api_key="your-api-key") as client:
    try:
        response = client.search.search("query")
    except OctenAuthenticationError as e:
        print(f"认证失败: {e}")
    except OctenRateLimitError as e:
        print(f"速率限制: {e}")
        print(f"请在 {e.retry_after} 秒后重试")
    except OctenTimeoutError as e:
        print(f"请求超时: {e}")
    except OctenConnectionError as e:
        print(f"连接错误: {e}")
    except OctenAPIError as e:
        print(f"API 错误: {e}")
        print(f"状态码: {e.status_code}")
        print(f"请求 ID: {e.request_id}")

🧪 开发

安装开发依赖

# 从源码安装开发版本
pip install -e ".[dev]"

运行测试

pytest tests/

代码格式化

black octen/
ruff check octen/ --fix

类型检查

mypy octen/

📝 许可证

MIT License - 详见 LICENSE 文件

🔗 链接

• 官方网站
• API 文档

📧 支持

如有问题或需要帮助，请：

• 查看 文档
• 发送邮件至 [e-mail]