文档中心

市场洞察

接口文档

市场洞察 Agent AI 洞察筛选数据订阅接口

本文档介绍市场洞察 Agent 的 AI 精筛数据通过 API 调用的配置接口。

安装 SDK

在使用 Insight SDK 之前，需要先参考本文档完成 SDK 的安装与基础初始化。本章节将分别介绍 Go 和 Python 语言的安装方法、环境要求、关键注意事项、初始化示例及高级配置。

环境要求

在安装和使用 Volcengine SDK 前，请确保你的开发环境满足以下最低要求。

Python

Python 版本: SDK 要求 Python 2.7 或更高版本。如果你需要使用 arkruntime 相关服务，则必须使用 Python 3.6 或更高版本。你可以通过以下命令检查本地的 Python 版本：

python --version

Windows 路径限制: 在 Windows 系统上，由于文件系统存在最大路径长度限制，过长的 SDK 文件路径可能导致安装失败。建议通过修改注册表来启用长路径支持：
- 按下 Win + R，输入 regedit 打开注册表编辑器。
- 找到 \HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\FileSystem 路径。
- 将其下的 LongPathsEnabled 变量值设置为 1。

Go

Go 版本: SDK 要求 Go 1.14 或更高版本。如果你需要使用 arkruntime 相关服务，则必须使用 Go 1.18 或更高版本。你可以通过以下命令检查本地的 Go 版本：

go version

注意事项

在开始安装和编码前，请仔细阅读以下几点重要建议，它们将帮助你建立更安全、稳定和高效的应用。

不要硬编码访问凭证 (AK/SK): 为了防止凭证泄露带来的安全风险，强烈建议通过环境变量或安全的配置中心来管理你的 Access Key 和 Secret Key。避免在代码中以明文形式硬编码。SDK 默认会从环境变量 VOLCSTACK_ACCESS_KEY_ID 和 VOLCSTACK_SECRET_ACCESS_KEY 中读取凭证。
默认使用 HTTPS: 为保证数据传输的安全性，SDK 的新版本默认使用 HTTPS 协议发起所有 API 请求。如果你因特殊情况需要降级到 HTTP，可以手动进行配置，但我们不推荐这样做。
Region 与 Endpoint 的基本认知: Volcengine 的服务部署在不同的地理区域 (Region)，每个区域都有一个独立的接入地址 (Endpoint)。在初始化 SDK 时，你必须正确指定 Region，SDK 会根据 Region 和服务名自动生成 Endpoint。例如，cn-beijing 是一个有效的 Region 值。
超时与代理的可选配置: SDK 允许你根据网络环境自定义请求超时时间和代理服务器。如果你的应用需要通过代理访问公网，或者需要对 API 请求的响应时间有严格控制，可以参考“高级配置”部分进行设置。

安装步骤

你可以根据自己的项目管理方式，选择以下任一方式安装 SDK。

Python

Pip 安装（推荐）

我们强烈推荐使用 pip 进行安装，这是最快捷的方式。执行以下命令即可安装最新版本的 SDK：

# 安装 volcengine-python-sdk 版本大于5.0.6即可
pip install volcengine-python-sdk==5.0.6

你也可以通过访问 SDK 在 GitHub 仓库的 Releases 页面查看所有历史版本。

Go

Go Modules 安装（推荐）

如果你的项目使用 Go Modules 管理依赖，可以直接在你的项目目录下执行以下命令，获取最新版本的 SDK：

# 获取最新版本的 Go SDK
go get github.com/volcengine/volcengine-go-sdk@latest

初始化

安装完成后，你需要对 SDK 进行初始化，主要是配置访问凭证（AK/SK）和区域（Region）。

Python

在 Python 中，我们建议通过 Configuration.set_default 方法设置全局默认配置。这样，在创建 API 客户端时就无需重复传入配置。
以下是一个最小化的初始化示例：

# -*- coding: utf-8 -*-
import os
import volcenginesdkcore
from volcenginesdkinsight import INSIGHTApi, PullPostRequest
from volcenginesdkcore.rest import ApiException

def main():
    # 建议从环境变量中读取凭证信息，避免硬编码
    ak = os.environ.get("VOLCSTACK_ACCESS_KEY_ID")
    sk = os.environ.get("VOLCSTACK_SECRET_ACCESS_KEY")
    region = "cn-beijing" # 替换为你的目标 Region

    if not all([ak, sk]):
        print("Error: Access Key ID or Secret Access Key not found in environment variables.")
        return

    # 1. 配置全局默认参数
    configuration = volcenginesdkcore.Configuration()
    configuration.ak = ak
    configuration.sk = sk
    configuration.region = region
    # configuration.debug = True # 如需开启调试，可取消此行注释
    volcenginesdkcore.Configuration.set_default(configuration)

    # 2. 创建 API 客户端实例
    api = INSIGHTApi()

    # 3. 构造请求
    req = PullPostRequest(
        task_id=12345, # 替换为你的订阅任务 ID
        start_time='2026-01-20 00:00:00', # start_time 仅在首次请求时生效
        end_time='2026-01-21 00:00:00',
        size=20,
    )

    try:
        # 4. 发起请求并处理响应
        resp = api.pull_post(req)
        print("Successfully pulled data:")
        for doc in resp.item_docs:
             print(f"  - PostID: {doc.post_id}, Title: {doc.title}")
    except ApiException as e:
        print(f"Exception when calling INSIGHTApi->pull_post: {e}\\n")
    except Exception as e:
        print(f"An unexpected error occurred: {e}\\n")


if __name__ == '__main__':
    main()

Go

在 Go 中，初始化过程主要包括创建 Config、Session，最后基于 Session 创建具体的服务客户端（如 insight.New(sess)）。
以下是一个最小化的初始化示例：

package main

