agent_streams

​

AgentStream

• created_at (datetime.datetime): The timestamp when the log stream was created.
• created_by (str): The identifier of the user who created the log stream.
• id (str): The unique identifier of the log stream.
• name (str): The name of the log stream.
• project_id (str): The ID of the project this log stream belongs to.
• updated_at (datetime.datetime): The timestamp when the log stream was last updated.
• additional_properties (dict): Additional properties associated with the log stream.

# Create a new log stream in a project
from splunk_ao.agent_streams import create_agent_stream

# Create by project ID
agent_stream = create_agent_stream(name="Production Logs", project_id="project-123")

# Create by project name
agent_stream = create_agent_stream(name="Production Logs", project_name="My AI Project")

# Get a log stream by name
from splunk_ao.agent_streams import get_agent_stream
agent_stream = get_agent_stream(name="Production Logs", project_name="My AI Project")

# List all agent streams in a project
from splunk_ao.agent_streams import list_agent_streams
agent_streams = list_agent_streams(project_name="My AI Project")
for stream in agent_streams:
    logger.info(f"Agent Stream: {stream.name} (ID: {stream.id})")

# Use a log stream with the context manager
from splunk_ao.openai import openai
from splunk_ao import splunk_ao_context

with splunk_ao_context(project="My AI Project", agent_stream="Production Logs"):
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello, world!"}]
    )

# Enable evaluators on an agent stream - RECOMMENDED APPROACH
from splunk_ao.agent_streams import enable_evaluators
from splunk_ao.schema.metrics import SplunkAOEvaluators

# Set environment variables first
# export SPLUNK_AO_AGENT_STREAM="Production Logs"
# export SPLUNK_AO_PROJECT="My AI Project"

# Clean and simple - just pass the evaluators!
local_evaluators = enable_evaluators(metrics=[
    SplunkAOEvaluators.correctness,
    SplunkAOEvaluators.completeness,
    "context_relevance"
])

# Alternative: Use explicit parameters
local_evaluators = enable_evaluators(
    agent_stream_name="Production Logs",
    project_name="My AI Project",
    metrics=["correctness", "completeness"]
)

​

enable_evaluators

def enable_evaluators(self,
                      metrics: builtins.list[SplunkAOEvaluators | Metric | LocalMetricConfig | str]) -> builtins.list[LocalMetricConfig]

• metrics (builtins.list[Union[SplunkAOEvaluators, Metric, LocalMetricConfig, str]]): List of evaluators to enable on this agent stream. Parameter name retained for API compatibility. Supports multiple input formats:
  • SplunkAOEvaluators enum values: Built-in evaluators like SplunkAOEvaluators.correctness
  • Metric objects: Custom evaluators with optional version specifications
  • LocalMetricConfig objects: Client-side evaluators with custom scoring functions
  • String names: Built-in evaluator names like “correctness” or “toxicity”

• ValueError: - If this AgentStream instance lacks required id or project_id attributes
• If any specified evaluators are unknown or unavailable
• If there are issues with evaluator configuration or registration
• GalileoHTTPException: If there are network or API errors when communicating with Splunk AO services

• builtins.list[LocalMetricConfig]: List of local evaluator configurations that must be computed client-side. Server-side evaluators are automatically registered with Splunk AO and don’t need to be returned since users don’t interact with them.

from splunk_ao.agent_streams import AgentStreams
from splunk_ao.schema.metrics import SplunkAOEvaluators

# Get an agent stream first
agent_streams = AgentStreams()
agent_stream = agent_streams.get(name="Production Logs", project_name="My AI Project")

# Enable evaluators directly - clean and intuitive!
local_evaluators = agent_stream.enable_evaluators([
    SplunkAOEvaluators.correctness,
    SplunkAOEvaluators.completeness,
    "context_relevance",
    "toxicity"
])

logger.info("Server-side evaluators enabled automatically")
logger.info(f"Need to process {len(local_evaluators)} local evaluators")

from splunk_ao.schema.metrics import Metric, LocalMetricConfig

def custom_scorer(trace_or_span):
    return 0.75  # Your scoring logic

local_evaluators = agent_stream.enable_evaluators([
    SplunkAOEvaluators.correctness,
    "completeness",
    Metric(name="domain_relevance", version=3),
    LocalMetricConfig(name="custom_evaluator", scorer_fn=custom_scorer)
])

# Process local evaluators if any
for local_evaluator in local_evaluators:
    logger.info(f"Need to process local evaluator: {local_evaluator.name}")

• The AgentStream instance must have valid id and project_id attributes
• These are automatically set when retrieving AgentStream objects via AgentStreams methods

• Use this method when you already have a AgentStream object
• More intuitive than specifying project/agent stream names again
• Cleaner object-oriented design pattern

​

AgentStreams

​

