> ## Documentation Index > Fetch the complete documentation index at: https://agent-observability-docs.splunk.com/llms.txt > Use this file to discover all available pages before exploring further. # agent_streams ## AgentStream Log streams are used to organize logs within a project on the Splunk AO platform. They provide a way to categorize and group related logs, making it easier to analyze and monitor specific parts of your application or different environments (e.g., production, staging, development). **Arguments** * `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. **Examples** ```python theme={null} # 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 ```python theme={null} def enable_evaluators(self, metrics: builtins.list[SplunkAOEvaluators | Metric | LocalMetricConfig | str]) -> builtins.list[LocalMetricConfig] ``` Enable evaluators directly on this agent stream instance. This is the most intuitive and clean way to enable evaluators when you already have a AgentStream object. The method leverages the agent stream's existing project\_id and id attributes, eliminating the need for redundant parameter specification and reducing the potential for errors. This approach is ideal for object-oriented workflows where you're working with AgentStream instances directly, and it provides the clearest semantic meaning: "enable these evaluators on this specific agent stream." **Arguments** * `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" **Raises** * `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 **Returns** * `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. **Examples** Basic usage with built-in evaluators: ```python theme={null} 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") ``` Advanced usage with custom evaluators: ```python theme={null} 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}") ``` **Notes** **Requirements:** * The AgentStream instance must have valid `id` and `project_id` attributes * These are automatically set when retrieving AgentStream objects via AgentStreams methods **Recommended Usage:** * 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 ```python theme={null} def create(self, name: str, *, project_id: str | None=None, project_name: str | None=None) -> AgentStream ``` Creates a new log stream. Exactly one of `project_id` or `project_name` must be provided. **Arguments** * `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. **Raises** * `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. **Returns** * `AgentStream`: The created log stream. ### enable\_evaluators ```python theme={null} 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] ``` Enable evaluators for an agent stream by configuring scorers. The project name can be provided via the 'project\_name' parameter or the SPLUNK\_AO\_PROJECT environment variable. The agent stream name can be provided via the 'agent\_stream\_name' parameter or the SPLUNK\_AO\_AGENT\_STREAM environment variable. **Arguments** * `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 **Raises** * `ValueError`: If agent stream or project cannot be found, or if evaluators are unknown. **Returns** * `builtins.list[LocalMetricConfig]`: Local evaluator configurations that must be computed client-side. Server-side evaluators are automatically registered with Splunk AO. **Examples** ```python theme={null} # 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 ```python theme={null} def get(self, *, id: str | None=None, name: str | None=None, project_id: str | None=None, project_name: str | None=None) -> AgentStream | None ``` Retrieves a log stream by id or name. **Arguments** * `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. **Raises** * `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. **Returns** * `Optional[AgentStream]`: The log stream if found, None otherwise. ### list ```python theme={null} 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] ``` Lists log streams. Exactly one of `project_id` or `project_name` must be provided. Returns a single page of results. Use `starting_token` (from `next_starting_token` on a prior response) to fetch subsequent pages. **Arguments** * `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). **Raises** * `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. **Returns** * `builtins.list[AgentStream]`: A page of log streams. ## create\_agent\_stream ```python theme={null} def create_agent_stream(name: str, project_id: str | None=None, project_name: str | None=None) -> AgentStream ``` Creates a new log stream. Exactly one of `project_id` or `project_name` must be provided. **Arguments** * `name` (`str`): The name of the log stream. **Raises** * `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. **Returns** * `AgentStream`: The created project. ## enable\_evaluators ```python theme={null} def enable_evaluators(*, agent_stream_name: str | None=None, project_name: str | None=None, metrics: builtins.list[SplunkAOEvaluators | Metric | LocalMetricConfig | str]) -> builtins.list[LocalMetricConfig] ``` Enable evaluators for an agent stream with flexible parameter and environment variable support. This unified function supports both explicit parameters and environment variable fallbacks, making it perfect for all use cases - from production CI/CD pipelines to development testing. **Flexible Usage Patterns:** * **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) SPLUNK\_AO\_PROJECT : str The name of the Splunk AO project (used when project\_name not provided) SPLUNK\_AO\_AGENT\_STREAM : str The name of the agent stream (used when agent\_stream\_name not provided) **Arguments** * `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") **Raises** * `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. **Returns** * `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. **Examples** ```python theme={null} # 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 ```python theme={null} def get_agent_stream(*, name: str | None=None, project_id: str | None=None, project_name: str | None=None) -> AgentStream | None ``` Retrieves a log stream by name. Exactly one of `project_id` or `project_name` must be provided. **Arguments** * `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. **Raises** * `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. **Returns** * `Optional[AgentStream]`: The log stream if found, None otherwise. ## list\_agent\_streams ```python theme={null} 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] ``` Lists log streams. Exactly one of `project_id` or `project_name` must be provided. Returns a single page of results. Use `starting_token` (from `next_starting_token` on a prior response) to fetch subsequent pages. **Arguments** * `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). **Raises** * `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. **Returns** * `builtins.list[AgentStream]`: A page of log streams.