本文为您介绍如何使用 AgentPilot SDK。

AgentPilot SDK 为开发者提供工具、接口和资源，简化应用开发，让开发者可快速集成 PromptPilot 的核心功能，开发自己的 Agent。

功能

优势

基于易用，模块化，易扩展，开放的设计原则，我们尽可能保证：

• 低侵入性（易用）：对于用户已有Agent/Workflow代码，通过修改几行代码就可以实现下文列举功能，集成简单。
• 模块化：可以单独或者任意组合使用，比如上文提到的评估，回流，反馈，prompt优化等模块。
• 易扩展和开放性：支持多种AI模型/接口，多种模态数据，无缝支持通用开源框架（比如LiteLLM, LangChain, LlamaIndex等开源LLM/Agent应用开发框架），并且易于通过已有功能扩展使用场景。

核心概念

任务和版本 (task and version)：
下图为 PromptPilot 的控制台（Console）和 SDK 的对应关系。一个任务Task，通常包含多个Prompt版本。每一个版本围绕一个 "Prompt" 包含相关版本信息，比如task_id, version, prompt 和变量列表，用于评估的标准（criteria），模型名称和配置如 temperature，top_p 等。
为了通过 SDK 发生数据回流，用户一般需要 task_id，version，AGENTPILOT_API_KEY (用于回流服务的鉴权)，ARK_API_KEY (用于方舟上的大模型推理)，即可完成大部分操作。

回流事件 (run)：
大模型的一次调用如果被抽样，就会对应一次“回流事件”，一个回流事件对应平台“批量”页面中的一行数据。

• 回流采样比率：默认采样率为0.1，即每条数据有0.1的概率回流，用户可以根据情况调整。
• 回流数据上限：默认为2000条，即某task_id, version下数据条数 > 2000时，回流会失败并在SDK调用处返回log信息。该限制是与 Console 页面对齐。如需提高上限请提交Oncall工单，研发团队同学会单独对接。
• 当前仅支持“文本理解”和“视觉理解”任务的评分模式。如果需要支持更多任务类型，比如多轮对话和 Tools，请提交工单。
• model_name 为模型名 + 版本，并且只能是 prompt 生成页模型下拉菜单中支持的模型。如果需要支持自定义或者第三方ep，请提交工单。

安装和配置

# 安装SDK
pip install agent-pilot-sdk

# 用于方舟推理服务鉴权的API_KEY
export ARK_API_KEY=xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

# 用于SDK 鉴权
export AGENTPILOT_API_KEY=yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy
export AGENTPILOT_API_URL=https://sd0roa2mjbbc44uo99850.apigateway-cn-beijing.volceapi.com

**AGENTPILOT_API_KEY的获取：**如下图所示，登陆“方舟” -> 左侧“Prompt 管理” -> 点击“Get API Key” -> 点击 “选择使用”，此时会拷贝你的API_KEY。

**task_id的获取：**登陆“方舟” > 左侧“Prompt 调优” > 点击上方加号“+”新建任务 > 选择 “单轮对话任务” > 选择 “评分模式”。新建任务后，点击上方的“任务名”可以获取该任务的task_id。

• 创建任务

• 获取task_id

Task 和 Prompt 管理

如何查看某一个 prompt 版本的相关信息？

调用接口 ap.get_prompt()，输入 task_id 和 version，可以得到该版本相关信息，如下：

import agent_pilot as ap

info = ap.get_prompt(task_id="ta-20250101000000-aaaaa", version="v2")

print(info.task_id)
print(info.version)
print(info.prompt)
print(info.variable_names)
print(info.criteria)
print(info.model_name)
print(info.temperature)
print(info.top_p)

如何列出一个 task 下的所有版本信息？

调用接口 ap.list_prompts()，只需要通过 task_id 即可。

import agent_pilot as ap

info_list = ap.list_prompts(task_id="ta-20250101000000-aaaaa")

如何获得 metric 相关信息？

调用接口 ap.get_metric()，只需要通过 task_id 即可。

import agent_pilot as ap

metric = ap.get_metric(task_id="ta-20250101000000-aaaaa", version="v2")
print(metric.criteria)

回流数据和反馈

如何得创建一个可以回流数据的 Ark client？

