You need to enable JavaScript to run this app.
导航
A/B测试
  • 文档首页
    • /A/B测试/开发指南/实验SDK集成/服务端SDK/大模型实验SDK集成
大模型实验SDK集成
最近更新时间：2025.03.21 19:20:47首次发布时间：2025.03.21 19:20:47
我的收藏
有用
有用
无用
无用

A/B测试在大模型（LLM）行业中的应用场景非常广泛，包括模型优化、用户体验提升、业务指标改进以及风险控制等，如果您使用DataTester开展大模型应用的A/B实验，需参考本文档对应用进行实验SDK集成，以便实现实验分流和大模型业务指标数据上报。

SDK集成说明

使用 DataTester 平台开展大模型实验时，可借助轻量化集成的 A/B 实验 SDK，能够迅速推动实验落地，并获取模型性能指标，从而更有效地评估各类大模型的业务效果。以下为集成实验SDK后的业务请求和实验指标上报的粗略流程。
Image
在大模型实验场景下，集成实验SDK后主要会实现：

  • 实验分流：与其他通用应用的实验场景类似，集成实验SDK后，会通过服务端实现实验分流，对不同用户下发不同实验版本的实验参数。
  • 大模型性能数据采集与上报：针对大模型应用的场景，可采集不同实验版本下的大模型性能数据并上报，便于您创建并分析大模型的性能指标。

注意事项

细分项

注意事项

实验类型

当前仅编程实验-服务端类型的实验支持开展大模型实验，因此大模型需在服务端进行实验SDK集成。

支持的开发语言

  • 当前仅支持Python，已验证Python版本为3.7-3.12之间。
  • Golang和Java即将上线。

集成方式

当前支持两种集成方式，您可按需选择。

  • 方式1：使用DataTester ArkAI Proxy（推荐）
    • 此方式封装了火山引擎的方舟模型SDK，结合DataTester的核心分流能力的同时，自动上报大模型应用在实验过程中采集到的模型性能数据，帮助您快速构建大模型实验所需的性能指标。
    • 此方式适用于使用火山引擎的方舟模型、且希望便捷快速集成的用户。
  • 方式2：使用三方AI SDK
    • 此方式需要您在集成DataTester的通用实验SDK（实现实验分流）的基础上，再调用第三方大模型的OpenAPI采集模型性能数据，并调用DataTester的接口上报至DataTester。
    • 此方式适用您此前已在应用中深度使用第三方大模型SDK且切换难度大的用户。

前提条件

在集成大模型实验SDK前，您需先在DataTester控制台上根据业务需要创建了一个大模型实验，或创建了大模型功能Feature，获取到了各个实验版本的实验参数SDK嵌入代码。
详情请参见编程实验-大模型实验。

SDK集成

假设当前待开展实验的应用为一个对话式机器人，您已在DataTester上创建了一个大模型实验，其中实验参数为llm_model，通过不同实验版本里llm_model的不同取值，来验证不同模型的效果。以下为两种实验SDK集成方式的集成操作指导。

使用DataTester ArkAI Proxy（推荐）

  1. 安装DataTester的大模型实验SDK。安装包如下。

    datatester_python_sdk-2.2.5rc0-py3-none-any.whl
    未知大小

  2. 编写服务端实验代码。

    from client.client import AbClient
from client.arkai.proxy import LLMProxy


# 初始化AB client
ab_client = AbClient(
    "{token}",
    meta_host="{meta_host}",
    track_host="{track_host}"
)
# 初始化llm client
# api_key和base_url获取参考方舟文档：https://www.volcengine.com/docs/82379/1319847
llm_client = LLMProxy(ab_client).get_client("{api_key}", "{base_url}")


