​
configure_langfuse

def configure_langfuse(
    public_key: Optional[str] = None,
    secret_key: Optional[str] = None,
    host: Optional[str] = None,
    debug: Optional[bool] = None,
    enabled: Optional[bool] = None
):
Configure Langfuse for CAMEL models. Parameters:
  • public_key (Optional[str]): Langfuse public key. Can be set via LANGFUSE_PUBLIC_KEY. (default: :obj:None)
  • secret_key (Optional[str]): Langfuse secret key. Can be set via LANGFUSE_SECRET_KEY. (default: :obj:None)
  • host (Optional[str]): Langfuse host URL. Can be set via LANGFUSE_HOST. (default: :obj:https://cloud.langfuse.com)
  • debug (Optional[bool]): Enable debug mode. Can be set via LANGFUSE_DEBUG. (default: :obj:None)
  • enabled (Optional[bool]): Enable/disable tracing. Can be set via LANGFUSE_ENABLED. (default: :obj:None)
Note: This function configures the native langfuse_context which works with @observe() decorators. Set enabled=False to disable all tracing.

​
is_langfuse_available

def is_langfuse_available():
Check if Langfuse is configured.

​
set_current_agent_session_id

def set_current_agent_session_id(session_id: str):
Set the session ID for the current agent in thread-local storage. Parameters:
  • session_id (str): The session ID to set for the current agent.

​
get_current_agent_session_id

def get_current_agent_session_id():
Returns: Optional[str]: The session ID for the current agent.

​
update_langfuse_trace

def update_langfuse_trace(
    session_id: Optional[str] = None,
    user_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    tags: Optional[List[str]] = None
):
Update the current Langfuse trace with session ID and metadata. Parameters:
  • session_id (Optional[str]): Optional session ID to use. If :obj:None uses the current agent’s session ID. (default: :obj:None)
  • user_id (Optional[str]): Optional user ID for the trace. (default: :obj:None)
  • metadata (Optional[Dict[str, Any]]): Optional metadata dictionary. (default: :obj:None)
  • tags (Optional[List[str]]): Optional list of tags. (default: :obj:None)
Returns: bool: True if update was successful, False otherwise.

​
update_current_observation

def update_current_observation(
    input: Optional[Dict[str, Any]] = None,
    output: Optional[Dict[str, Any]] = None,
    model: Optional[str] = None,
    model_parameters: Optional[Dict[str, Any]] = None,
    usage_details: Optional[Dict[str, Any]] = None,
    **kwargs
):
Update the current Langfuse observation with input, output, model, model_parameters, and usage_details. Parameters:
  • input (Optional[Dict[str, Any]]): Optional input dictionary. (default: :obj:None)
  • output (Optional[Dict[str, Any]]): Optional output dictionary. (default: :obj:None)
  • model (Optional[str]): Optional model name. (default: :obj:None)
  • model_parameters (Optional[Dict[str, Any]]): Optional model parameters dictionary. (default: :obj:None)
  • usage_details (Optional[Dict[str, Any]]): Optional usage details dictionary. (default: :obj:None)
Returns: None

​
get_langfuse_status

def get_langfuse_status():
Returns: Dict[str, Any]: Status information including configuration state.

​
observe

def observe(*args, **kwargs):