create

def create(self,
           name: str,
           *,
           project_id: str | None=None,
           project_name: str | None=None) -> AgentStream

• name (str): The name of the log stream.
• project_id (Optional[str]): The ID of the project to create the log stream in. Defaults to None.
• project_name (Optional[str]): The name of the project to create the log stream in. Defaults to None.

• ValueError: If neither or both project_id and project_name are provided, or if the project is not found.
• HTTPValidationError: If the server validation fails.
• errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
• httpx.TimeoutException: If the request takes longer than Client.timeout.

• AgentStream: The created log stream.

​

enable_evaluators

def enable_evaluators(self,
                      *,
                      agent_stream_name: str | None=None,
                      project_name: str | None=None,
                      metrics: builtins.list[SplunkAOEvaluators | Metric | LocalMetricConfig | str]) -> builtins.list[LocalMetricConfig]

• agent_stream_name (Optional[str]): The name of the agent stream. Takes precedence over the SPLUNK_AO_AGENT_STREAM environment variable. Defaults to None.
• project_name (Optional[str]): The name of the project. Takes precedence over the SPLUNK_AO_PROJECT environment variable. Defaults to None.
• metrics (builtins.list[Union[SplunkAOEvaluators, Metric, LocalMetricConfig, str]]): List of evaluators to enable. Parameter name retained for API compatibility. Can include:
  • SplunkAOEvaluators enum values (e.g., SplunkAOEvaluators.correctness)
  • Metric objects with name and optional version
  • LocalMetricConfig objects for custom local evaluators
  • String names of built-in evaluators

• ValueError: If agent stream or project cannot be found, or if evaluators are unknown.

• builtins.list[LocalMetricConfig]: Local evaluator configurations that must be computed client-side. Server-side evaluators are automatically registered with Splunk AO.

# Enable built-in evaluators with explicit parameters
from splunk_ao.agent_streams import AgentStreams
from splunk_ao.schema.metrics import SplunkAOEvaluators

agent_streams = AgentStreams()
local_evaluators = agent_streams.enable_evaluators(
    agent_stream_name="Production Logs",
    project_name="My AI Project",
    metrics=[
        SplunkAOEvaluators.correctness,
        SplunkAOEvaluators.completeness,
        "context_relevance",
    ],
)

# Enable evaluators using environment variables
# export SPLUNK_AO_AGENT_STREAM="Production Logs"
# export SPLUNK_AO_PROJECT="My AI Project"
local_evaluators = agent_streams.enable_evaluators(
    metrics=["correctness", "completeness"]
)

# Enable custom evaluators with mixed parameters
from splunk_ao.schema.metrics import Metric, LocalMetricConfig

def custom_scorer(trace_or_span):
    return 0.85  # Custom scoring logic

# export SPLUNK_AO_PROJECT="My AI Project"
local_evaluators = agent_streams.enable_evaluators(
    agent_stream_name="Production Logs",  # Explicit agent stream
    # project_name from env var
    metrics=[
        Metric(name="my_custom_evaluator", version=2),
        LocalMetricConfig(name="local_evaluator", scorer_fn=custom_scorer)
    ]
)

​

get

def get(self,
        *,
        id: str | None=None,
        name: str | None=None,
        project_id: str | None=None,
        project_name: str | None=None) -> AgentStream | None

• id (Optional[str]): The id of the log stream. Defaults to None.
• name (Optional[str]): The name of the log stream. Defaults to None.
• project_id (Optional[str]): The ID of the project. Defaults to None.
• project_name (Optional[str]): The name of the project. Defaults to None.

• ValueError: If neither or both id and name are provided, or if neither or both project_id and project_name are provided.
• errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
• httpx.TimeoutException: If the request takes longer than Client.timeout.

• Optional[AgentStream]: The log stream if found, None otherwise.

​

list

def list(self,
         *,
         project_id: str | None=None,
         project_name: str | None=None,
         limit: Unset | int=100,
         starting_token: Unset | int=0) -> builtins.list[AgentStream]

• project_id (Optional[str]): The ID of the project to list log streams for.
• project_name (Optional[str]): The name of the project to list log streams for.
• limit (Union[Unset, int]): The maximum number of log streams to return per page. Defaults to 100.
• starting_token (Union[Unset, int]): The pagination token to start from. Defaults to 0 (first page).

• ValueError: If neither or both project_id and project_name are provided, if the named project is not found, if the server returns a validation error, or if the response is unexpectedly empty.
• errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
• httpx.TimeoutException: If the request takes longer than Client.timeout.

• builtins.list[AgentStream]: A page of log streams.

​

create_agent_stream

def create_agent_stream(name: str,
                        project_id: str | None=None,
                        project_name: str | None=None) -> AgentStream

• name (str): The name of the log stream.

• errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
• httpx.TimeoutException: If the request takes longer than Client.timeout.

• AgentStream: The created project.

​

enable_evaluators

def enable_evaluators(*,
                      agent_stream_name: str | None=None,
                      project_name: str | None=None,
                      metrics: builtins.list[SplunkAOEvaluators | Metric | LocalMetricConfig | str]) -> builtins.list[LocalMetricConfig]

• Environment-only: Just pass evaluators, names from env vars (production/CI)
• Explicit parameters: Specify project/agent stream names directly (development)
• Mixed approach: Combine explicit params with environment fallbacks

​

Environment Variables (Optional Fallbacks)

• agent_stream_name (Optional[str]): The name of the agent stream. Takes precedence over SPLUNK_AO_AGENT_STREAM environment variable. If None, will use SPLUNK_AO_AGENT_STREAM env var. Defaults to None.
• project_name (Optional[str]): The name of the project. Takes precedence over SPLUNK_AO_PROJECT environment variable. If None, will use SPLUNK_AO_PROJECT env var. Defaults to None.
• metrics (builtins.list[Union[SplunkAOEvaluators, Metric, LocalMetricConfig, str]]): List of evaluators to enable on the agent stream. Parameter name retained for API compatibility. Can include:
  • SplunkAOEvaluators enum values (e.g., SplunkAOEvaluators.correctness)
  • Metric objects with name and optional version for custom evaluators
  • LocalMetricConfig objects for client-side custom scoring functions
  • String names of built-in evaluators (e.g., “correctness”, “toxicity”)

• ValueError: If agent stream or project cannot be found, or if any specified evaluators are unknown.
• errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
• httpx.TimeoutException: If the request takes longer than Client.timeout.

• builtins.list[LocalMetricConfig]: List of local evaluator configurations that must be computed client-side. Server-side evaluators are automatically registered with Splunk AO and don’t need to be returned since users don’t interact with them.

# Enable built-in evaluators with explicit parameters
from splunk_ao.agent_streams import enable_evaluators
from splunk_ao.schema.metrics import SplunkAOEvaluators

local_evaluators = enable_evaluators(
    agent_stream_name="Production Logs",
    project_name="My AI Project",
    metrics=[
        SplunkAOEvaluators.correctness,
        SplunkAOEvaluators.completeness,
        "context_relevance",
    ],
)

# Enable evaluators using environment variables only
# export SPLUNK_AO_AGENT_STREAM="Production Logs"
# export SPLUNK_AO_PROJECT="My AI Project"
local_evaluators = enable_evaluators(metrics=["correctness", "completeness"])

# Enable custom and local evaluators with environment variable fallbacks
from splunk_ao.schema.metrics import Metric, LocalMetricConfig
from galileo_core.schemas.logging.step import StepType

def response_length_scorer(trace_or_span):
    '''Custom evaluator that scores based on response length'''
    if hasattr(trace_or_span, "output") and trace_or_span.output:
        return min(len(trace_or_span.output) / 100.0, 1.0)  # Normalize 0-1
    return 0.0

local_evaluators = enable_evaluators(
    agent_stream_name="Development Logs",
    metrics=[
        SplunkAOEvaluators.correctness,
        "toxicity",
        Metric(name="my_custom_evaluator", version=2),
        LocalMetricConfig(
            name="response_length",
            scorer_fn=response_length_scorer,
            scorable_types=[StepType.llm],
            aggregatable_types=[StepType.trace],
        ),
    ],
)

# Process local evaluators
for local_evaluator in local_evaluators:
    logger.info(f"Need to process local evaluator: {local_evaluator.name}")

​

get_agent_stream

def get_agent_stream(*,
                     name: str | None=None,
                     project_id: str | None=None,
                     project_name: str | None=None) -> AgentStream | None

• name (Optional[str]): The name of the log stream. Defaults to None.
• project_id (Optional[str]): The ID of the project. Defaults to None.
• project_name (Optional[str]): The name of the project. Defaults to None.

• ValueError: If neither or both project_id and project_name are provided.
• errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
• httpx.TimeoutException: If the request takes longer than Client.timeout.

• Optional[AgentStream]: The log stream if found, None otherwise.

​

list_agent_streams

def list_agent_streams(*,
                       project_id: str | None=None,
                       project_name: str | None=None,
                       limit: Unset | int=100,
                       starting_token: Unset | int=0) -> builtins.list[AgentStream]

• project_id (str): The id of the project.
• project_name (str): The name of the project.
• limit (Union[Unset, int]): The maximum number of log streams to return per page. Defaults to 100.
• starting_token (Union[Unset, int]): The pagination token to start from. Defaults to 0 (first page).

• errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
• httpx.TimeoutException: If the request takes longer than Client.timeout.

• builtins.list[AgentStream]: A page of log streams.