调用接口 ap.probing()，对一个 Ark client 打 patch 即可。之后该 client 既可以当作大模型推理调用，也可以回流数据到 Console 某一个 Task。

import agent_pilot as ap
from volcenginesdkarkruntime import Ark
import os

# 创建 client
client = Ark(api_key=os.getenv('ARK_API_KEY'))
ap.probing(object=client)

如何回流数据？

调用接口 ap.render()，渲染 task_id, version，和 variables，得到渲染后的 input_dict，将其传入 Ark client 调用 completion接口。系统会自动获取对应版本的 prompt，拼装成大模型推理请求的输入并得到模型回答。大模型请求的 completion 接口与 Ark client兼容。

import agent_pilot as ap
from volcenginesdkarkruntime import Ark
import os
import time

# 创建 client
client = Ark(api_key=os.getenv('ARK_API_KEY'))
ap.probing(object=client, sample_rate=1.0)

# 渲染推理请求的输入
input_dict = ap.render(
    task_id = "ta-20250101000000-aaaaa",
    version = "v2",
    variables = {
        "变量1": "变量1的内容",
        "变量2": "变量2的内容",
    },
)

# 调用模型推理，和方舟SDK一样
completion = client.chat.completions.create(model='doubao-1.5-pro-32k-250115', **input_dict)
print(completion.choices[0].message)

此时，登录 Console 页面可以看到系统新增了一条数据，包括变量和模型回答。

如何回流人工评分？

打 patch 后的 Ark client 与标准 Ark completion 接口不同，其推理返回值 completion 带有一个回流事件 (run)，作为一条数据的后续追踪 (track) 的唯一标识 (run.id)，用户只要拥有这个标识，即可追加后续操作，比如添加人工评分和AI评分等。
承接上面代码，调用接口 ap.track_feedback()，传入 feedback，其格式是一个 dict，通过字段 human_score, human_analysis, human_confidence即可传入人工评分。回流后，登录 Console 页面可以到上面那条数据，并且添加了评分和评分原因。该功能与 Console 页面手动评分功能对齐。

import agent_pilot as ap
from volcenginesdkarkruntime import Ark
import os
import time

...

# 得到上述回流事件 run 的唯一标识 run.id
run = completion.run
if run is not None:
    print(run.task_id)
    print(run.version)
    print(run.id)

# 追踪并添加人工反馈分数
ap.track_feedback(
    task_id=run.task_id,
    version=run.version,
    run_id=run.id,
    feedback={
        "human_score": 5,
        "human_analysis": "该条数据的人工评分原因",
        "human_confidence": 1.0,
    },
)

time.sleep(2)

在线评估

如何提交在线评估请求？

调用 ap.eval.evaluate() 接口，用户需要输入一个 example 和一个 metric，均为 dict 类型，提供相关字段即可。

import agent_pilot as ap

# 待评估样本
example = {
    "prompt": "这是一个prompt模板，它有{{变量1}}和{{变量2}}",
    "variables": {
        "变量1": "变量1的内容",
        "变量2": "变量2的内容",
    },
    "response": "模型回答的内容",
}

# 评分标准
metric = {
    "criteria": "文本表达流畅、逻辑清晰。回答内容充分，无明显遗漏。",
}

# 调用评估
result = ap.eval.evaluate(
    example=example,
    metric=metric,
)
print(result.score)
print(result.analysis)
print(result.confidence)

特别的，我们提供如下可选参数

• example_id：用户可以通过指定该 id ，将待评估样本和评估结果一一对应起来，方便用户查找。
• reference：如果用户评分标准 criteria 需要将模型回答和参照回答进行比较，可以通过该字段传递该条样本的参照回答，比如 criteria 为 "模型回答与参照回答内容一致。"

import agent_pilot as ap

# 待评估样本
example = {
    "example_id": "ex_id_1",                  # 可选填
    "prompt": "这是一个prompt模板，它有{{变量1}}和{{变量2}}",
    "variables": {
        "变量1": "变量1的内容",
        "变量2": "变量2的内容",
    },
    "response": "模型回答的内容",
    "reference": "参照回答的内容",               # 可选填
}

# 评分标准
metric = {
    "criteria": "模型回答与参照回答内容一致。",
}

