Evaluating LLM Tool Calls with Laminar

import os
import json
from openai import OpenAI
from lmnr import Laminar, observe
from dotenv import load_dotenv

load_dotenv()

# Initialize Laminar for automatic tracing
Laminar.initialize()

client = OpenAI()

@observe(name="analyze_data")
def analyze_data(messages: list):
    """Main agent function that accepts input messages and handles tool calls"""
    
    # Hardcode system message at the beginning
    full_messages = [
        {
            "role": "system",
            "content": """You are a data analysis assistant. Use the available tools to help users analyze their data and generate insights. Always:
1. Query the appropriate database first
2. Create visualizations when helpful
3. Provide clear summaries of findings
4. Compare time periods when relevant

IMPORTANT: Always start by understanding what data you need, then query it, then process it."""
        }
    ]
    
    # Add input messages (skip any existing system messages)
    for msg in messages:
        if msg["role"] != "system":
            full_messages.append(msg)
    
    # Loop to handle multiple rounds of tool calls
    max_iterations = 5  # Prevent infinite loops
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        # Make LLM call
        response = client.chat.completions.create(
            model="o4-mini",
            messages=full_messages,
            tools=tools,
            tool_choice="auto"
        )
        
        # Add assistant's response to messages
        assistant_message = {
            "role": "assistant",
            "content": response.choices[0].message.content
        }
        if response.choices[0].message.tool_calls:
            assistant_message["tool_calls"] = response.choices[0].message.tool_calls
        full_messages.append(assistant_message)
        
        # If no tool calls, we have our final response
        if not response.choices[0].message.tool_calls:
            return response.choices[0].message.content
        
        # Execute tool calls
        for tool_call in response.choices[0].message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = json.loads(tool_call.function.arguments)
            
            # Simulate tool execution (replace with actual implementations)
            result = simulate_tool_execution(tool_name, tool_args)
            
            # Add tool result to messages
            full_messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": tool_name,
                "content": result
            })
    
    # If we've hit max iterations, return what we have
    return response.choices[0].message.content or "Analysis completed with multiple tool calls."

def simulate_tool_execution(tool_name, args):
    # this will mark this span as a tool span
    with Laminar.start_as_current_span(name=tool_name, span_type="TOOL"):
        """Simulate tool execution - replace with real implementations"""
        if tool_name == "query_database":
            return "Sample data: Revenue Q4 2024: $1.2M, Q3 2024: $1.0M"
        elif tool_name == "create_visualization":
            return "Chart created successfully"
        elif tool_name == "generate_summary":
            return "Key insight: 20% revenue growth quarter-over-quarter"
        elif tool_name == "compare_periods":
            return "Q4 vs Q3: +20% increase in revenue"
        elif tool_name == "detect_anomalies":
            return "No significant anomalies detected"
        
        return "Tool executed successfully"

# Example usage in production
if __name__ == "__main__":
    messages = [
        {
            "role": "user",
            "content": "How did our revenue perform last quarter compared to the previous quarter?"
        }
    ]
    result = analyze_data(messages)
    print(result)

import os
import json
from openai import OpenAI
from lmnr import Laminar, LaminarClient, observe
from dotenv import load_dotenv

load_dotenv()

# Initialize Laminar for automatic tracing
Laminar.initialize(api_key=os.environ["LMNR_PROJECT_API_KEY"])
laminar_client = LaminarClient()

client = OpenAI()

