diff --git a/litellm/integrations/callback_configs.json b/litellm/integrations/callback_configs.json index c2b0c4ddce..3a69c9a793 100644 --- a/litellm/integrations/callback_configs.json +++ b/litellm/integrations/callback_configs.json @@ -104,6 +104,51 @@ }, "description": "Datadog Custom Metrics Integration" }, + { + "id": "galileo", + "displayName": "Galileo", + "logo": "galileo.ico", + "supports_key_team_logging": false, + "dynamic_params": { + "GALILEO_API_KEY": { + "type": "password", + "ui_name": "API Key", + "description": "Galileo Cloud API key (app.galileo.ai). Omit for enterprise username/password auth.", + "required": false + }, + "GALILEO_PROJECT_ID": { + "type": "text", + "ui_name": "Project ID", + "description": "Galileo project ID to log traces to", + "required": true + }, + "GALILEO_LOG_STREAM_ID": { + "type": "text", + "ui_name": "Log Stream ID", + "description": "Galileo log stream ID for v2 spans logging (optional)", + "required": false + }, + "GALILEO_BASE_URL": { + "type": "text", + "ui_name": "Base URL", + "description": "Galileo API base URL (e.g. https://api.galileo.ai for Cloud, or your enterprise API URL)", + "required": false + }, + "GALILEO_USERNAME": { + "type": "text", + "ui_name": "Username", + "description": "Galileo enterprise username (legacy Observe auth; use instead of API key)", + "required": false + }, + "GALILEO_PASSWORD": { + "type": "password", + "ui_name": "Password", + "description": "Galileo enterprise password (legacy Observe auth)", + "required": false + } + }, + "description": "Galileo AI Observability Integration" + }, { "id": "datadog_cost_management", "displayName": "Datadog Cost Management", diff --git a/litellm/integrations/galileo.py b/litellm/integrations/galileo.py index a598124f61..8fef90c24e 100644 --- a/litellm/integrations/galileo.py +++ b/litellm/integrations/galileo.py @@ -1,8 +1,13 @@ +from __future__ import annotations + import json import os import re -from typing import Any, Dict, List, Optional, Tuple, cast +import uuid +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Tuple, Union, cast +import httpx from pydantic import BaseModel, Field import litellm @@ -12,11 +17,15 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import ( convert_content_list_to_str, get_content_from_model_response, ) +from litellm.types.llms.openai import ( + AllMessageValues, + HttpxBinaryResponseContent, + ResponsesAPIResponse, +) from litellm.llms.custom_httpx.http_handler import ( get_async_httpx_client, httpxSpecialProvider, ) -from litellm.types.llms.openai import AllMessageValues GALILEO_CLOUD_API_BASE_URL = "https://api.galileo.ai" # Cap the in-memory buffer so persistent flush failures (e.g. Galileo @@ -33,6 +42,11 @@ class LLMResponse(BaseModel): model: str num_input_tokens: int num_output_tokens: int + num_total_tokens: int + cost: Optional[float] = Field( + default=None, + description="Total cost of the LLM call in USD as computed by LiteLLM.", + ) output_logprobs: Optional[Dict[str, Any]] = Field( default=None, description="Optional. When available, logprobs are used to compute Uncertainty.", @@ -121,10 +135,14 @@ class GalileoObserve(CustomLogger): @staticmethod def _galileo_input_messages( - messages: Optional[List[Any]], input_text: str + messages: Optional[Any], input_text: str ) -> List[Dict[str, str]]: + if isinstance(messages, dict): + messages = messages.get("messages") if not messages: return [{"role": "user", "content": input_text}] + if not isinstance(messages, list): + return [{"role": "user", "content": input_text}] galileo_messages: List[Dict[str, str]] = [] for message in messages: @@ -147,13 +165,59 @@ class GalileoObserve(CustomLogger): return [{"role": "user", "content": input_text}] @staticmethod - def _record_to_v2_span(record: Dict[str, Any]) -> Dict[str, Any]: - created_at = record.get("created_at", "") + def _local_timezone(): + return datetime.now().astimezone().tzinfo or timezone.utc + + @staticmethod + def _format_created_at(dt: Union[datetime, Any]) -> str: + """Serialize timestamps as UTC ISO-8601 for Galileo.""" + if not isinstance(dt, datetime): + return str(dt) + + if dt.tzinfo is None: + # LiteLLM often passes naive datetimes in local time; convert to UTC + # instead of appending Z to local time (which shifts Traces tab sorting). + dt = dt.replace(tzinfo=GalileoObserve._local_timezone()) + + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + @staticmethod + def _normalize_created_at(created_at: str) -> str: if created_at and not re.search(r"(Z|[+-]\d{2}:?\d{2})$", created_at): - created_at = f"{created_at}Z" + return f"{created_at}Z" + return created_at + + @staticmethod + def _token_metrics_from_record(record: Dict[str, Any]) -> Dict[str, Any]: + num_input_tokens = int(record.get("num_input_tokens") or 0) + num_output_tokens = int(record.get("num_output_tokens") or 0) + num_total_tokens = int(record.get("num_total_tokens") or 0) + if num_total_tokens == 0 and (num_input_tokens or num_output_tokens): + num_total_tokens = num_input_tokens + num_output_tokens + metrics: Dict[str, Any] = { + "num_input_tokens": num_input_tokens, + "num_output_tokens": num_output_tokens, + "num_total_tokens": num_total_tokens, + } + cost = record.get("cost") + if cost is not None: + metrics["cost"] = float(cost) + return metrics + + @staticmethod + def _record_to_v2_span( + record: Dict[str, Any], + *, + trace_id: str, + span_id: str, + ) -> Dict[str, Any]: + created_at = GalileoObserve._normalize_created_at(record.get("created_at", "")) span: Dict[str, Any] = { "type": "llm", + "id": span_id, + "trace_id": trace_id, + "parent_id": trace_id, "name": record.get("node_type", "litellm"), "created_at": created_at, "input": GalileoObserve._galileo_input_messages( @@ -167,14 +231,49 @@ class GalileoObserve(CustomLogger): "model": record.get("model"), "metrics": { "duration_ns": int(record.get("latency_ms", 0)) * 1_000_000, - "num_input_tokens": record.get("num_input_tokens"), - "num_output_tokens": record.get("num_output_tokens"), + **GalileoObserve._token_metrics_from_record(record), }, } if record.get("tags"): span["tags"] = record["tags"] return span + @staticmethod + def _record_to_v2_trace(record: Dict[str, Any]) -> Dict[str, Any]: + trace_id = str(uuid.uuid4()) + span_id = str(uuid.uuid4()) + created_at = GalileoObserve._normalize_created_at(record.get("created_at", "")) + + return { + "type": "trace", + "id": trace_id, + "name": record.get("node_type", "litellm"), + "created_at": created_at, + "input": record.get("input_text", ""), + "output": record.get("output_text", ""), + "status_code": record.get("status_code", 200), + "metrics": { + "duration_ns": int(record.get("latency_ms", 0)) * 1_000_000, + **GalileoObserve._token_metrics_from_record(record), + }, + "spans": [ + GalileoObserve._record_to_v2_span( + record, trace_id=trace_id, span_id=span_id + ) + ], + } + + def _build_traces_payload(self, records: List[dict]) -> Dict[str, Any]: + payload: Dict[str, Any] = { + "traces": [self._record_to_v2_trace(record) for record in records], + "logging_method": "api_direct", + "reliable": False, + "is_complete": True, + } + if self.log_stream_id: + payload["log_stream_id"] = self.log_stream_id + return payload + def _get_ingest_request(self) -> Optional[Tuple[str, Dict[str, Any]]]: if not self.base_url or not self.project_id: return None @@ -184,105 +283,457 @@ class GalileoObserve(CustomLogger): # flush_in_memory_records) aren't silently dropped when we later clear # the in-memory buffer. records = list(self.in_memory_records) + payload = self._build_traces_payload(records) if self.use_v2_api: - payload: Dict[str, Any] = { - "spans": [self._record_to_v2_span(record) for record in records], - "reliable": False, - } - if self.log_stream_id: - payload["log_stream_id"] = self.log_stream_id return ( - f"{self.base_url}/v2/projects/{self.project_id}/spans", + f"{self.base_url}/ingest/traces/{self.project_id}", payload, ) + # Username/password auth logs in for a JWT and uses the standard v2 traces API. return ( - f"{self.base_url}/projects/{self.project_id}/observe/ingest", - {"records": records}, + f"{self.base_url}/v2/projects/{self.project_id}/traces", + payload, ) + @staticmethod + def _redact_headers(headers: Optional[Dict[str, str]]) -> Dict[str, str]: + if not headers: + return {} + redacted: Dict[str, str] = {} + for key, value in headers.items(): + if key.lower() in {"authorization", "galileo-api-key"} and value: + redacted[key] = ( + f"{value[:8]}...{value[-4:]}" if len(value) > 12 else "***" + ) + else: + redacted[key] = value + return redacted + + def _log_flush_config(self) -> None: + verbose_logger.debug( + "Galileo Logger flush config: use_v2_api=%s base_url=%s project_id=%s " + "log_stream_id=%s api_key_set=%s username_set=%s record_count=%s", + self.use_v2_api, + self.base_url, + self.project_id, + self.log_stream_id, + bool(self.api_key), + bool(self.username), + len(self.in_memory_records), + ) + + @staticmethod + def _log_v2_payload_validation(payload: Dict[str, Any]) -> None: + missing_fields: List[str] = [] + traces = payload.get("traces", []) + if not traces: + missing_fields.append("traces") + + for trace_index, trace in enumerate(traces): + if not isinstance(trace, dict): + continue + for field in ("id", "type", "spans"): + if field not in trace: + missing_fields.append(f"traces[{trace_index}].{field}") + + trace_id = trace.get("id") + for span_index, span in enumerate(trace.get("spans", [])): + if not isinstance(span, dict): + continue + for field in ("id", "trace_id", "parent_id"): + if field not in span: + missing_fields.append( + f"traces[{trace_index}].spans[{span_index}].{field}" + ) + if trace_id and span.get("trace_id") != trace_id: + missing_fields.append( + f"traces[{trace_index}].spans[{span_index}].trace_id mismatch" + ) + + if missing_fields: + verbose_logger.debug( + "Galileo Logger: ingest /traces payload validation issues: %s", + missing_fields, + ) + + def _log_flush_payload(self, url: str, payload: Dict[str, Any]) -> None: + traces = payload.get("traces", []) + verbose_logger.debug( + "Galileo Logger flush URL: %s trace_count=%s", + url, + len(traces) if isinstance(traces, list) else 0, + ) + if self.use_v2_api and "/ingest/traces/" in url: + self._log_v2_payload_validation(payload) + + @staticmethod + def _log_http_status_error(error: httpx.HTTPStatusError, url: str) -> None: + response = error.response + verbose_logger.debug( + "Galileo Logger HTTP error: status=%s url=%s", + response.status_code, + url, + ) + verbose_logger.debug( + "Galileo Logger HTTP error response body: %s", + response.text, + ) + try: + verbose_logger.debug( + "Galileo Logger HTTP error response json: %s", + response.json(), + ) + except Exception: + pass + + @staticmethod + def _build_prompt(kwargs: Dict[str, Any]) -> Dict[str, Any]: + optional_params = kwargs.get("optional_params", {}) or {} + prompt: Dict[str, Any] = {"messages": kwargs.get("messages")} + if optional_params.get("functions") is not None: + prompt["functions"] = optional_params["functions"] + if optional_params.get("tools") is not None: + prompt["tools"] = optional_params["tools"] + return prompt + + @staticmethod + def _serialize_galileo_output(value: Any) -> Optional[str]: + if value is None: + return None + if isinstance(value, str): + return value + + def _json_default(obj: Any) -> Any: + if hasattr(obj, "model_dump"): + return obj.model_dump() + return str(obj) + + return json.dumps(value, default=_json_default) + + @staticmethod + def _prompt_to_input_text(prompt: Dict[str, Any]) -> str: + messages = prompt.get("messages") + if messages is not None: + text = GalileoObserve._input_text_from_messages(messages) + if text: + return text + return json.dumps(prompt, default=str) + + @staticmethod + def _get_chat_content_for_galileo(response_obj: litellm.ModelResponse) -> Any: + if response_obj.choices and len(response_obj.choices) > 0: + message = response_obj["choices"][0]["message"] + if hasattr(message, "json"): + message_json = message.json() + if isinstance(message_json, str): + return json.loads(message_json) + return message_json + return message + return None + + @staticmethod + def _get_text_completion_content_for_galileo( + response_obj: litellm.TextCompletionResponse, + ) -> Optional[str]: + if response_obj.choices and len(response_obj.choices) > 0: + return response_obj.choices[0].text + return None + + @staticmethod + def _get_responses_api_content_for_galileo( + response_obj: ResponsesAPIResponse, + ) -> Any: + if hasattr(response_obj, "output") and response_obj.output: + return response_obj.output + return None + + @staticmethod + def _langfuse_style_rerank_prompt(kwargs: Dict[str, Any]) -> Dict[str, Any]: + """Match Langfuse rerank input: prompt = {"messages": kwargs.get("messages")}.""" + return {"messages": kwargs.get("messages")} + + def _get_galileo_input_output_content( + self, + kwargs: Dict[str, Any], + response_obj: Any, + level: str = "DEFAULT", + status_message: Optional[str] = None, + ) -> Tuple[str, Optional[str], Any]: + """ + Mirror Langfuse _get_langfuse_input_output_content for Galileo ingest. + + Returns (input_text, output_text, messages_for_span). output_text None skips ingest. + """ + call_type = kwargs.get("call_type") + prompt = self._build_prompt(kwargs) + + if ( + level == "ERROR" + and status_message is not None + and isinstance(status_message, str) + ): + return self._prompt_to_input_text(prompt), status_message, prompt + + if response_obj is not None and ( + call_type == "embedding" + or isinstance(response_obj, litellm.EmbeddingResponse) + ): + return self._prompt_to_input_text(prompt), None, prompt + + if response_obj is not None and isinstance(response_obj, litellm.ModelResponse): + output = self._get_chat_content_for_galileo(response_obj) + return ( + self._prompt_to_input_text(prompt), + self._serialize_galileo_output(output), + kwargs.get("messages") or [], + ) + + if response_obj is not None and isinstance( + response_obj, HttpxBinaryResponseContent + ): + return self._prompt_to_input_text(prompt), "speech-output", prompt + + if response_obj is not None and isinstance( + response_obj, litellm.TextCompletionResponse + ): + output = self._get_text_completion_content_for_galileo(response_obj) + return ( + self._prompt_to_input_text(prompt), + self._serialize_galileo_output(output), + kwargs.get("messages") or [], + ) + + if response_obj is not None and isinstance(response_obj, litellm.ImageResponse): + output = response_obj.get("data", None) + return ( + self._prompt_to_input_text(prompt), + self._serialize_galileo_output(output), + prompt, + ) + + if response_obj is not None and isinstance( + response_obj, litellm.TranscriptionResponse + ): + output = response_obj.get("text", None) + return ( + self._prompt_to_input_text(prompt), + self._serialize_galileo_output(output), + prompt, + ) + + if response_obj is not None and isinstance( + response_obj, litellm.RerankResponse + ): + output = response_obj.results + rerank_prompt = self._langfuse_style_rerank_prompt(kwargs) + return ( + json.dumps(rerank_prompt, default=str), + self._serialize_galileo_output(output), + rerank_prompt, + ) + + if response_obj is not None and isinstance(response_obj, ResponsesAPIResponse): + output = self._get_responses_api_content_for_galileo(response_obj) + return ( + self._prompt_to_input_text(prompt), + self._serialize_galileo_output(output), + kwargs.get("messages") or [], + ) + + if ( + call_type == "_arealtime" + and response_obj is not None + and isinstance(response_obj, list) + ): + input_val = kwargs.get("input") + return ( + self._serialize_galileo_output(input_val) or "", + self._serialize_galileo_output(response_obj), + input_val, + ) + + if ( + call_type == "pass_through_endpoint" + and response_obj is not None + and isinstance(response_obj, dict) + ): + output = response_obj.get("response", "") + return ( + self._prompt_to_input_text(prompt), + self._serialize_galileo_output(output), + prompt, + ) + + if response_obj is not None and isinstance(response_obj, dict): + output = get_content_from_model_response(response_obj) + return ( + self._prompt_to_input_text(prompt), + self._serialize_galileo_output(output), + kwargs.get("messages") or [], + ) + + return self._prompt_to_input_text(prompt), None, kwargs.get("messages") or [] + def get_output_str_from_response( self, response_obj: Any, kwargs: Dict[str, Any] ) -> Optional[str]: - if response_obj is None: - return None - if kwargs.get("call_type", None) == "embedding" or isinstance( - response_obj, litellm.EmbeddingResponse - ): - return None - if isinstance(response_obj, litellm.TextCompletionResponse): - return response_obj.choices[0].text - if isinstance(response_obj, litellm.ImageResponse): - return json.dumps(response_obj["data"], default=str) - if isinstance(response_obj, (litellm.ModelResponse, dict)): - return get_content_from_model_response(response_obj) - return None + _, output_text, _ = self._get_galileo_input_output_content( + kwargs=kwargs, response_obj=response_obj + ) + return output_text + + @staticmethod + def _input_text_from_messages(messages: Any) -> str: + """Return a plain-string summary of the input suitable for the trace-level input field.""" + if isinstance(messages, str): + return messages + if not isinstance(messages, list): + return "" + # Use the last user/human message so the trace table shows the actual prompt + for msg in reversed(messages): + if not isinstance(msg, dict): + continue + if str(msg.get("role", "")).lower() in ("user", "human"): + content = msg.get("content") or "" + if isinstance(content, list): + content = " ".join( + b.get("text", "") if isinstance(b, dict) else str(b) + for b in content + ) + if content: + return str(content) + # Fallback: first non-empty content of any role + for msg in messages: + if isinstance(msg, dict): + content = msg.get("content") or "" + if isinstance(content, list): + content = " ".join( + b.get("text", "") if isinstance(b, dict) else str(b) + for b in content + ) + if content: + return str(content) + return "" async def async_log_success_event( self, kwargs: Any, response_obj: Any, start_time: Any, end_time: Any ): verbose_logger.debug("On Async Success") + try: + await self._async_log_success_event_impl( + kwargs=kwargs, + response_obj=response_obj, + start_time=start_time, + end_time=end_time, + ) + except Exception: + verbose_logger.exception( + "Galileo Logger: unexpected error in async_log_success_event" + ) + async def _async_log_success_event_impl( + self, kwargs: Any, response_obj: Any, start_time: Any, end_time: Any + ): if not self._is_configured(): verbose_logger.debug( - "Galileo Logger: skipping flush — set GALILEO_PROJECT_ID and " - "either GALILEO_API_KEY (hosted) or GALILEO_USERNAME/GALILEO_PASSWORD " - "(enterprise Observe)." + "Galileo Logger: skipping — GALILEO_PROJECT_ID=%s GALILEO_API_KEY=%s GALILEO_BASE_URL=%s", + bool(self.project_id), + bool(self.api_key), + bool(self.base_url), ) return - _latency_ms = int((end_time - start_time).total_seconds() * 1000) - _call_type = kwargs.get("call_type", "litellm") - input_text = litellm.utils.get_formatted_prompt( - data=kwargs, call_type=_call_type + slo: Optional[Dict[str, Any]] = kwargs.get("standard_logging_object") + if slo is None: + verbose_logger.debug( + "Galileo Logger: no standard_logging_object in kwargs, skipping" + ) + return + + _call_type: str = str( + slo.get("call_type") or kwargs.get("call_type") or "litellm" ) - _usage = response_obj.get("usage", {}) or {} - num_input_tokens = _usage.get("prompt_tokens", 0) - num_output_tokens = _usage.get("completion_tokens", 0) + input_text, output_text, messages = self._get_galileo_input_output_content( + kwargs=kwargs, response_obj=response_obj + ) + if output_text is None: + verbose_logger.debug( + "Galileo Logger: skipping %s — no text output to log", _call_type + ) + return - output_text = self.get_output_str_from_response( - response_obj=response_obj, kwargs=kwargs + raw_start = slo.get("startTime") + raw_end = slo.get("endTime") + if raw_start is None or raw_end is None: + verbose_logger.debug( + "Galileo Logger: standard_logging_object missing startTime/endTime, " + "falling back to start_time/end_time params" + ) + if not isinstance(start_time, datetime) or not isinstance( + end_time, datetime + ): + return + start_ts = start_time + end_ts = end_time + if start_ts.tzinfo is None: + start_ts = start_ts.replace(tzinfo=GalileoObserve._local_timezone()) + if end_ts.tzinfo is None: + end_ts = end_ts.replace(tzinfo=GalileoObserve._local_timezone()) + start_ts = start_ts.astimezone(timezone.utc) + end_ts = end_ts.astimezone(timezone.utc) + else: + start_ts = datetime.fromtimestamp(float(raw_start), tz=timezone.utc) + end_ts = datetime.fromtimestamp(float(raw_end), tz=timezone.utc) + _latency_ms = max(0, int((end_ts - start_ts).total_seconds() * 1000)) + num_input_tokens = int(slo.get("prompt_tokens") or 0) + num_output_tokens = int(slo.get("completion_tokens") or 0) + num_total_tokens = int(slo.get("total_tokens") or 0) + if num_total_tokens == 0 and (num_input_tokens or num_output_tokens): + num_total_tokens = num_input_tokens + num_output_tokens + + request_record = LLMResponse( + latency_ms=_latency_ms, + status_code=200, + input_text=input_text, + output_text=output_text, + node_type=_call_type, + model=str(slo.get("model") or kwargs.get("model") or "-"), + num_input_tokens=num_input_tokens, + num_output_tokens=num_output_tokens, + num_total_tokens=num_total_tokens, + cost=slo.get("response_cost"), + created_at=GalileoObserve._format_created_at(start_ts), ) - if output_text is not None: - request_record = LLMResponse( - latency_ms=_latency_ms, - status_code=200, - input_text=input_text, - output_text=output_text, - node_type=_call_type, - model=kwargs.get("model", "-"), - num_input_tokens=num_input_tokens, - num_output_tokens=num_output_tokens, - created_at=start_time.strftime( - "%Y-%m-%dT%H:%M:%S" - ), # timestamp str constructed in "%Y-%m-%dT%H:%M:%S" format + request_dict = request_record.model_dump() + if isinstance(messages, dict): + messages = messages.get("messages") + if isinstance(messages, list) and messages: + request_dict["messages"] = messages + self.in_memory_records.append(request_dict) + verbose_logger.debug( + "Galileo Logger: queued record, in_memory=%d", len(self.in_memory_records) + ) + + # Bound the buffer so persistent flush failures cannot grow it + # without limit. Drop the oldest records once we exceed the cap. + if len(self.in_memory_records) > GALILEO_MAX_IN_MEMORY_RECORDS: + dropped = len(self.in_memory_records) - GALILEO_MAX_IN_MEMORY_RECORDS + self.in_memory_records = self.in_memory_records[ + -GALILEO_MAX_IN_MEMORY_RECORDS: + ] + verbose_logger.warning( + "Galileo Logger: in-memory buffer exceeded %s records; " + "dropped %s oldest record(s). Check Galileo connectivity/credentials.", + GALILEO_MAX_IN_MEMORY_RECORDS, + dropped, ) - request_dict = request_record.model_dump() - messages = kwargs.get("messages") - if messages: - request_dict["messages"] = messages - self.in_memory_records.append(request_dict) - - # Bound the buffer so persistent flush failures cannot grow it - # without limit. Drop the oldest records once we exceed the cap. - if len(self.in_memory_records) > GALILEO_MAX_IN_MEMORY_RECORDS: - dropped = len(self.in_memory_records) - GALILEO_MAX_IN_MEMORY_RECORDS - self.in_memory_records = self.in_memory_records[ - -GALILEO_MAX_IN_MEMORY_RECORDS: - ] - verbose_logger.warning( - "Galileo Logger: in-memory buffer exceeded %s records; " - "dropped %s oldest record(s). Check Galileo connectivity/credentials.", - GALILEO_MAX_IN_MEMORY_RECORDS, - dropped, - ) - - if len(self.in_memory_records) >= self.batch_size: - await self.flush_in_memory_records() + if len(self.in_memory_records) >= self.batch_size: + await self.flush_in_memory_records() async def flush_in_memory_records(self): if not self.in_memory_records: @@ -296,15 +747,23 @@ class GalileoObserve(CustomLogger): ingest_request = self._get_ingest_request() if ingest_request is None: verbose_logger.debug( - "Galileo Logger: missing GALILEO_BASE_URL or GALILEO_PROJECT_ID" + "Galileo Logger: missing GALILEO_BASE_URL or GALILEO_PROJECT_ID — skipping flush" ) return if not await self._ensure_headers(): - verbose_logger.debug("Galileo Logger: could not set request headers") + verbose_logger.debug( + "Galileo Logger: could not set request headers — skipping flush" + ) return url, payload = ingest_request + self._log_flush_config() + self._log_flush_payload(url=url, payload=payload) + verbose_logger.debug( + "Galileo Logger flush headers: %s", + self._redact_headers(self.headers), + ) verbose_logger.debug("flushing in memory records to %s", url) try: @@ -313,6 +772,12 @@ class GalileoObserve(CustomLogger): headers=self.headers, json=payload, ) + except httpx.HTTPStatusError as e: + self._log_http_status_error(error=e, url=url) + verbose_logger.debug( + "Galileo Logger: failed to flush in memory records: %s", e + ) + return except Exception as e: verbose_logger.debug( "Galileo Logger: failed to flush in memory records: %s", e @@ -323,6 +788,11 @@ class GalileoObserve(CustomLogger): verbose_logger.debug( "Galileo Logger: successfully flushed in memory records" ) + verbose_logger.debug( + "Galileo Logger flush response: status=%s body=%s", + response.status_code, + response.text, + ) del self.in_memory_records[:records_in_payload] else: verbose_logger.debug("Galileo Logger: failed to flush in memory records") diff --git a/litellm/proxy/_experimental/out/assets/logos/galileo.ico b/litellm/proxy/_experimental/out/assets/logos/galileo.ico new file mode 100644 index 0000000000..c50b9de4df Binary files /dev/null and b/litellm/proxy/_experimental/out/assets/logos/galileo.ico differ diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 6de8f0a3cd..5059a6f2e5 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -3516,6 +3516,19 @@ class AllCallbacks(LiteLLMPydanticObjectBase): ui_callback_name="Traceloop", ) + galileo: CallbackOnUI = CallbackOnUI( + litellm_callback_name="galileo", + litellm_callback_params=[ + "GALILEO_API_KEY", + "GALILEO_PROJECT_ID", + "GALILEO_LOG_STREAM_ID", + "GALILEO_BASE_URL", + "GALILEO_USERNAME", + "GALILEO_PASSWORD", + ], + ui_callback_name="Galileo", + ) + class SpendLogsMetadata(TypedDict): """ diff --git a/tests/test_litellm/integrations/test_galileo.py b/tests/test_litellm/integrations/test_galileo.py index aab220f46b..8ce5eb776f 100644 --- a/tests/test_litellm/integrations/test_galileo.py +++ b/tests/test_litellm/integrations/test_galileo.py @@ -1,5 +1,6 @@ import os import sys +from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -7,6 +8,8 @@ import pytest sys.path.insert(0, os.path.abspath("../..")) from litellm.integrations.galileo import GalileoObserve +from litellm.types.llms.openai import HttpxBinaryResponseContent, ResponsesAPIResponse +from litellm.types.rerank import RerankResponse from litellm.types.utils import ( Choices, EmbeddingResponse, @@ -15,6 +18,7 @@ from litellm.types.utils import ( Message, ModelResponse, TextCompletionResponse, + TranscriptionResponse, ) @@ -46,16 +50,97 @@ async def test_galileo_v2_ingest_url_and_headers(galileo_v2_env): url, payload = logger._get_ingest_request() assert ( url - == "https://api.galileo.ai/v2/projects/86ff8ebe-a297-4134-b167-748bdd8d2c20/spans" + == "https://api.galileo.ai/ingest/traces/86ff8ebe-a297-4134-b167-748bdd8d2c20" ) assert payload["log_stream_id"] == "76c4ea50-8aa3-4771-a0d7-8567b112210f" - assert payload["spans"][0]["type"] == "llm" - assert payload["spans"][0]["output"]["content"] == "hello" + assert payload["is_complete"] is True + assert payload["traces"][0]["type"] == "trace" + assert payload["traces"][0]["spans"][0]["type"] == "llm" + assert payload["traces"][0]["spans"][0]["output"]["content"] == "hello" + assert payload["traces"][0]["spans"][0]["metrics"]["num_total_tokens"] == 3 + assert payload["traces"][0]["metrics"]["num_input_tokens"] == 1 + assert payload["traces"][0]["metrics"]["num_output_tokens"] == 2 + assert payload["traces"][0]["metrics"]["num_total_tokens"] == 3 + assert payload["traces"][0]["spans"][0]["trace_id"] == payload["traces"][0]["id"] assert await logger._ensure_headers() is True assert logger.headers["Galileo-API-Key"] == "test-api-key" +def test_galileo_token_metrics_from_record_falls_back_to_sum(): + metrics = GalileoObserve._token_metrics_from_record( + {"num_input_tokens": 5, "num_output_tokens": 7} + ) + assert metrics == { + "num_input_tokens": 5, + "num_output_tokens": 7, + "num_total_tokens": 12, + } + + +def test_galileo_token_metrics_from_record_sums_zero_total(): + metrics = GalileoObserve._token_metrics_from_record( + {"num_input_tokens": 5, "num_output_tokens": 7, "num_total_tokens": 0} + ) + assert metrics == { + "num_input_tokens": 5, + "num_output_tokens": 7, + "num_total_tokens": 12, + } + + +def test_galileo_token_metrics_from_record_includes_cost(): + metrics = GalileoObserve._token_metrics_from_record( + { + "num_input_tokens": 1, + "num_output_tokens": 2, + "num_total_tokens": 3, + "cost": 0.000855, + } + ) + assert metrics["cost"] == 0.000855 + + +def test_galileo_input_text_from_messages(): + assert GalileoObserve._input_text_from_messages("hello") == "hello" + assert ( + GalileoObserve._input_text_from_messages( + [{"role": "user", "content": "test responses api 1"}] + ) + == "test responses api 1" + ) + + +def test_galileo_get_output_str_responses_api(galileo_v2_env): + from litellm.types.llms.openai import ResponsesAPIResponse + + logger = GalileoObserve() + resp_dict = { + "id": "resp_123", + "created_at": 1, + "output": [ + { + "id": "msg_1", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "output_text", + "text": "Hi! How can I help?", + "annotations": [], + } + ], + } + ], + } + response = ResponsesAPIResponse(**resp_dict) + result = logger.get_output_str_from_response(response, {"call_type": "aresponses"}) + assert result is not None + assert '"Hi! How can I help?"' in result + assert '"type": "message"' in result + + def test_galileo_v2_span_preserves_message_roles(galileo_v2_env): record = { "latency_ms": 1, @@ -72,7 +157,36 @@ def test_galileo_v2_span_preserves_message_roles(galileo_v2_env): {"role": "user", "content": "hello"}, ], } - span = GalileoObserve._record_to_v2_span(record) + span = GalileoObserve._record_to_v2_span( + record, trace_id="trace-id", span_id="span-id" + ) + assert span["input"] == [ + {"role": "system", "content": "be helpful"}, + {"role": "user", "content": "hello"}, + ] + + +def test_galileo_v2_span_unwraps_prompt_messages(galileo_v2_env): + record = { + "latency_ms": 1, + "status_code": 200, + "input_text": "fallback", + "output_text": "ok", + "node_type": "pass_through_endpoint", + "model": "gpt-5.2", + "num_input_tokens": 0, + "num_output_tokens": 0, + "created_at": "2026-05-25T12:00:00", + "messages": { + "messages": [ + {"role": "system", "content": "be helpful"}, + {"role": "user", "content": "hello"}, + ] + }, + } + span = GalileoObserve._record_to_v2_span( + record, trace_id="trace-id", span_id="span-id" + ) assert span["input"] == [ {"role": "system", "content": "be helpful"}, {"role": "user", "content": "hello"}, @@ -94,7 +208,8 @@ def test_galileo_output_text_from_model_response(galileo_v2_env): ) output = logger.get_output_str_from_response(response, {"call_type": "acompletion"}) - assert output == "assistant reply" + assert output is not None + assert '"assistant reply"' in output @pytest.mark.asyncio @@ -199,6 +314,19 @@ def test_galileo_input_messages_fallbacks(): ) == [{"role": "user", "content": "fallback"}] +def test_galileo_format_created_at_converts_local_naive_to_utc(): + from datetime import timedelta + + ist = timezone(timedelta(hours=5, minutes=30)) + + with patch.object(GalileoObserve, "_local_timezone", return_value=ist): + local_naive = datetime(2026, 6, 4, 9, 44, 49) + assert GalileoObserve._format_created_at(local_naive) == "2026-06-04T04:14:49Z" + + aware_utc = datetime(2026, 6, 4, 4, 14, 49, tzinfo=timezone.utc) + assert GalileoObserve._format_created_at(aware_utc) == "2026-06-04T04:14:49Z" + + def test_galileo_record_to_v2_span_with_tags_and_offset(): span = GalileoObserve._record_to_v2_span( { @@ -212,13 +340,17 @@ def test_galileo_record_to_v2_span_with_tags_and_offset(): "num_output_tokens": 2, "created_at": "2026-05-25T12:00:00", "tags": ["t1"], - } + }, + trace_id="trace-id", + span_id="span-id", ) assert span["tags"] == ["t1"] assert span["created_at"].endswith("Z") offset = GalileoObserve._record_to_v2_span( - {"created_at": "2026-05-25T12:00:00-05:00"} + {"created_at": "2026-05-25T12:00:00-05:00"}, + trace_id="trace-id", + span_id="span-id", ) assert offset["created_at"] == "2026-05-25T12:00:00-05:00" @@ -243,9 +375,118 @@ def test_galileo_get_output_str_variants(galileo_v2_env): image_resp = ImageResponse(data=[ImageObject(url="https://x/y.png")]) assert "y.png" in logger.get_output_str_from_response(image_resp, {}) + speech_resp = HttpxBinaryResponseContent(response=MagicMock()) + assert ( + logger.get_output_str_from_response(speech_resp, {"call_type": "aspeech"}) + == "speech-output" + ) + + transcription_resp = TranscriptionResponse(text="hello world") + assert ( + logger.get_output_str_from_response( + transcription_resp, {"call_type": "atranscription"} + ) + == "hello world" + ) + + realtime_output = [{"type": "response", "text": "hi"}] + assert ( + logger.get_output_str_from_response( + realtime_output, + {"call_type": "_arealtime", "input": {"session": "abc"}}, + ) + == '[{"type": "response", "text": "hi"}]' + ) + + pass_through_output = {"response": "passthrough-body", "status": 200} + assert ( + logger.get_output_str_from_response( + pass_through_output, {"call_type": "pass_through_endpoint"} + ) + == "passthrough-body" + ) + + model_resp = ModelResponse( + choices=[Choices(message=Message(content="chat reply", role="assistant"))] + ) + assert '"chat reply"' in logger.get_output_str_from_response( + model_resp, + {"call_type": "acompletion", "messages": [{"role": "user", "content": "hi"}]}, + ) + assert logger.get_output_str_from_response("not-a-supported-type", {}) is None +def test_galileo_get_input_output_error_status_message(galileo_v2_env): + logger = GalileoObserve() + input_text, output_text, _ = logger._get_galileo_input_output_content( + kwargs={"messages": [{"role": "user", "content": "fail me"}]}, + response_obj=None, + level="ERROR", + status_message="provider timeout", + ) + assert input_text == "fail me" + assert output_text == "provider timeout" + + +def test_galileo_get_output_str_rerank_response(galileo_v2_env): + logger = GalileoObserve() + rerank_response = RerankResponse( + results=[ + {"index": 2, "relevance_score": 0.98}, + {"index": 0, "relevance_score": 0.12}, + ] + ) + output = logger.get_output_str_from_response( + rerank_response, {"call_type": "arerank"} + ) + assert output is not None + assert '"index": 2' in output + assert '"relevance_score": 0.98' in output + + +@pytest.mark.asyncio +async def test_galileo_async_log_success_rerank(galileo_v2_env): + import datetime + + logger = GalileoObserve() + rerank_response = RerankResponse(results=[{"index": 1, "relevance_score": 0.95}]) + + mock_response = MagicMock() + mock_response.is_success = True + mock_response.status_code = 201 + + with patch.object(logger.async_httpx_handler, "post", return_value=mock_response): + await logger.async_log_success_event( + kwargs={ + "call_type": "arerank", + "model": "cohere/rerank-english-v3.0", + "query": "What is the capital of the United States?", + "documents": ["doc-a", "doc-b"], + "standard_logging_object": { + "call_type": "arerank", + "model": "cohere/rerank-english-v3.0", + "messages": "What is the capital of the United States?", + "prompt_tokens": 0, + "completion_tokens": 0, + "total_tokens": 0, + "response_cost": 0.0, + "startTime": datetime.datetime( + 2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc + ).timestamp(), + "endTime": datetime.datetime( + 2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc + ).timestamp(), + }, + }, + response_obj=rerank_response, + start_time=datetime.datetime(2026, 5, 25, 12, 0, 0), + end_time=datetime.datetime(2026, 5, 25, 12, 0, 1), + ) + + assert logger.in_memory_records == [] + + def test_galileo_get_ingest_request_unconfigured(monkeypatch): monkeypatch.delenv("GALILEO_API_KEY", raising=False) monkeypatch.delenv("GALILEO_BASE_URL", raising=False) @@ -260,11 +501,27 @@ def test_galileo_get_ingest_request_legacy(monkeypatch): monkeypatch.setenv("GALILEO_PASSWORD", "pw") monkeypatch.setenv("GALILEO_BASE_URL", "https://galileo.example/") monkeypatch.setenv("GALILEO_PROJECT_ID", "proj") + monkeypatch.setenv("GALILEO_LOG_STREAM_ID", "stream-id") logger = GalileoObserve() - logger.in_memory_records = [{"foo": "bar"}] + logger.in_memory_records = [ + { + "latency_ms": 1, + "status_code": 200, + "input_text": "hi", + "output_text": "ok", + "node_type": "acompletion", + "model": "gpt", + "num_input_tokens": 1, + "num_output_tokens": 1, + "num_total_tokens": 2, + "created_at": "2026-05-25T12:00:00", + } + ] url, payload = logger._get_ingest_request() - assert url == "https://galileo.example/projects/proj/observe/ingest" - assert payload == {"records": [{"foo": "bar"}]} + assert url == "https://galileo.example/v2/projects/proj/traces" + assert "traces" in payload + assert payload["log_stream_id"] == "stream-id" + assert payload["traces"][0]["input"] == "hi" @pytest.mark.asyncio @@ -360,6 +617,48 @@ async def test_galileo_flush_resets_headers_on_401(monkeypatch): assert logger.in_memory_records == [{"records": "x"}] +@pytest.mark.asyncio +async def test_galileo_async_log_success_preserves_passthrough_messages( + galileo_v2_env, +): + import datetime + + logger = GalileoObserve() + logger.batch_size = 2 + messages = [ + {"role": "system", "content": "be helpful"}, + {"role": "user", "content": "hi"}, + ] + + await logger.async_log_success_event( + kwargs={ + "call_type": "pass_through_endpoint", + "model": "gpt", + "messages": messages, + "standard_logging_object": { + "call_type": "pass_through_endpoint", + "model": "gpt", + "prompt_tokens": 1, + "completion_tokens": 2, + "total_tokens": 0, + "response_cost": 0.001, + "startTime": datetime.datetime( + 2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc + ).timestamp(), + "endTime": datetime.datetime( + 2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc + ).timestamp(), + }, + }, + response_obj={"response": "ok"}, + start_time=datetime.datetime(2026, 5, 25, 12, 0, 0), + end_time=datetime.datetime(2026, 5, 25, 12, 0, 1), + ) + + assert logger.in_memory_records[0]["messages"] == messages + assert logger.in_memory_records[0]["num_total_tokens"] == 3 + + @pytest.mark.asyncio async def test_galileo_async_log_success_appends_and_flushes(galileo_v2_env): import datetime @@ -387,11 +686,26 @@ async def test_galileo_async_log_success_appends_and_flushes(galileo_v2_env): "call_type": "acompletion", "model": "gpt", "messages": [{"role": "user", "content": "hi"}], + "standard_logging_object": { + "call_type": "acompletion", + "model": "gpt", + "messages": [{"role": "user", "content": "hi"}], + "prompt_tokens": 1, + "completion_tokens": 2, + "total_tokens": 3, + "response_cost": 0.001, + "startTime": datetime.datetime( + 2026, 5, 25, 12, 0, 0, tzinfo=datetime.timezone.utc + ).timestamp(), + "endTime": datetime.datetime( + 2026, 5, 25, 12, 0, 1, tzinfo=datetime.timezone.utc + ).timestamp(), + }, }, response_obj=response, start_time=datetime.datetime(2026, 5, 25, 12, 0, 0), end_time=datetime.datetime(2026, 5, 25, 12, 0, 1), ) - assert "/v2/projects/" in flushed_url["url"] + assert "/ingest/traces/" in flushed_url["url"] assert logger.in_memory_records == [] diff --git a/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py index 1befa9a72b..dfc9f0361c 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_callback_management_endpoints.py @@ -257,3 +257,12 @@ class TestCallbackManagementEndpoints: assert ( has_detailed_params ), "Expected at least one callback to have detailed parameter configuration" + + galileo_config = next( + (config for config in response_data if config.get("id") == "galileo"), + None, + ) + assert galileo_config is not None + assert galileo_config["displayName"] == "Galileo" + assert "GALILEO_API_KEY" in galileo_config["dynamic_params"] + assert "GALILEO_PROJECT_ID" in galileo_config["dynamic_params"] diff --git a/ui/litellm-dashboard/public/assets/logos/galileo.ico b/ui/litellm-dashboard/public/assets/logos/galileo.ico new file mode 100644 index 0000000000..c50b9de4df Binary files /dev/null and b/ui/litellm-dashboard/public/assets/logos/galileo.ico differ diff --git a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx index 187c05a468..0334c55f66 100644 --- a/ui/litellm-dashboard/src/components/callback_info_helpers.tsx +++ b/ui/litellm-dashboard/src/components/callback_info_helpers.tsx @@ -43,6 +43,21 @@ export const CALLBACK_CONFIGS: CallbackConfig[] = [ }, description: "Custom Callback API Logging Integration", }, + { + id: "galileo", + displayName: "Galileo", + logo: `${asset_logos_folder}galileo.ico`, + supports_key_team_logging: false, + dynamic_params: { + GALILEO_API_KEY: "password", + GALILEO_PROJECT_ID: "text", + GALILEO_LOG_STREAM_ID: "text", + GALILEO_BASE_URL: "text", + GALILEO_USERNAME: "text", + GALILEO_PASSWORD: "password", + }, + description: "Galileo AI Observability Integration", + }, { id: "datadog", displayName: "Datadog",