import (
        "fmt"
        "os"

        "github.com/volcengine/volcengine-go-sdk/service/insight"
        "github.com/volcengine/volcengine-go-sdk/volcengine"
        "github.com/volcengine/volcengine-go-sdk/volcengine/credentials"
        "github.com/volcengine/volcengine-go-sdk/volcengine/session"
)

func main() {
        // 建议从环境变量中读取凭证信息，避免硬编码
        ak := os.Getenv("VOLCSTACK_ACCESS_KEY_ID")
        sk := os.Getenv("VOLCSTACK_SECRET_ACCESS_KEY")
        region := "cn-beijing" // 替换为你的目标 Region

        if ak == "" || sk == "" {
                fmt.Println("Error: Access Key ID or Secret Access Key not found in environment variables.")
                return
        }

        // 1. 创建配置对象
        config := volcengine.NewConfig().
                WithRegion(region).
                WithCredentials(credentials.NewStaticCredentials(ak, sk, ""))
                // WithDebug(true) // 如需开启调试，可取消此行注释

        // 2. 创建 Session
        sess, err := session.NewSession(config)
        if err != nil {
                panic(fmt.Errorf("failed to create session: %w", err))
        }

        // 3. 创建 Insight 服务客户端
        svc := insight.New(sess)

        // 4. 构造请求
        req := &insight.PullPostInput{
                TaskID:    volcengine.Int32(12345), // 替换为你的订阅任务 ID
                StartTime: volcengine.String("2026-01-20 00:00:00"), // start_time 仅在首次请求时生效
                EndTime:   volcengine.String("2026-01-21 00:00:00"),
                Size:      volcengine.Int32(20),
        }

        // 5. 发起请求并处理响应
        resp, err := svc.PullPost(req)
        if err != nil {
                fmt.Printf("failed to call PullPost: %v\\n", err)
                return
        }

        fmt.Println("Successfully pulled data:")
        for _, doc := range resp.ItemDocs {
                fmt.Printf("  - PostID: %s, Title: %s\\n", volcengine.StringValue(doc.PostID), volcengine.StringValue(doc.Title))
        }
}

高级配置

SDK 提供了一系列高级配置选项，允许你对客户端行为进行精细化控制，以适应不同的网络环境和业务需求。

代理配置

如果你的应用需要通过代理服务器访问火山引擎服务，可以进行如下配置。SDK 支持通过代码或环境变量设置代理。

注意: 代码中设置的代理优先级高于环境变量。支持的环境变量包括 http_proxy/HTTP_PROXY 和 https_proxy/HTTPS_PROXY。

Python

import volcenginesdkcore

configuration = volcenginesdkcore.Configuration()
# ... 其他配置 ...

# 设置 HTTP 和 HTTPS 代理
configuration.http_proxy = "http://your_proxy_address:port"
configuration.https_proxy = "https://your_proxy_address:port"

volcenginesdkcore.Configuration.set_default(configuration)

Go

import "github.com/volcengine/volcengine-go-sdk/volcengine"

config := volcengine.NewConfig().
    // ... 其他配置 ...
    WithHTTPProxy("http://your_proxy_address:port").
    WithHTTPSProxy("https://your_proxy_address:port")

// ... 创建 Session 和服务客户端 ...

超时配置

你可以为 API 请求设置连接超时和读取超时，以避免因网络问题导致长时间等待。

Python

connect_timeout: 建立连接的超时时间（秒）。
read_timeout: 等待服务器响应的超时时间（秒）。

import volcenginesdkcore

configuration = volcenginesdkcore.Configuration()
# ... 其他配置 ...

configuration.connect_timeout = 10  # 设置连接超时为 10 秒
configuration.read_timeout = 20   # 设置读取超时为 20 秒

volcenginesdkcore.Configuration.set_default(configuration)

Go

在 Go SDK 中，超时配置通过自定义 http.Client 实现。

Dialer.Timeout: 对应连接超时。
Client.Timeout: 对应读取超时。

import (
    "net"
    "net/http"
    "time"
    "github.com/volcengine/volcengine-go-sdk/volcengine"
)

transport := &http.Transport{
    DialContext: (&net.Dialer{
        Timeout:   10 * time.Second, // 设置连接超时为 10 秒
        KeepAlive: 30 * time.Second,
    }).DialContext,
    // ... 其他 Transport 配置 ...
}

client := &http.Client{
    Transport: transport,
    Timeout:   20 * time.Second, // 设置读取超时为 20 秒
}

config := volcengine.NewConfig().
    // ... 其他配置 ...
    WithHTTPClient(client)

// ... 创建 Session 和服务客户端 ...

重试配置

SDK 内置了针对网络错误和部分服务端限流错误的自动重试机制，以提高应用的健壮性。

Python

auto_retry: 是否开启自动重试（默认为 True）。
num_max_retries: 最大重试次数（默认为 3）。
retry_error_codes: 一个集合，包含需要触发重试的业务错误码（例如 {"RequestLimitExceeded"}）。

import volcenginesdkcore

configuration = volcenginesdkcore.Configuration()
# ... 其他配置 ...

configuration.auto_retry = True
configuration.num_max_retries = 5  # 设置最大重试次数为 5
configuration.retry_error_codes = {"Throttling", "RequestLimitExceeded"}

volcenginesdkcore.Configuration.set_default(configuration)

Go

WithMaxRetries: 设置最大重试次数。设置为 0 可禁用重试。

import "github.com/volcengine/volcengine-go-sdk/volcengine"

config := volcengine.NewConfig().
    // ... 其他配置 ...
    WithMaxRetries(5) // 设置最大重试次数为 5

// ... 创建 Session 和服务客户端 ...

调试与日志

开启调试模式可以帮助你查看详细的请求和响应信息，便于排查问题。

Python

debug: 设为 True 以开启调试模式，SDK 会打印详细的日志。
logger_file: 指定日志输出的文件路径。

import volcenginesdkcore

configuration = volcenginesdkcore.Configuration()
# ... 其他配置 ...

configuration.debug = True
configuration.logger_file = "sdk.log"  # 将日志输出到 sdk.log 文件

