Dustin N. Thank you for the explanation. Could you please share the right way to get the current span id given my usage code? I tried accessing it as suggested above by RunLLM but it returned zeros.
async def _make_request(
self, request_data: Dict, thread_id: str, uid: str
) -> ModelResponse | CustomStreamWrapper:
try:
with using_attributes(
session_id=thread_id,
user_id=uid,
):
# Get the current span and its context
current_span = trace.get_current_span()
span_context = current_span.get_span_context()
trace_id = format(span_context.trace_id, '032x')
span_id = format(span_context.span_id, '016x')
# Log or store the trace ID and span ID as needed
logger.debug(f"Trace ID: {trace_id}, Span ID: {span_id}")
# Create task with timeout
stream_task = asyncio.create_task(litellm.acompletion(**request_data))
# Wait for the task with timeout
stream = await asyncio.wait_for(
stream_task, timeout=self.TIMEOUT_SECONDS
)
return streamIs there a simple way to track root span id for a call if I am using LiteLLM? I am trying to integrate feedback from the UI. My code for reference -
# app/services/chat/chat_client.py
import logging
from typing import Dict, Optional
import asyncio
from tenacity import retry, stop_after_attempt, wait_fixed, RetryError
import litellm
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.types.utils import ModelResponse
from openinference.instrumentation import using_attributes
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
class ChatClientError(Exception):
"""Custom exception for ChatClient errors"""
pass
class ChatRetriesExceededError(Exception):
"""Custom exception for ChatClient retries exceeded errors"""
pass
class ChatClient:
_instance = None
# Configuration constants
TIMEOUT_SECONDS = 6 # Reduced timeout for each attempt (6 seconds)
MAX_RETRIES = 3 # Number of retries (3 total attempts)
def __new__(cls):
if cls._instance is None:
cls._instance = super(ChatClient, cls).__new__(cls)
return cls._instance
@retry(
stop=stop_after_attempt(MAX_RETRIES),
wait=wait_fixed(0),
reraise=True,
)
async def _make_request(
self, request_data: Dict, thread_id: str, uid: str
) -> ModelResponse | CustomStreamWrapper:
try:
with using_attributes(
session_id=thread_id,
user_id=uid,
):
# Create task with timeout
stream_task = asyncio.create_task(litellm.acompletion(**request_data))
# Wait for the task with timeout
stream = await asyncio.wait_for(
stream_task, timeout=self.TIMEOUT_SECONDS
)
return stream
except asyncio.TimeoutError:
logger.error(f"Request timed out after {self.TIMEOUT_SECONDS} seconds")
raise ChatClientError("Request timed out")
except Exception as e:
logger.error(f"Error making request: {str(e)}")
raise ChatClientError(f"LLM request failed: {str(e)}")
@staticmethod
async def get_response_stream(
uid: str,
thread_id: str,
model: str,
messages: list,
tools: Optional[list] = None,
parallel_tool_calls: Optional[bool] = False,
response_format: Optional[Dict[str, str]] = {"type": "text"},
) -> ModelResponse | CustomStreamWrapper:
litellm.drop_params = True
litellm._logging._disable_debugging()
# Prepare the request data
request_data = {
"model": model,
"messages": messages,
"stream": True,
}
if tools:
request_data["tools"] = tools
request_data["tool_choice"] = "auto"
request_data["parallel_tool_calls"] = parallel_tool_calls
try:
# Try to make the request with retries
return await ChatClient()._make_request(request_data, thread_id, uid)
except RetryError:
logger.error("All retry attempts failed")
raise ChatRetriesExceededError("All retry attempts failed")
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
raise ChatClientError(f"Unexpected error: {str(e)}")
chat_client = ChatClient()Added a pull request addressing this issue https://github.com/Arize-ai/openinference/pull/1246, waiting for a review
Hello, team! Thanks for building this package. I am trying to send LiteLLM acompletion calls to a self hosted Arize Phoenix Collector. I am able to receive the traces of the calls to acompletion but in the traces, I do not see the streaming output being captured for these calls (attaching screenshot). According to the issue #1188, the support for streaming responses was added in openinference.instrumentation.litellm 0.1.6. I am not sure what I am doing wrong, I would really appreciate some help with this! My requirements.txt looks like this
dependencies = [
"uvicorn[standard]==0.34.0",
"fastapi[standard]==0.115.6",
"litellm==1.59.6",
"arize-phoenix-otel==0.7.1",
"openinference.instrumentation.litellm==0.1.6",
"opentelemetry-instrumentation-fastapi==0.50b0"
]Here is how I am setting it up in my main.py
...
app = create_app()
tracer_provider = register(
project_name="braince-ai",
endpoint="http://phoenix.phoenix.svc.cluster.local:4317", # Sends traces using gRPC
)
LiteLLMInstrumentor().instrument(tracer_provider=tracer_provider)
...
My chat client that calls litellm acompletion looks like this
import logging
from typing import Dict, Optional
import litellm
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.types.utils import ModelResponse
logger = logging.getLogger(__name__)
class ChatClient:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(ChatClient, cls).__new__(cls)
return cls._instance
@staticmethod
async def get_response_stream(
model: str,
messages: list,
tools: Optional[list] = None,
parallel_tool_calls: Optional[bool] = False,
response_format: Optional[Dict[str, str]] = {"type": "text"}
) -> ModelResponse | CustomStreamWrapper:
litellm.drop_params=True
litellm._logging._disable_debugging()
#litellm.set_verbose=True
# Prepare the request data
request_data = {
"model": model,
"messages": messages,
"stream": True,
}
# logger.debug(f"Request data: {request_data}")
# Add tools if provided
if tools:
request_data["tools"] = tools
request_data["tool_choice"] = "auto"
request_data["parallel_tool_calls"] = parallel_tool_calls
# Create the stream
stream = await litellm.acompletion(**request_data)
return stream
chat_client = ChatClient()