@observe("analyze_data")
def analyze_data(user_query: str):
    
    # ... rest of the code ...

    # get trace id from the current span
    trace_id = Laminar.get_trace_id()
    
    # add trace id to the response
    return {"trace_id": trace_id, "response": # ... original response ...}

# Example of how to add tag with feedback to the trace
def add_negative_feedback(trace_id: str):
    """Tag the trace with user feedback using the trace ID"""
    
    # Tag the entire trace with user feedback
    # This applies the tag to the top-level span of the trace
    laminar_client.tags.tag(trace_id, "unhelpful")

from lmnr import evaluate, LaminarDataset
from openai import OpenAI
import json
import os
from dotenv import load_dotenv

load_dotenv()

client = OpenAI()

# Same tools as production - consistency is key
tools = [
    {
        "type": "function",
        "function": {
            "name": "query_database",
            "description": "Execute SQL queries to retrieve data from the database",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SQL query to execute"},
                    "database": {"type": "string", "enum": ["analytics", "sales", "users"]}
                },
                "required": ["query", "database"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_visualization",
            "description": "Create charts and graphs from data",
            "parameters": {
                "type": "object",
                "properties": {
                    "data": {"type": "string", "description": "Data to visualize"},
                    "chart_type": {"type": "string", "enum": ["line", "bar", "pie", "scatter"]},
                    "title": {"type": "string", "description": "Chart title"}
                },
                "required": ["data", "chart_type"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "compare_periods",
            "description": "Compare metrics across different time periods",
            "parameters": {
                "type": "object",
                "properties": {
                    "metric": {"type": "string", "description": "Metric to compare"},
                    "period1": {"type": "string", "description": "First time period"},
                    "period2": {"type": "string", "description": "Second time period"}
                },
                "required": ["metric", "period1", "period2"]
            }
        }
    }
]

def data_analysis_agent(data):
    """Executor function that tests new system prompts"""
    
    # Get the original messages from the dataset
    original_messages = data["input"]
    
    # Create new messages with improved system prompt
    # This is where you test prompt improvements!
    messages = [
        {
            "role": "system", 
            "content": """You are a data analysis assistant. Use the available tools to help users analyze their data and generate insights. Always:
1. Query the appropriate database first
2. Create visualizations when helpful  
3. Provide clear summaries of findings
4. Compare time periods when relevant

IMPORTANT: Always start by understanding what data you need, then query it, then process it."""
        }
    ]
    
    # Add all non-system messages from the original conversation
    for msg in original_messages:
        if msg["role"] != "system":
            messages.append(msg)
    
    response = client.chat.completions.create(
        model="o4-mini",
        messages=messages,
        tools=tools,
        tool_choice="auto"
    )
    
    return response

def evaluate_tool_selection(output, target):
    """Evaluator to check if correct tools were selected"""
    
    # Extract actual tool calls from the response
    actual_tool_calls = []
    if hasattr(output.choices[0].message, 'tool_calls') and output.choices[0].message.tool_calls:
        actual_tool_calls = [
            {
                "name": call.function.name,
                "arguments": json.loads(call.function.arguments)
            }
            for call in output.choices[0].message.tool_calls
        ]
    
    # Extract expected tool calls from target - fix based on actual structure
    if isinstance(target, dict):
        expected_tool_calls = target.get("output", [])
        # Handle nested list structure: [[{tool_objects}]] -> [{tool_objects}]
        if expected_tool_calls and isinstance(expected_tool_calls[0], list):
            expected_tool_calls = expected_tool_calls[0]
    elif isinstance(target, list):
        expected_tool_calls = target
    else:
        # If target is something else (like a string), try to parse it
        try:
            if isinstance(target, str):
                parsed_target = json.loads(target)
                expected_tool_calls = parsed_target.get("output", []) if isinstance(parsed_target, dict) else parsed_target
            else:
                expected_tool_calls = []
        except:
            expected_tool_calls = []
    
    # Check if we called the expected tools
    expected_tool_names = [tool["name"] for tool in expected_tool_calls]
    actual_tool_names = [tool["name"] for tool in actual_tool_calls]

    # Simple binary check: did we call all expected tools?
    for expected_tool in expected_tool_names:
        if expected_tool not in actual_tool_names:
            return 0
    
    return 1

# Run the evaluation
evaluate(
    data=LaminarDataset("eval_dataset"),
    executor=data_analysis_agent,
    evaluators={
        "tool_selection": evaluate_tool_selection
    },
    project_api_key=os.environ["LMNR_PROJECT_API_KEY"],
    group_name="improved_system_prompt_v1"
)