volcenginesdkcore.Configuration.set_default(configuration)

Go

WithDebug: 设为 true 以开启调试模式。
WithLogWriter: 指定日志的输出位置，例如文件。

import (
    "os"
    "github.com/volcengine/volcengine-go-sdk/volcengine"
)

// 将日志输出到文件
file, _ := os.Create("sdk.log")

config := volcengine.NewConfig().
    // ... 其他配置 ...
    WithDebug(true).
    WithLogWriter(file)

// ... 创建 Session 和服务客户端 ...

常见问题

在 Windows 上安装 Python SDK 时，为什么会因为路径太长而失败？
1. 这是因为 Windows 系统默认限制了文件路径的最大长度（通常是 260 个字符）。volcengine-python-sdk 包含的服务和模型众多，导致文件层级较深，路径可能超出限制。
解决方案:
1. 如“环境要求”部分所述，你需要修改 Windows 注册表以启用长路径支持。找到 HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\FileSystem 并将 LongPathsEnabled 的值设为 1。
如何将 SDK 的请求协议从 HTTPS 切换到 HTTP？
1. 尽管我们强烈不建议这样做，但如果你确实需要在测试环境或特殊网络下使用 HTTP，可以按以下方式配置：
  1. Python: 设置 configuration.scheme = "http"。
  2. Go: 调用 config.WithDisableSSL(true)。
如何自定义服务的 Endpoint？
1. 在标准情况下，SDK 会根据你指定的 Region 自动生成服务地址。但在某些场景下，例如访问私有化部署的集群或特殊的测试环境，你可能需要手动指定 Endpoint。
  1. Python: 设置 configuration.host = "your-custom-endpoint.com"。
  2. Go: 调用 config.WithEndpoint("your-custom-endpoint.com")。

SDK 鉴权

在调用 Insight SDK 之前，您必须完成鉴权配置。SDK 通过验证您的访问凭据（Access Credentials）来确保请求的合法性与安全性。本章节将详细介绍支持的鉴权方式、适用场景及配置步骤，帮助您快速完成集成。

鉴权方式

火山引擎 SDK 支持多种鉴权机制，以适应不同的安全需求和运行环境。您可以根据业务场景选择最合适的方式：

鉴权方式	核心特点	适用场景	Go 支持	Python 支持
AK/SK（永久密钥）	简单、静态、永久有效。	开发测试、服务端等安全可控的环境。	✔	✔
STS 临时凭证	短期有效、权限可控，包含 AK、SK 和 Session Token。	需要临时授权的场景，如移动端或 Web 应用的后端代理。	✔	✔
AssumeRole / Provider	动态获取、自动续期、权限精细。	长期运行且需自动管理凭证的场景，如部署在火山引擎 ECS、EKS 上的应用，或集成 OIDC/SAML 的联邦认证场景。	✔	✔

AK/SK 鉴权

AK/SK (Access Key ID / Secret Access Key) 是您访问火山引擎服务的永久身份凭证，与您的账户关联。此方式配置简单，适用于在安全可控的服务端环境中调用 SDK。

操作步骤

获取 AK/SK
1. 登录火山引擎控制台。
2. 进入 访问控制 (IAM) > 身份管理 > 用户页面。
3. 选择或创建一个用户，进入其详情页，在 访问密钥 页签下创建并记录 AK/SK。
配置凭证 我们强烈推荐使用环境变量来存储和加载 AK/SK，以避免在代码中硬编码敏感信息。
- 方式一：配置环境变量（推荐）
在您的运行环境中设置以下环境变量，SDK 会自动读取：

export VOLCSTACK_ACCESS_KEY_ID="yourAccessKeyId"
export VOLCSTACK_SECRET_ACCESS_KEY="yourAccessKeySecret"

方式二：在代码中配置（不推荐用于生产）

直接在初始化代码中设置 AK/SK。

代码示例

Python

import os
import volcenginesdkcore
from volcenginesdkinsight import INSIGHTApi, PullPostRequest

# 推荐：从环境变量读取 AK/SK
ak = os.getenv("VOLCSTACK_ACCESS_KEY_ID", "yourAkHere")
sk = os.getenv("VOLCSTACK_SECRET_ACCESS_KEY", "yourSkHere")
region = "cn-beijing"

# 1. 初始化全局配置
configuration = volcenginesdkcore.Configuration()
configuration.ak = ak
configuration.sk = sk
configuration.region = region
volcenginesdkcore.Configuration.set_default(configuration)

# 2. 创建 API 客户端
api_instance = INSIGHTApi()

# 3. 构造请求（此处仅为示例，实际参数请替换）
req = PullPostRequest(
    task_id=12345,
    start_time="2026-01-20 00:00:00",
    end_time="2026-01-21 00:00:00",
    size=20,
)

try:
    # 4. 发起请求
    resp = api_instance.pull_post(req)
    print(resp)
except Exception as e:
    print(f"An error occurred: {e}")

Go

package main

import (
        "fmt"
        "os"

        "github.com/volcengine/volcengine-go-sdk/service/insight"
        "github.com/volcengine/volcengine-go-sdk/volcengine"
        "github.com/volcengine/volcengine-go-sdk/volcengine/credentials"
        "github.com/volcengine/volcengine-go-sdk/volcengine/session"
)