# 对话机器人核心入口
def chatbot_by_llm_proxy():
    # 以model类型做A/B实验，火山方舟对应于model ID或者推理接入点，如下默认值所示
    variant_key, default_value = "llm_model", "ep-xxx"
    # 随机分流获取model类型，注意：这里变体值是火山方舟的模型ID或者推理接入点
    llm_model = ab_client.activate(variant_key, "{decision_id}", "{track_id}", default_value, None)
    messages = [
        {"role": "system", "content": "你是人工智能助手"},
        {"role": "user", "content": "常见的十字花科植物有哪些？"},
    ]
    for chunk in llm_client.stream(llm_model, messages, "{track_id}"):
        yield chunk.choices[0].delta.content

    此方式下，相较于通用A/B实验场景，大模型实验仅需在业务代码中增加一个llm client即可，业务逻辑中的大模型调用代码都和openAI的SDK协议兼容，可以复用。集成后，会自动收集大模型调用的性能指标，并上报到DataTester。

    代码模块

    参数配置要点

    初始化AB client

    {token}、{meta_host}、{track_host}的配置与普通实验SDK集成的配置一致，详情请参见Python SDK。

    注意

    不同环境的host配置方式不一致，请先确认您当前的DataTester为哪种环境后，再进行配置。

    初始化llm client

    说明

    大部分DataTester用户的地域为华北 2 (北京)，您可使用华北 2 (北京)地域的域名即可。

    应用的业务逻辑

    以上示例中：

    • variant_key, default_value = "llm_model", "ep-xxx"：定义了当前实验版本中的实验变量名和取值，本次示例的变量名为创建大模型实验时设置的llm_model，取值为业务使用的大模型类型的model_id或推理接入点，详情请参见model ID或者推理接入点说明文档。
    • {decision_id}、{track_id}：
      • decision_id: 本地分流用户标识，不用于事件上报，需替换为真实用户标识id。
      • track_id: 事件上报用户标识，用于事件上报，需替换为客户的真实用户标识id。

  3. 如果您将大模型相关的配置托管在DataTester的Feature Flag中，还可以通过get_client_by_ff_config方法创建llm client，代码如下：

    # 通过Feature Flag的配置初始化llm client
llm_client = LLMProxy(ab_client).get_client_by_ff_config("{ff_key}", "{decision_id}")

    其中：

    • {ff_key}：为此前已创建的模型配置Feature的key名称，您可在DataTester控制台页面中查看。
      Image
    • {decision_id}：本地分流用户标识，不用于事件上报，需替换为真实用户标识id。

    注意

    将大模型相关的配置托管在DataTester的Feature Flag中时，您需要先在DataTester的Feature列表中创建大模型配置Feature，且创建时，变体类型一定要配置为json，变体参数取值需按要求配置，否则会解析失败，无法创建llm client。

    {
  "api_key": "{api_key}",  // 可选，若不在Feature变体中配置api_key，则需要在实验SDK集成代码中调用get_client_by_ff_config传递default_api_key
  "base_url": "{base_url}",
  "client_options": {    // 可选，支持方舟Ark客户端初始化所需的所有参数，您可按需配置llm client初始化参数
      "max_retries": 3,
      "timeout": 600
  }
}

使用三方AI SDK

以下以火山引擎方舟 SDK 和常见的 OpenAI SDK 为例，为您展示如何集成 SDK 并上报大模型性能指标。

  1. 安装三方AI SDK。
    • 火山方舟SDK ：pip install --upgrade "volcengine-python-sdk[ark]"，详细集成操作请参见火山方舟 Python SDK 安装与初始化。
    • OpenAI SDK： pip install --upgrade "openai>=1.0"，详细集成操作请参见OpenAP API。
    • 若使用poetry，则 poetry add volcengine-python-sdk 或者 poetry add openai。
  2. 编写服务端实验代码。

    注意

    使用三方AI SDK时，服务端代码的编写要点包括：

    • 需要将大模型调用中获取性能数据，构建到DataTester SDK提供的LLMUsage Object中。
      • 在流式和非流式调用中如何收集性能数据，可以参考下文的获取大模型性能指标样例代码章节。
      • DataTester为您提供的一个上报大模型性能数据的预置事件，您可以在业务代码中将需要上报的大模型性能数据构建到LLMUsage Object中，后续这些性能数据会作为大模型预置事件的事件属性，一起上报至DataTester，便于后续您基于上报数据创建并分析大模型性能指标。支持上报的大模型性能数据类型和预置事件的介绍请参见下文的预置大模型性能事件章节。
    • 实际分流后，调用A/B SDK的接口send_llm_usage即可完成性能指标事件的上报。

    以下为服务端代码样例。

    import httpx