# 调用评估
result = ap.eval.evaluate(
    example=example,
    metric=metric,
)
print(result.score)
print(result.analysis)
print(result.confidence)
print(result.example_id == example["example_id"])
print(result.reference == example["reference"])

此外，为了方便用户，我们还支持以下两种方式构建样本

• 以 messages 字段输入，对齐 Ark client 的 messages 字段接口。
• 以 input 字段输入，直接对大模型提问。

# 以 messages 字段直接输入，对齐 Ark client 的 messages 字段接口
example = {
    "messages": [
        {"role": "system", "content": "你是一个客气礼貌的点餐机器人，帮助用户点餐。"},
        {"role": "user", "content": "你们家的水煮鱼辣不辣？"},
        {"role": "assistant", "content": "你好，可以选择微辣，中辣，重辣。"},
        {"role": "user", "content": "帮我来一份微辣。"},
    ],
    "response": "好的，一份水煮鱼微辣，为您点单，请稍后。",
}

# 以 input 字段输入，直接对大模型提问
example = {
    "input": "这是一个prompt模板，它有变量1的内容和变量2的内容",
    "response": "模型回答的内容",
}

如何回流AI评分？

跟回流人工评分一样，通过回流事件 (run) 的唯一标识 (run.id) 追踪 (track) 到该事件，并调用接口 ap.track_feedback()，传入 feedback 添加AI评分作为反馈。
与人工评分不同的是，AI评分需要传入 human_score, human_analysis, human_confidence 字段作为 feedback。回流后登陆 Console 页面可以看到 run.id 所对应的样本更新了评分和评分原因，旁边并显示“AI 评分”字样。该功能与 Console 页面智能评分功能对齐。

import agent_pilot as ap
from volcenginesdkarkruntime import Ark
import os
import time

# 创建 client
client = Ark(api_key=os.getenv('ARK_API_KEY'))
ap.probing(object=client, sample_rate=1.0)

# 渲染推理请求的输入
input_dict = ap.render(
    task_id = "ta-20250603034328-Yabji",
    version = "v1",
    variables = {
        "变量1": "变量1的内容ffff",
        "变量2": "变量2的内容ffff",
    },
)

# 调用模型推理，和方舟SDK一样
completion = client.chat.completions.create(model='doubao-1.5-pro-32k-250115', **input_dict)
print(completion.choices[0].message)

run = completion.run

if run is not None:
    result = ap.eval.evaluate(
        example={
            "example_id": "example_id_1",
            "input": input_dict["messages"][0]["content"][0]["text"],
            "response": completion.choices[0].message.content,
        },
        metric={
            "criteria": "文本表达流畅、逻辑清晰。回答内容充分，无明显遗漏。",
        }
    )
    print(result.score)
    print(result.analysis)
    print(result.confidence)

    ap.track_feedback(
        task_id=run.task_id,
        version=run.version,
        run_id=run.id,
        feedback={
            "metric_score": result.score,
            "metric_analysis": result.analysis,
            "metric_confidence": result.confidence,
        },
    )
    
    time.sleep(2)

如何生成评分标准？

调用接口 ap.eval.generate_criteria()，通过传入一个样本列表 examples 生成评分标准，其中 examples 每一个元素是一个 example dict。该功能对齐智能评分的生成评分标准功能。
特别的，样本列表中每一个样本可以包含评分 score 和评分原因 analysis。生成评分标准时，系统不区分是人工评分还是AI评分。

import agent_pilot as ap

# 样本列表
examples = [
    {
        "prompt": "这是一个prompt模板，它有{{变量1}}和{{变量2}}",
        "variables": {
            "变量1": "样本1中变量1的内容",
            "变量2": "样本1中变量2的内容",
        },
        "response": "模型回答1",
        "reference": "参照回答1",            # 可选填
        "score": 5,                         # 可选填
        "analysis": "评分原因1",             # 可选填
    },
    {
        "prompt": "这是一个prompt模板，它有{{变量1}}和{{变量2}}",
        "variables": {
            "变量1": "样本2中变量1的内容",
            "变量2": "样本2中变量2的内容",
        },
        "response": "模型回答2",
        "reference": "参照回答2",            # 可选填
        "score": 1,                         # 可选填
        "analysis": "评分原因2",             # 可选填
    },
]