func main() {
        // 方式一：从环境变量自动加载（推荐）
        // 请确保已设置 VOLCSTACK_ACCESS_KEY_ID 和 VOLCSTACK_SECRET_ACCESS_KEY
        config := volcengine.NewConfig().
                WithRegion("cn-beijing").
                WithCredentials(credentials.NewEnvCredentials())

        // 方式二：在代码中静态配置（不推荐用于生产）
        /*
        ak := os.Getenv("VOLCSTACK_ACCESS_KEY_ID")
        sk := os.Getenv("VOLCSTACK_SECRET_ACCESS_KEY")
        if ak == "" || sk == "" {
                ak = "yourAkHere"
                sk = "yourSkHere"
        }
        config := volcengine.NewConfig().
                WithRegion("cn-beijing").
                WithCredentials(credentials.NewStaticCredentials(ak, sk, ""))
        */

        // 2. 创建会话
        sess, err := session.NewSession(config)
        if err != nil {
                panic(fmt.Errorf("failed to create session: %w", err))
        }

        // 3. 创建 Insight 服务客户端
        svc := insight.New(sess)

        // 4. 构造请求（此处仅为示例，实际参数请替换）
        req := &insight.PullPostInput{
                TaskID:    volcengine.Int32(12345),
                StartTime: volcengine.String("2026-01-20 00:00:00"),
                EndTime:   volcengine.String("2026-01-21 00:00:00"),
                Size:      volcengine.Int32(20),
        }

        // 5. 发起请求
        resp, err := svc.PullPost(req)
        if err != nil {
                panic(fmt.Errorf("failed to call PullPost: %w", err))
        }

        fmt.Printf("Response: %s\n", resp.String())
}

STS 临时凭证鉴权

STS (Security Token Service) 提供临时的访问凭证，包含一个额外的 SessionToken。这种方式安全级别更高，适用于需要将访问权限临时授予第三方应用或用户的场景。

操作步骤

获取 STS 凭证您的服务端需要调用 STS 的 AssumeRole 接口来获取临时访问凭证，包括 AccessKeyId、SecretAccessKey 和 SessionToken。
配置凭证 将获取到的三元组配置到 SDK 中。同样，推荐使用环境变量。
- 方式一：配置环境变量（推荐）

export VOLCSTACK_ACCESS_KEY_ID="yourTempAccessKeyId"
export VOLCSTACK_SECRET_ACCESS_KEY="yourTempSecretAccessKey"
export VOLCSTACK_SESSION_TOKEN="yourSessionToken"

- **方式二：在代码中配置**

在初始化代码中设置 AK、SK 和 Session Token。

代码示例

Python

import os
import volcenginesdkcore
from volcenginesdkinsight import INSIGHTApi

# 从环境变量或服务端下发获取临时凭证
temp_ak = os.getenv("VOLCSTACK_ACCESS_KEY_ID", "yourTempAkHere")
temp_sk = os.getenv("VOLCSTACK_SECRET_ACCESS_KEY", "yourTempSkHere")
session_token = os.getenv("VOLCSTACK_SESSION_TOKEN", "yourTokenHere")
region = "cn-beijing"

# 初始化全局配置
configuration = volcenginesdkcore.Configuration()
configuration.ak = temp_ak
configuration.sk = temp_sk
configuration.session_token = session_token # 设置 Session Token
configuration.region = region
volcenginesdkcore.Configuration.set_default(configuration)

# 创建 API 客户端
api_instance = INSIGHTApi()
# ...后续调用与 AK/SK 方式相同

Go

package main

import (
        "os"

        "github.com/volcengine/volcengine-go-sdk/volcengine"
        "github.com/volcengine/volcengine-go-sdk/volcengine/credentials"
        "github.com/volcengine/volcengine-go-sdk/volcengine/session"
        "github.com/volcengine/volcengine-go-sdk/service/insight"
)

func main() {
        // 从环境变量或服务端下发获取临时凭证
        tempAk := os.Getenv("VOLCSTACK_ACCESS_KEY_ID")
        tempSk := os.Getenv("VOLCSTACK_SECRET_ACCESS_KEY")
        token := os.Getenv("VOLCSTACK_SESSION_TOKEN")
        region := "cn-beijing"

    if tempAk == "" || tempSk == "" || token == "" {
        tempAk = "yourTempAkHere"
        tempSk = "yourTempSkHere"
        token = "yourTokenHere"
    }

        // 方式一：从环境变量自动加载（推荐）
        // config := volcengine.NewConfig().
        //         WithRegion(region).
        //         WithCredentials(credentials.NewEnvCredentials())

        // 方式二：在代码中静态配置
        config := volcengine.NewConfig().
                WithRegion(region).
                WithCredentials(credentials.NewStaticCredentials(tempAk, tempSk, token))

        sess, err := session.NewSession(config)
        if err != nil {
                panic(err)
        }

        svc := insight.New(sess)
        // ...后续调用与 AK/SK 方式相同
}

AssumeRole / Provider 鉴权

这是一种更高级、更安全的凭证管理方式。SDK 内部封装了对 STS AssumeRole 接口的调用和凭证的自动刷新逻辑，开发者无需关心临时凭证的获取和有效期管理。
此方式尤其适用于部署在火山引擎基础设施（如 ECS、EKS）上的长期运行服务，或需要与 OIDC/SAML 等身份提供商集成的联邦认证场景。

操作步骤

准备扮演角色 (Role) 在 访问控制 (IAM) 中创建一个角色，并为其授予访问 Insight 服务的最小权限。记录下该角色的名称和所属主账号 ID。
配置 Provider 在 SDK 初始化时，配置一个凭证提供者 (Credential Provider)，并传入扮演角色所需的信息（通常是另一个有权限扮演角色的子账号 AK/SK）。

代码示例

Python

Python SDK 提供了 StsCredentialProvider，可以自动完成 AssumeRole 并刷新凭证。

import os
import volcenginesdkcore
from volcenginesdkcore.auth.providers.sts_provider import StsCredentialProvider
from volcenginesdkinsight import INSIGHTApi

# 用于扮演角色的子账号 AK/SK，推荐从环境变量读取
assume_role_ak = os.getenv("ASSUME_ROLE_AK", "yourSubAccountAk")
assume_role_sk = os.getenv("ASSUME_ROLE_SK", "yourSubAccountSk")

# 角色信息
# RoleName: 欲扮演的角色名，例如 "InsightReaderRole"
# AccountId: 该角色所属的主账号 ID，例如 "2100000000"
role_name = "YourInsightReaderRole"
account_id = "2100000000"
region = "cn-beijing"