from volcenginesdkarkruntime import Ark
from openai import OpenAI
from client import client as ab
from client.arkai.const import LLMUsage

"""LLMUsage结构
@dataclass
class LLMUsage:
    model_name: str      # 大模型名
    prompt_tokens: int     # 输入Tokens
    completion_tokens: int # 输出Tokens
    total_tokens: int     # 总Tokens
    start_time: float     # 开始调用时间
    first_token_time: float # 首Token时间
    reasoning_end_time: float # 推理结束时间
    end_time: float             # 模型调用结束时间
    first_token_latency: float # 首Token延时
    reasoning_latency: float     # 推理延时
    total_latency: float         # 总延时
    status: str = "success"     # 状态,success: 调用成功 error：失败

"""


# 初始化大模型客户端，以下火山方舟和openAI按照自己需求二选一即可
# 初始化方舟客户端
llm_client = Ark(
    # The output time of the reasoning model is relatively long. Please increase the timeout period.
    api_key="{api_key}",  # 火山方舟API Key
    timeout=httpx.Timeout(timeout=1800),
)
# Or 初始化openAI SDK
# llm_client = OpenAI(
#    api_key = "{api_key}",
#    base_url = "https://ark.cn-beijing.volces.com/api/v3",
#)

# 初始化A/B实验客户端
ab_client = ab.AbClient(
    "{app_key}",
    meta_host="{meta_host}",
    track_host="{track_host}"
)  

def call_model(model: str, messages: list):
    # 伪代码调用，这里不在举例
    # 获取completion
    # 收集性能指标，构建到LLMUsage
    
    return completion, llm_usage
    


# 业务做实验场景1: 对话式机器人使用流式调用
def chatbot():  # 对话机器人核心入口
    # 以做不同模型效果实验为例，实验的变体为llm_model
    variant_key, default_llm_model = "llm_model", "deepseek_r1"
    decision_id, user_id = "{descison_id}", "{user_id}"
    
    # 调用分流客户端获取模型配置
    llm_model = ab_client.activate(variant_key, decision_id, user_id, default_llm_model, None)
    
    # 调用大模型获取数据,并计算性能数据
    completion, llm_usage = call_model(llm_model, {messages})
    
    # 上报性能指标事件
    ab_client.send_llm_usage(user_id, llm_usage)

通用参考

获取大模型性能指标样例代码

import httpx
from volcenginesdkarkruntime import Ark
from openai import OpenAI
from client import client as ab
from client.arkai.const import LLMUsage

"""LLMUsage结构
@dataclass
class LLMUsage:
    model_name: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    start_time: float
    fist_token_time: float
    reasoning_end_time: float
    end_time: float
    fist_token_latency: float
    reasoning_latency: float
    total_latency: float
    status: str = "success"

"""


# 初始化大模型客户端，以下火山方舟和openAI按照自己需求二选一即可
# 初始化方舟客户端
llm_client = Ark(
    # The output time of the reasoning model is relatively long. Please increase the timeout period.
    api_key="{api_key}",  # 火山方舟API Key
    timeout=httpx.Timeout(timeout=1800),
)
# Or 初始化openAI SDK
# llm_client = OpenAI(
#    api_key = "{api_key}",
#    base_url = "https://ark.cn-beijing.volces.com/api/v3",
#)

# 初始化A/B实验客户端
ab_client = ab.AbClient(
    "{app_key}",
    meta_host="{meta_host}",
    track_host="{track_host}"
)  