criteria = ap.eval.generate_criteria(
    examples=examples,
)
print(criteria)

Prompt 优化和报告获取

如何提交 prompt 优化任务？

调用接口 ap.optimize.create()，传入 task_id 并指定 base_version，系统会创建一个优化 job 并绑定一个 OptimizeJob 对象 opt_job。该对象包含 job_id 信息，作为该优化 job 的唯一标识。

import agent_pilot as ap
import time

# 调用接口 ap.optimize.create() 创建优化 job
opt_job = ap.optimize.create(task_id="ta-20250101000000-aaaaa", base_version="v2")
print(opt_job.job_id)
print(opt_job.task_id)
print(opt_job.base_version)

如何查看优化状态和进度？

通过 opt_job 对象，调用接口 opt_job.get_job_info()，可以查看当前优化 job 信息，包括 state 和 progress。

• 优化结束后的 state 只能为 {ap.OptimizeState.SUCCESS, ap.OptimizeState.FAILED} 中的一个，因此如果不是这两个 state，可以等待一定时间后继续查询。
• 优化运行中的 state 只能为 {ap.OptimizeState.CREATED, ap.OptimizeState.RUNNING} 中的一个。

承接上面代码，每5秒查询一次优化优化 job 信息：

import agent_pilot as ap
import time

...

# 通过 opt_job 对象查询 JobInfo
job_info = opt_job.get_job_info()
while job_info.state not in {ap.OptimizeState.SUCCESS, ap.OptimizeState.FAILED}:
    # 查看当前 state
    print(job_info.state)
    # 查看当前 progress
    if job_info.progress is not None:
        print(job_info.progress.percent)         # 当前优化 job 完成百分比
        print(job_info.progress.optimal_prompt)  # 当前最优的 prompt
        print(job_info.optimized_version)

    # 5秒后重新获取 JobInfo
    time.sleep(5)
    job_info = opt_job.get_job_info()

# 优化 job 结束后，状态为 SUCCESS 或 FAILED
print(job_info.state in {ap.OptimizeState.SUCCESS, ap.OptimizeState.FAILED})

注意：优化任务优化中，也可以通过 PromptPilot 的 Console 界面查看优化进度如下

如何获取优化报告？

如果优化成功，调用接口 opt_job.get_report()，可以获得优化报告。
承接上面代码：

import agent_pilot as ap

...

if job_info.state == ap.OptimizeState.SUCCESS:
    report = opt_job.get_report()
    opt, base = report.opt, report.base

    # 比较优化前后prompt
    print(opt.prompt, base.prompt)

    # 比较优化前后metric
    print(opt.metric, base.metric)

    # 比较优化前后平均分
    print(opt.avg_score, base.avg_score)

    # 列出优化前后data records
    print(opt.records, base.records)

Prompt 生成

调用接口 ap.generate_prompt_stream()，传入 task_description 和模型配置，流式生成 prompt。该功能与 Console 的 Prompt 生成页功能对齐。

import agent_pilot as ap

generated_prompt = ""
usage = None
print("Streaming response:")
for chunk in ap.generate_prompt_stream(
    task_description="我需要一个最强带货，输入是用户提问，输出是产品介绍",
    model_name="doubao-1.5-pro-32k-250115",
    temperature=1.0,
    top_p=0.7,
):
    # 处理流式输出
    generated_prompt += chunk.data.content
    print(chunk.data.content, end="", flush=True)
    if chunk.event == "usage":
        usage = chunk.data.usage

print(f"Generated prompt: {str(generated_prompt)}")
print(f"Usage: {usage}")

Jupyter Notebook 示例

• sdk_demo_open.ipynb：模拟线上生成Prompt，采集数据，实时评估，回流数据，开始优化作业的示例代码
• step_1_create_task_and_upload_short.xlsx ：模拟用户上传到平台的数据
• step_2_use_sdk_to_track_short.xlsx：模拟用户通过回流接口传到平台的数据

sdk_demo_open.ipynb

未知大小

step_1_create_task_and_upload_short.xlsx

未知大小

step_2_use_sdk_to_track_short.xlsx

未知大小