configuration = volcenginesdkcore.Configuration()
configuration.region = region

# 配置 StsCredentialProvider
configuration.credential_provider = StsCredentialProvider(
    ak=assume_role_ak,
    sk=assume_role_sk,
    role_name=role_name,
    account_id=account_id,
    duration_seconds=3600,  # 凭证有效期，默认 3600 秒
    host="sts.volcengineapi.com", # STS 服务地址
    region=region,
)

volcenginesdkcore.Configuration.set_default(configuration)

# 创建的客户端将自动使用 Provider 管理的凭证
api_instance = INSIGHTApi()
# ...后续调用与 AK/SK 方式相同

Go

Go SDK 提供了 credentials.NewStsCredentials 来实现类似功能。

package main

import (
        "os"
        "time"

        "github.com/volcengine/volcengine-go-sdk/volcengine"
        "github.com/volcengine/volcengine-go-sdk/volcengine/credentials"
        "github.com/volcengine/volcengine-go-sdk/volcengine/session"
        "github.com/volcengine/volcengine-go-sdk/service/insight"
)

func main() {
        // 用于扮演角色的子账号 AK/SK，推荐从环境变量获取
        assumeRoleAk := os.Getenv("ASSUME_ROLE_AK")
        assumeRoleSk := os.Getenv("ASSUME_ROLE_SK")
        
    if assumeRoleAk == "" || assumeRoleSk == "" {
        assumeRoleAk = "yourSubAccountAk"
        assumeRoleSk = "yourSubAccountSk"
    }

        region := "cn-beijing"

        // 配置 StsCredentials
        config := volcengine.NewConfig().
                WithRegion(region).
                WithCredentials(credentials.NewStsCredentials(credentials.StsValue{
                        AccessKey:       assumeRoleAk,
                        SecurityKey:     assumeRoleSk,
                        // RoleName: 欲扮演的角色名
                        RoleName:        "YourInsightReaderRole",
                        // AccountId: 角色所属的主账号 ID
                        AccountId:       "2100000000",
                        Host:            "sts.volcengineapi.com",   // STS 服务地址
                        Region:          region,
                        DurationSeconds: 900,                       // 凭证有效期，单位秒
                        Timeout:         5 * time.Second,           // 请求 STS 的超时时间
                }))

        sess, err := session.NewSession(config)
        if err != nil {
                panic(err)
        }
    
    // 创建的客户端将自动使用 Provider 管理的凭证
        svc := insight.New(sess)
        // ...后续调用与 AK/SK 方式相同
}

限制与安全建议

密钥安全：切勿在代码中硬编码任何永久性 AK/SK。请始终使用环境变量、配置中心或火山引擎提供的实例角色等安全方式管理您的凭证。
最小权限原则：为您的 AK/SK 或角色授予完成任务所需的最小权限集。避免使用拥有账户完全控制权的根用户凭证。
临时凭证有效期：在使用 STS 或 AssumeRole 时，请根据业务场景设置尽可能短的凭证有效期，以降低凭证泄露后的风险。建议有效期不超过 1 小时。
错误处理：请务必在代码中添加对鉴权失败错误的捕获和处理逻辑。常见的鉴权错误码包括 InvalidAccessKeyId、SignatureDoesNotMatch 等，具体请参考各服务的 API 文档。

快速开始

阅读本章节后，你将学会如何使用 Insight SDK，在指定的时间范围内基于 task_id 拉取订阅内容，并通过分页循环的方式完整地消费所有数据。

场景介绍

本章节的示例将演示一个核心场景：拉取指定订阅任务在最近一天内的全部数据，并打印每条内容的 post_id、title 和 publish_time。
我们将遵循 Insight API 的分页规范，即 start_time 参数仅在首次请求时用于确定数据范围，后续的分页拉取将完全依赖 page_token 进行，以保证数据的一致性和完整性。

注意事项

在开始编码前，请了解以下关键点，以确保顺利集成和稳定运行：

时间范围约束：start_time 仅在发起首次请求（即不携带 page_token 的请求）时生效，用于划定数据拉取的起始边界。一旦分页开始，后续请求中的 start_time 和 end_time 参数将被忽略。如需更改时间范围，必须发起一次全新的请求。
分页游标不可解析：分页游标 page_token 是一个不透明的字符串，请勿尝试解析其内容或将其用于除分页外的其他目的。每次请求成功后，使用响应中的 next_page_token 作为下一次请求的 page_token 即可。
大字段处理建议：content、ocr、asr 和 ocr_high 字段可能包含大量文本，在处理时请注意内存消耗。如果仅需元数据，建议按需读取，避免不必要的数据加载。
数据快照与排序：首次请求会生成一个数据快照。分页过程中，即使该时间段内有新数据产生，这些新数据也不会出现在当前翻页的结果中，以确保数据不重复、不遗漏。数据默认按 update_time 降序排列。
错误与限流处理：请务必在代码中加入对 API 错误的捕获逻辑，特别是对于网络异常和限流错误（如 Throttling、RequestLimitExceeded），建议结合 SDK 的重试机制进行处理。详细配置请参考“安装与配置”章节。

实现流程

通过 PullPost API 拉取数据的完整流程通常包括以下步骤：