# 大模型调用proxy
# 代码中会详细说明如何获取大模型服务调用的性能数据，并返回给到业务端
# 流式调用
def stream(model: str, messages: list):
    """ stream request
    :param model: model endpoint
    :param messages: input messages， eg:
        [
            {"role": "system", "content": "你是豆包，是由字节跳动开发的 AI 人工智能助手"},
            {"role": "user", "content": "花椰菜是什么？"},
            {"role": "assistant", "content": "花椰菜又称菜花、花菜，是一种常见的蔬菜。"},
        ]
    :return: stream resp
    """
    start_time, first_token_time, reasoning_token_time, usage_chunk = time.time(), None, None, None
    stream = llm_client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
        stream_options={"include_usage": True}
    )
    for chunk in stream:
        if first_token_time is None:
            first_token_time = time.time()
        if not chunk.choices:
            usage_chunk = chunk
            continue
        if hasattr(chunk.choices[0].delta, "reasoning_content"):
            yield getattr(chunk.choices[0].delta, "reasoning_content")
        else:
            if reasoning_token_time is None:
                reasoning_token_time = time.time()
            yield chunk.choices[0].delta.content

    # 返回首 token 耗时、总耗时，token消耗等usage信息
    usage = usage_chunk.usage
    end_time = time.time()
    yield LLMUsage(usage_chunk.model, usage.prompt_tokens, usage.completion_tokens,
                   usage.total_tokens, start_time, first_token_time, reasoning_token_time, end_time,
                   first_token_time-start_time, reasoning_token_time-start_time,
                   end_time-start_time)



# 标准调用，一次性返回所有数据，适用于离线场景
def standard(model: str, messages: list) -> tuple[str, LLMUsage]:
    start_time, reasoning_token_time = time.time(), None
    completion = llm_client.chat.completions.create(
        model=model,
        messages=messages,
    )
    content = ""
    if hasattr(completion.choices[0].message, "reasoning_content"):
        content = getattr(completion.choices[0].message, "reasoning_content")
    if not content:
        content = completion.choices[0].message.content

    usage = completion.usage
    end_time = time.time()
    return content, LLMUsage(completion.model, usage.prompt_tokens, usage.completion_tokens,
                             usage.total_tokens, start_time, 0, 0, end_time, 0, 0, end_time-start_time)
    

# 业务做实验场景1: 对话式机器人使用流式调用
def chatbot():  # 对话机器人核心入口
    # 以做不同模型效果实验为例，实验的变体为llm_model
    variant_key = "llm_model"
    decision_id, user_id = "{descison_id}", "{user_id}"
    default_llm_model = "deepseek_r1"
    
    # 调用分流客户端获取模型配置
    llm_model = ab_client.activate(variant_key, decision_id, user_id, default_llm_model, None)
    
    # 调用大模型获取数据
    for m in stream(llm_model, {messages}):
        if isinstance(m, LLMUsage):
            # 上报性能指标事件
            ab_client.send_llm_usage(user_id, m)
            continue
        yield m

# 业务做实验场景2: 离线分析使用标准调用
def offline_analysis_by_llm():  # 离线分析核心入口
    # 以做不同模型效果实验为例，实验的变体为llm_model
    variant_key = "llm_model"
    decision_id, user_id = "{descison_id}", "{user_id}"
    default_llm_model = "deepseek_r1"
    
    # 调用分流客户端获取模型配置
    llm_model = ab_client.activate(variant_key, decision_id, user_id, default_llm_model, None)
    
    # 调用大模型获取数据
    content, llm_usage = standard(llm_model, {messages})
    # 上报性能指标事件
    ab_client.send_llm_usage(user_id, m)
   
   return content

预置大模型性能事件

用于大模型实验分析指标组创建使用。

{
    "user": {
        "user_unique_id": "{uid}"
    },
    "header": {
        "app_id": {app_id}
    },
    "events": [
        {
            "event": "$llm_usage_log",    // 事件名称
            "local_time_ms": 1489573628001,   //事件发生的时间戳
            "params": "{
                  \"$llm_model_name\":\"deepseek_r1\", // 模型名
                  \"$llm_prompt_tokens\": 1000,        // prompt tokens 
                  \"$llm_completion_tokens\": 2000,     // 输出Tokens
                  \"$llm_total_tokens\": 3000,          // 总tokens
                  \"$llm_first_token_latency\": 2.5,    // 首token耗时(s)
                  \"$llm_reasoning_latency\": 20.15,    // 推理耗时(s)
                  \"$llm_total_latency\": 30.25,        // 总耗时(s)
                  \"$llm_status\": \"success\",         // 请求的状态      
                  \"$llm_start_time\": 1740930479247,
                  \"$llm_first_token_time\": 1740930480247,
                  \"$llm_reasoning_end_time\": 1740930481247,
                  \"$llm_end_time\": 1740930482247
              }"
        }
    ]
}