跳转到主要内容

API 参考

以下是 osmosis-ai Python SDK 的完整接口文档。SDK 围绕两大核心功能组织:

奖励与评估

用于通过本地规则或基于 LLM 的评分标准对 LLM 输出进行评分的装饰器和函数

远程 Rollout

构建与训练基础设施集成的自定义智能体循环

Reward API

装饰器

@osmosis_reward

用于本地奖励函数的装饰器,无需 API 调用即可计算分数。适用于确定性评估逻辑(精确匹配、正则表达式、关键词匹配)。 函数签名:
@osmosis_reward
def function_name(
    solution_str: str,
    ground_truth: str,
    extra_info: dict = None,
    **kwargs
) -> float
参数:
  • solution_str (str, 必填) - 待评估的文本
  • ground_truth (str, 必填) - 参考答案
  • extra_info (dict, 可选) - 额外上下文信息
  • **kwargs (必填) - 未来兼容性(见下方警告)
返回值: float - 分数值 示例:
from osmosis_ai import osmosis_reward

@osmosis_reward
def exact_match(solution_str: str, ground_truth: str, extra_info: dict = None, **kwargs) -> float:
    return 1.0 if solution_str.strip() == ground_truth.strip() else 0.0

@osmosis_rubric

用于基于 LLM 评估函数的装饰器。适用于需要语义理解的主观评估(帮助性、语气、质量)。 函数签名:
@osmosis_rubric
def function_name(
    solution_str: str,
    ground_truth: str | None,
    extra_info: dict,
    **kwargs
) -> float
参数:
  • solution_str (str, 必填) - 待评估的文本
  • ground_truth (str | None, 必填) - 参考答案(可为 None)
  • extra_info (dict, 必填) - 配置和上下文信息
  • **kwargs (必填) - 未来兼容性(见下方警告)
返回值: float - 分数值 示例:
from osmosis_ai import osmosis_rubric, evaluate_rubric

@osmosis_rubric
def quality_check(solution_str: str, ground_truth: str | None, extra_info: dict, **kwargs) -> float:
    return evaluate_rubric(
        rubric="Evaluate response quality",
        solution_str=solution_str,
        model_info={"provider": "openai", "model": "gpt-5.2"},
        ground_truth=ground_truth
    )

评分标准评估

evaluate_rubric()

使用基于 LLM 的评分标准评估文本。这是 LLM 驱动评估的核心函数。 函数签名:
def evaluate_rubric(
    rubric: str,
    solution_str: str,
    model_info: dict,
    ground_truth: str | None = None,
    original_input: str | None = None,
    metadata: dict | None = None,
    score_min: float = 0.0,
    score_max: float = 1.0,
    timeout: int | None = None,
    return_details: bool = False
) -> float | dict
参数:
ParameterTypeRequiredDescription
rubricstr自然语言评估标准
solution_strstr待评估的文本
model_infodictLLM 提供商配置
ground_truthstr参考答案
original_inputstr原始用户查询
metadatadict额外上下文信息
score_minfloat最低分数(默认:0.0)
score_maxfloat最高分数(默认:1.0)
timeoutint请求超时时间(秒)
return_detailsbool返回完整响应(默认:False)
model_info 结构:
{
    "provider": "openai",           # Required
    "model": "gpt-5.2",         # Required
    "api_key": "sk-...",            # Optional
    "api_key_env": "OPENAI_API_KEY", # Optional
    "timeout": 30                   # Optional
}
返回值:
  • float - 分数(当 return_details=False 时)
  • dict - 包含分数、解释、原始数据的完整响应(当 return_details=True 时)
示例:
from osmosis_ai import evaluate_rubric

score = evaluate_rubric(
    rubric="Evaluate how helpful the response is.",
    solution_str="Click 'Forgot Password' to reset.",
    model_info={"provider": "openai", "model": "gpt-5.2"}
)

异常

MissingAPIKeyError

当未找到某个提供商的 API 密钥时抛出。
from osmosis_ai import MissingAPIKeyError

try:
    score = evaluate_rubric(...)
except MissingAPIKeyError as e:
    print(f"API key not found: {e}")

ProviderRequestError

当提供商请求失败时抛出。
from osmosis_ai import ProviderRequestError

try:
    score = evaluate_rubric(...)
except ProviderRequestError as e:
    print(f"Provider error: {e}")

ModelNotFoundError

当指定的模型不可用时抛出(ProviderRequestError 的子类)。
from osmosis_ai import ModelNotFoundError

try:
    score = evaluate_rubric(...)
except ModelNotFoundError as e:
    print(f"Model not found: {e}")

类型

ModelInfo (TypedDict)

from osmosis_ai.rubric_types import ModelInfo

model_info: ModelInfo = {
    "provider": "openai",
    "model": "gpt-5.2",
    "api_key_env": "OPENAI_API_KEY",
    "timeout": 30
}
FieldTypeRequiredDescription
providerstr提供商名称(见支持的提供商)
modelstr模型标识符
api_keystr直接提供 API 密钥值
api_key_envstrAPI 密钥的环境变量名
timeoutfloat请求超时时间(秒)
score_minfloat最低分数覆盖值
score_maxfloat最高分数覆盖值
system_promptstr | None评估器的自定义系统提示词
original_inputstr | None用于上下文的原始用户查询
reasoning_effortstr | None推理努力程度(因提供商而异)

RewardRubricRunResult (TypedDict)

return_details=True 时返回:
from osmosis_ai.rubric_types import RewardRubricRunResult

result: RewardRubricRunResult = {
    "score": 0.85,              # float
    "explanation": "...",       # str
    "raw": {...}                # Any - raw LLM response
}


完整奖励函数示例

from osmosis_ai import osmosis_reward, osmosis_rubric, evaluate_rubric
from dotenv import load_dotenv

load_dotenv()

# Local reward function
@osmosis_reward
def exact_match(solution_str: str, ground_truth: str, extra_info: dict = None, **kwargs) -> float:
    return 1.0 if solution_str.strip() == ground_truth.strip() else 0.0

# Remote rubric evaluator
@osmosis_rubric
def semantic_eval(solution_str: str, ground_truth: str | None, extra_info: dict, **kwargs) -> float:
    return evaluate_rubric(
        rubric="Compare semantic similarity (0-1 scale)",
        solution_str=solution_str,
        ground_truth=ground_truth,
        model_info={"provider": "openai", "model": "gpt-5.2"}
    )

# Usage
solution = "The capital of France is Paris"
truth = "Paris is France's capital"

local_score = exact_match(solution, truth)
semantic_score = semantic_eval(solution, truth, {})

print(f"Exact match: {local_score}")      # 0.0
print(f"Semantic: {semantic_score}")      # ~1.0

Remote Rollout API

构建与 Osmosis 训练基础设施集成的自定义智能体循环。您的智能体作为 HTTP 服务器运行,而训练集群负责 LLM 推理和轨迹收集。

快速概览

ComponentDescription
RolloutAgentLoop实现智能体的基类
RolloutContext包含 chat()complete()error() 方法的执行上下文
RolloutRequest包含消息、参数和元数据的初始请求
create_app()创建 FastAPI 服务器的工厂函数

远程 Rollout 文档

有关完整的 Remote Rollout API 文档、指南和示例,请查看专门的远程 Rollout 章节。

后续步骤