初始化客户端：根据你选择的语言（Go 或 Python），配置访问凭证（AK/SK）和区域（Region），创建 Insight 服务的客户端实例。
构造首次请求：创建一个 PullPost 请求对象，并设置以下核心参数：
1. task_id：你的订阅任务 ID。（您可点击监测任务管理-点击对应任务-进入任务结果详情页后，通过上方页面链接 risk/后数字获取对应 task_id ，例如：页面链接为：https://console.volcengine.com/insight/ai/monitor/risk/1509，您对应的 task_id 为1509）
2. start_time 和 end_time：定义首次拉取的时间范围。
3. size：每页返回的数据条数（可选，默认为 20，最大 100）。
执行请求与分页循环：
1. 使用客户端发起 PullPost 请求。
2. 检查响应中的 has_more (Python) 或 HasMore (Go) 字段。如果为 true，表示还有后续数据。
3. 将响应中的 next_page_token (Python) 或 NextPageToken (Go) 保存下来，用于下一次请求。
4. 在循环中，构造新的 PullPost 请求，仅需传入 task_id 和上一步获取的 page_token。
处理返回数据：遍历当前页的 item_docs (Python) 或 ItemDocs (Go) 列表，获取每条数据的内容并进行业务处理（如存入数据库、发送到消息队列等）。
结束条件与收尾：当响应中的 has_more / HasMore 为 false 时，表示所有数据均已拉取完毕，循环终止。

Post 对象字段说明

posts 数组中的每个对象代表一条独立的数据，其详细字段如下：

字段名	类型	是否必填	说明	约束与示例
`post_id`	String	是	数据的唯一标识符。	`"12132132132132"`
`title`	String	是	数据标题。	-
`content`	String	是	数据正文。	注意： `content`, `ocr`, `asr`，`ocr_high` 可能为大字段，内容较长。
`summary`	String	是	数据摘要。
`ocr`	String	是	图片光学字符识别（OCR）提取的文本内容。
`ocr_high`	String	是	高质量图片光学字符识别提取的文本内容。高质量ocr会进行内容消重、标点符号标记等内容清理，相较ocr语义更连贯，但覆盖较低（附加增量信息，仅对部分内容做高质量OCR识别）
`asr`	String	是	音频语音识别（ASR）提取的文本内容。	-
`url`	String	是	原始数据链接。	-
`publish_time`	String	是	数据发布时间。	`示例："2026-01-01 00:00:00"`
`main_domain`	String	是	数据来源的主域名。	`"douyin.com"`
`is_follow`	Boolean	是	是否为转发、跟帖或引用等非原创类型。	`true` 表示是，`false` 表示否。
`media_name`	String	是	发布作者或媒体的名称。	-
`WarnLevel`	String	是	根据设置显示的内容预警分级。	-
`Keywords`	String	是	内容命中的关键词内容。	如同时命中多个关键词，用“,”分隔。
`fans_count`	Integer	是	作者的粉丝数。	当无法获取时，返回 `0` 或 `null`。
`LevelName`	String	是	作者的认证情况。	如“达人用户”、“蓝V用户”、“普通用户”、“无”。
`locations`	Array	是	数据内容中提及的地理位置信息点（Point of Interest）。	-
`risk_type`	Array	是	风险标签列表，用于标识数据中可能存在的风险类型。
`emotion`	String	是	情感分析标签。
`reason`	String	是	AI 精筛原因，解释为何此条数据被模型选中。	-
`dedup_id`	String	是	聚类ID。用于标识内容相似或同源的一组数据。	具有相同 `dedup_id` 的数据可视为同一事件或话题的强相关内容，可用于客户端去重。
`update_time`	String	是	数据入库或更新时间。可用于增量同步。	`示例："2026-01-01 00:00:00"`

Location 对象字段说明

数组中的每个对象代表一个地理位置信息，其详细字段如下：

字段名	类型	是否必填	说明
`name`	String	是	识别出的地理位置名称，例如某个具体地点或街道名。
`region`	String	否	所属区域，如“CN”。
`province`	String	否	省份。
`city`	String	否	城市。
`district`	String	否	区/县。
`town`	String	否	乡/镇/街道。
`region_code`	String	否	区域行政代码。
`city_code`	String	否	城市行政代码。
`district_code`	String	否	区/县行政代码。
`town_code`	String	否	乡/镇/街道行政代码。
`location`	String	否	完整的地理位置描述。

示例代码

以下是 Go 和 Python 的完整示例代码，展示了如何实现上述流程。代码已包含从环境变量读取配置、分页循环和错误处理的逻辑，可直接复制运行。

Python

# -*- coding: utf-8 -*-
"""Insight 数据订阅示例：使用 PullPost 按时间增量拉取并通过游标分页。

运行前准备：
1. 安装 SDK：
   pip install volcengine-python-sdk
2. 通过环境变量配置访问凭证和订阅任务：
   - VOLCSTACK_ACCESS_KEY_ID：你的访问密钥 AK；
   - VOLCSTACK_SECRET_ACCESS_KEY：你的访问密钥 SK；
   - VOLCSTACK_REGION：服务地域，例如 cn-beijing；
   - INSIGHT_TASK_ID：你的订阅任务 ID（整数）。
"""

from __future__ import print_function

import os
import sys
from datetime import datetime, timedelta

import volcenginesdkcore
from volcenginesdkcore.rest import ApiException
from volcenginesdkinsight import INSIGHTApi, PullPostRequest


def init_configuration():
    """初始化全局配置，从环境变量读取 AK/SK/Region。"""
    configuration = volcenginesdkcore.Configuration()
    configuration.ak = os.getenv("VOLCSTACK_ACCESS_KEY_ID", "YOUR_AK")
    configuration.sk = os.getenv("VOLCSTACK_SECRET_ACCESS_KEY", "YOUR_SK")
    configuration.region = os.getenv("VOLCSTACK_REGION", "cn-beijing")
    # configuration.debug = True # 如需开启调试，可取消此行注释
    volcenginesdkcore.Configuration.set_default(configuration)


