trinity.common.models.experience_extraction module

trinity.common.models.experience_extraction module#

trinity.common.models.experience_extraction.get_routed_experts_layout(model_path: str, trust_remote_code: bool = True) → Tuple[int, int] | None[source]#
trinity.common.models.experience_extraction.decode_sglang_routed_experts(routed_experts_value: Any, total_tokens: int, layout: Tuple[int, int]) → Tensor | None[source]#
trinity.common.models.experience_extraction.decode_vllm_routed_experts(routed_experts_value: str | None) → Tensor | None[source]#
trinity.common.models.experience_extraction.convert_api_output_to_experience(output, multi_modal_inputs: dict[str, Tensor] | None = None, routed_experts_layout: Tuple[int, int] | None = None) → List[Experience][source]#

Convert a non-stream API output to a list of experiences.

Parameters:

  • output – Completion output from API client.

  • multi_modal_inputs – Optional training-time multimodal tensors aligned with the prompt tokens.

  • routed_experts_layout – Optional (num_layers, topk) layout used to decode routed experts.

class trinity.common.models.experience_extraction.HistoryRecordingStream(stream, history: List[Experience], is_async: bool = False)[source]#

Bases: object

__init__(stream, history: List[Experience], is_async: bool = False) → None[source]#
close() → None[source]#
async aclose() → None[source]#
trinity.common.models.experience_extraction.extract_logprobs(choice) → Tensor[source]#