def main():
    """主函数，执行数据拉取与分页逻辑。"""
    task_id_str = os.getenv("INSIGHT_TASK_ID")
    if not task_id_str:
        print("错误：请先通过环境变量 INSIGHT_TASK_ID 设置订阅任务 ID。", file=sys.stderr)
        sys.exit(1)

    try:
        task_id = int(task_id_str)
    except ValueError:
        print(f"错误：INSIGHT_TASK_ID '{task_id_str}' 必须是有效的整数。", file=sys.stderr)
        sys.exit(1)

    init_configuration()
    api = INSIGHTApi()

    # 示例：拉取最近 1 天的数据。
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=1)

    page_token = None
    page_size = 20
    total_count = 0
    page_num = 0

    print(f"开始拉取任务 {task_id} 从 {start_time.strftime('%Y-%m-%d %H:%M:%S')} 到 {end_time.strftime('%Y-%m-%d %H:%M:%S')} 的数据...")

    while True:
        page_num += 1
        print(f"--- 第 {page_num} 页 ---")

        # 构造请求。注意：start_time 和 end_time 仅在首次请求（page_token 为 None）时生效。
        # 后续分页请求会沿用首次请求确定的数据快照。
        req_params = {
            "task_id": task_id,
            "size": page_size,
            "page_token": page_token,
        }
        if page_token is None:
            req_params["start_time"] = start_time.strftime("%Y-%m-%d %H:%M:%S")
            req_params["end_time"] = end_time.strftime("%Y-%m-%d %H:%M:%S")

        req = PullPostRequest(**req_params)

        try:
            resp = api.pull_post(req)
        except ApiException as e:
            print(f"错误：调用 INSIGHT PullPost 失败: {e}", file=sys.stderr)
            break
        except Exception as e:
            print(f"错误：发生未知错误: {e}", file=sys.stderr)
            break

        # 处理当前页数据。SDK 返回的字段为 item_docs。
        if resp.item_docs and len(resp.item_docs) > 0:
            current_page_count = len(resp.item_docs)
            total_count += current_page_count
            print(f"成功拉取 {current_page_count} 条数据 (累计: {total_count})")
            for doc in resp.item_docs:
                print(f"  - PostID: {doc.post_id}, PublishTime: {doc.publish_time}, Title: {doc.title}")
        else:
            print("当前页无数据。")

        # 检查是否还有更多数据
        if not resp.has_more:
            print("\n所有数据已拉取完毕。")
            break

        # 更新分页游标
        page_token = resp.next_page_token
        if not page_token:
            # 理论上 has_more 为 True 时必须返回 next_page_token，这里仅作为防御性兜底。
            print("警告：has_more 为 True 但 next_page_token 为空，提前终止分页。", file=sys.stderr)
            break

        # 为避免达到 QPS 限制，可以在每次请求后加入短暂延时
        # import time
        # time.sleep(0.1)

    print(f"\n拉取完成，共处理 {total_count} 条数据。")


if __name__ == "__main__":
    main()

Go

// Insight 数据订阅示例：使用 PullPost 按时间增量拉取并通过游标分页。
//
// 运行前准备：
// 1. 安装并初始化 Go SDK 依赖；
// 2. 通过环境变量配置访问凭证和订阅任务：
//    - VOLCSTACK_ACCESS_KEY_ID：你的访问密钥 AK；
//    - VOLCSTACK_SECRET_ACCESS_KEY：你的访问密钥 SK；
//    - VOLCSTACK_REGION：服务地域，例如 cn-beijing；
//    - INSIGHT_TASK_ID：你的订阅任务 ID（整数）。

package main

import (
        "fmt"
        "log"
        "os"
        "strconv"
        "time"

        "github.com/volcengine/volcengine-go-sdk/service/insight"
        "github.com/volcengine/volcengine-go-sdk/volcengine"
        "github.com/volcengine/volcengine-go-sdk/volcengine/credentials"
        "github.com/volcengine/volcengine-go-sdk/volcengine/session"
)

func main() {
        ak := os.Getenv("VOLCSTACK_ACCESS_KEY_ID")
        sk := os.Getenv("VOLCSTACK_SECRET_ACCESS_KEY")
        region := os.Getenv("VOLCSTACK_REGION")
        taskIDStr := os.Getenv("INSIGHT_TASK_ID")

        if ak == "" || sk == "" || region == "" || taskIDStr == "" {
                log.Fatal("错误：请确保已设置 VOLCSTACK_ACCESS_KEY_ID, VOLCSTACK_SECRET_ACCESS_KEY, VOLCSTACK_REGION, 和 INSIGHT_TASK_ID 环境变量。")
        }

        taskID, err := strconv.ParseInt(taskIDStr, 10, 32)
        if err != nil {
                log.Fatalf("错误：INSIGHT_TASK_ID '%s' 必须是有效的整数: %v", taskIDStr, err)
        }

        // 初始化配置
        config := volcengine.NewConfig().
                WithRegion(region).
                WithCredentials(credentials.NewStaticCredentials(ak, sk, ""))
                // WithDebug(true) // 如需开启调试，可取消此行注释

        // 创建会话
        sess, err := session.NewSession(config)
        if err != nil {
                log.Fatalf("错误：创建 Session 失败: %v", err)
        }

        // 创建 Insight 客户端
        client := insight.New(sess)

        // 示例：拉取最近 1 天的数据
        endTime := time.Now().UTC()
        startTime := endTime.Add(-24 * time.Hour)

        var pageToken *string
        pageSize := int32(20)
        totalCount := 0
    pageNum := 0

        fmt.Printf("开始拉取任务 %d 从 %s 到 %s 的数据...\n", taskID, startTime.Format(time.RFC3339), endTime.Format(time.RFC3339))

        for {
        pageNum++
        fmt.Printf("--- 第 %d 页 ---\n", pageNum)

                input := &insight.PullPostInput{
                        TaskID: volcengine.Int32(int32(taskID)),
                        Size:   volcengine.Int32(pageSize),
                }

                if pageToken != nil {
                        input.SetPageToken(*pageToken)
                } else {
                        // 注意：StartTime 和 EndTime 仅在首次请求（pageToken 为 nil）时生效。
                        // 后续分页请求会沿用首次请求确定的数据快照。
                        input.SetStartTime(startTime.Format("2006-01-02 15:04:05"))
                        input.SetEndTime(endTime.Format("2006-01-02 15:04:05"))
                }

                output, err := client.PullPost(input)
                if err != nil {
                        log.Fatalf("错误：调用 INSIGHT PullPost 失败: %v", err)
                }

                // 处理当前页数据
                if output.ItemDocs != nil && len(output.ItemDocs) > 0 {
                        currentPageCount := len(output.ItemDocs)
                        totalCount += currentPageCount
                        fmt.Printf("成功拉取 %d 条数据 (累计: %d)\n", currentPageCount, totalCount)
                        for _, doc := range output.ItemDocs {
                                if doc == nil {
                                        continue
                                }
                                fmt.Printf("  - PostID: %s, PublishTime: %s, Title: %s\n",
                                        volcengine.StringValue(doc.PostID),
                                        volcengine.StringValue(doc.PublishTime),
                                        volcengine.StringValue(doc.Title))
                        }
                } else {
            fmt.Println("当前页无数据。")
        }


                // 检查是否还有更多数据
                if output.HasMore == nil || !*output.HasMore {
                        fmt.Println("\n所有数据已拉取完毕。")
                        break
                }

                // 更新分页游标
                if output.NextPageToken == nil || *output.NextPageToken == "" {
                        // 理论上 HasMore 为 true 时必须返回 NextPageToken，这里加一层防御性判断。
                        log.Println("警告：HasMore 为 true 但 NextPageToken 为空，提前终止分页。")
                        break
                }
                pageToken = output.NextPageToken
        
        // 为避免达到 QPS 限制，可以在每次请求后加入短暂延时
        // time.Sleep(100 * time.Millisecond)
        }
    
    fmt.Printf("\n拉取完成，共处理 %d 条数据。\n", totalCount)
}

验证结果

当你运行上述示例代码后，如果配置正确，你将在控制台看到类似以下的输出。这表明 SDK 已成功发起请求并开始分页拉取数据。

开始拉取任务 12345 从 2026-01-20T10:00:00Z 到 2026-01-21T10:00:00Z 的数据...
--- 第 1 页 ---
成功拉取 20 条数据 (累计: 20)
  - PostID: 7428...90, PublishTime: 2026-01-21 09:50:00, Title: ...
  - PostID: 7428...91, PublishTime: 2026-01-21 09:48:15, Title: ...
...
--- 第 2 页 ---
成功拉取 20 条数据 (累计: 40)
  - PostID: 7428...110, PublishTime: 2026-01-21 08:30:00, Title: ...
...
--- 第 3 页 ---
成功拉取 5 条数据 (累计: 45)
  - PostID: 7428...115, PublishTime: 2026-01-21 08:05:10, Title: ...

所有数据已拉取完毕。

拉取完成，共处理 45 条数据。

成功的标志：

控制台持续打印“成功拉取 X 条数据”的日志。
最终打印“所有数据已拉取完毕”。
整个过程没有因 ApiException 或其他错误而中断。

高级配置

为了增强数据拉取过程的鲁棒性，你可以在 SDK 初始化时组合使用代理、超时和重试配置。这些配置在“安装与配置”章节有详细介绍。在分页消费这类长时间运行的任务中，合理配置它们尤为重要：

超时配置：为 PullPost 请求设置一个合理的连接和读取超时时间，可以防止因网络波动导致进程长时间卡死。
重试配置：开启 SDK 的自动重试功能，并针对 Throttling（限流）等错误码进行配置，可以在遇到临时性服务端问题时自动恢复，提高数据拉取的成功率。
代理配置：如果你的运行环境需要通过代理访问公网，请确保已正确配置 HTTP/HTTPS 代理。

常见问题

如何避免重复消费数据？ 本接口的“数据快照”机制已在分页层面保证了数据不重复。但如果在多次独立任务中拉取重叠的时间段，可能会获取到相同的数据。 解决方案：在你的业务逻辑中，使用每条数据唯一的 post_id 作为幂等键。在将数据存入数据库或处理前，检查 post_id 是否已存在，从而实现业务层面的去重。
如何控制请求频率和每页大小？
1. 每页大小：通过 size 参数控制，最大值为 100。如果需要处理大量数据，建议使用较大的 size 以减少请求次数。
2. 请求频率：请遵守 API 的 QPS（每秒请求数）限制。如果你在循环中拉取速度过快，可能会收到限流错误。可以在每次请求之间加入适当的延时（例如 time.sleep(0.1)），或者在收到限流错误时执行指数退避策略。SDK 的重试机制已内置了退避逻辑。

最近更新时间：2026.09.11 14:01:29

这个页面对您有帮助吗？

有用

无用

市场洞察

环境要求 #

Python #

Go #

注意事项 #

安装步骤 #

Python #

Pip 安装（推荐）

Go #

Go Modules 安装（推荐）

初始化 #

Python #

Go #

高级配置 #

代理配置 #

Python

Go

超时配置 #

Python

Go

重试配置 #

Python

Go

调试与日志 #

Python

Go

常见问题 #

鉴权方式 #

AK/SK 鉴权 #

操作步骤 #

代码示例 #

Python

Go

STS 临时凭证鉴权 #

操作步骤 #

代码示例 #

Python

Go

AssumeRole / Provider 鉴权 #

操作步骤 #

代码示例 #

Python

Go

限制与安全建议 #

场景介绍 #

注意事项 #

实现流程 #

Post 对象字段说明 #

Location 对象字段说明 #

示例代码 #

Python #

Go #

验证结果 #

高级配置 #

常见问题 #

环境要求

Python

Go

注意事项

安装步骤

Python

Go

初始化

Python

Go

高级配置

代理配置

超时配置

重试配置

调试与日志

常见问题

鉴权方式

AK/SK 鉴权

操作步骤

代码示例

STS 临时凭证鉴权

操作步骤

代码示例

AssumeRole / Provider 鉴权

操作步骤

代码示例

限制与安全建议

场景介绍

注意事项

实现流程

Post 对象字段说明

Location 对象字段说明

示例代码

Python

Go

验证结果

高级配置

常见问题