Refining AI Agents Using Observability

## Example: Analyzing a week's worth of logs
import json
from collections import Counter


def analyze_logs(log_file):
    """Analyze patterns in agent logs."""
    tool_calls = []
    unknown_responses = []
    reasoning_lengths = []

    with open(log_file, "r") as f:
        for line in f:
            log = json.loads(line)

            # Track tool usage
            if "tool_call" in log:
                tool_calls.append(log["tool_call"]["name"])

            # Track "I don't know" responses
            if "I don't know" in log.get("response", ""):
                unknown_responses.append(log["query"])

            # Track reasoning chain length
            if "reasoning_steps" in log:
                reasoning_lengths.append(len(log["reasoning_steps"]))

    return {
        "tool_usage": Counter(tool_calls),
        "unknown_queries": unknown_responses,
        "avg_reasoning_length": sum(reasoning_lengths) / len(reasoning_lengths)
        if reasoning_lengths
        else 0,
    }


## Analyze the logs
patterns = analyze_logs("assistant_logs.jsonl")

print("Tool Usage:")
for tool, count in patterns["tool_usage"].most_common():
    print(f"  {tool}: {count} times")

print(
    f"\nAverage reasoning chain length: {patterns['avg_reasoning_length']:.1f} steps"
)

print(
    f"\nQueries resulting in 'I don't know': {len(patterns['unknown_queries'])}"
)
for query in patterns["unknown_queries"][:5]:  # Show first 5
    print(f"  - {query}")

## Using Claude Sonnet 4.5 for its superior tool-use capabilities
from anthropic import Anthropic

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

## Define a news search tool
tools = [
    {
        "name": "search_news",
        "description": "Search for recent news articles about a topic. Use this when the user asks about current events or recent developments.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query for news articles",
                }
            },
            "required": ["query"],
        },
    }
]


def search_news(query):
    """Simulate a news search (in practice, call a real news API)."""
    # This would call a real news API
    return (
        f"Recent articles about {query}: [Article 1], [Article 2], [Article 3]"
    )


## Now when the agent encounters a question about recent events,
## it has a tool to help instead of saying "I don't know"
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    tools=tools,
    messages=[
        {
            "role": "user",
            "content": "What happened in the tech industry this week?",
        }
    ],
)

print(response.content)

## Improved system prompt that emphasizes checking memory first
system_prompt = """You are a helpful personal assistant.

Before using any tools, always check if you already have the information you need:

1. Review the conversation history
2. Check if the user has already provided this information
3. Only use tools if you genuinely need new information

This saves time and provides faster responses."""

## The agent will now be more thoughtful about when to use tools

system_prompt = """You are a helpful personal assistant.

When reasoning through problems, be concise and direct:

Good example:
User asks: "What's 15% of 80?"
Reasoning: Need to calculate 15% of 80. That's $0.15 \times 80 = 12$.
Answer: 12

Avoid overthinking simple problems. Break down complex problems, but keep each step clear and necessary."""

def calculate_retry_rate(logs):
    """Calculate how often tool calls need to be retried."""
    total_tool_calls = 0
    retried_calls = 0

    for log in logs:
        if "tool_call" in log:
            total_tool_calls += 1
            if log.get("retry_count", 0) > 0:
                retried_calls += 1

    return retried_calls / total_tool_calls if total_tool_calls > 0 else 0


retry_rate = calculate_retry_rate(logs)
print(f"Tool retry rate: {retry_rate:.1%}")

def calculate_avg_reasoning_length(logs):
    """Calculate average number of reasoning steps."""
    lengths = [
        len(log.get("reasoning_steps", []))
        for log in logs
        if "reasoning_steps" in log
    ]
    return sum(lengths) / len(lengths) if lengths else 0


avg_length = calculate_avg_reasoning_length(logs)
print(f"Average reasoning chain length: {avg_length:.1f} steps")

def analyze_response_times(logs):
    """Break down response times by query type."""
    from collections import defaultdict

    times_by_type = defaultdict(list)

    for log in logs:
        query_type = log.get("query_type", "unknown")
        response_time = log.get("response_time_ms", 0)
        times_by_type[query_type].append(response_time)

    # Calculate averages
    avg_times = {
        qtype: sum(times) / len(times) for qtype, times in times_by_type.items()
    }

    return avg_times


response_times = analyze_response_times(logs)
for query_type, avg_time in sorted(
    response_times.items(), key=lambda x: x[1], reverse=True
):
    print(f"{query_type}: {avg_time:.0f}ms")

## Week 1: Initial observation
## Logs show: Agent uses web search 40 times, but 15 of those searches
## are for information already in the conversation history

## Week 2: Make a change
## Update the system prompt to emphasize checking memory first
new_prompt = """Before searching the web, always check:

1. Has the user already mentioned this in our conversation?
2. Did I already look this up earlier in this session?

Only search if you genuinely need new information."""

## Week 3: Measure impact
## New logs show: Agent uses web search 28 times, with only 2 redundant searches
## Success! Redundant searches dropped from 37.5% to 7.1%

## Week 4: Move to next improvement
## Now focus on the next pattern: reasoning chains are too long for simple questions

def check_performance_goal(logs, max_response_time_ms=5000):
    """Check if agent meets response time goal."""
    response_times = [log.get("response_time_ms", 0) for log in logs]

    within_goal = sum(1 for t in response_times if t <= max_response_time_ms)
    total = len(response_times)

    success_rate = within_goal / total if total > 0 else 0

    print(
        f"Response time goal: {success_rate:.1%} of responses under {max_response_time_ms}ms"
    )
    print(f"Average response time: {sum(response_times) / total:.0f}ms")

    return success_rate


success_rate = check_performance_goal(logs)

if success_rate < 0.90:  # Goal: 90% of responses under 5 seconds
    print("Need to optimize response time!")

def check_for_anomalies(logs, window_size=100):
    """Simple anomaly detection for production logs."""
    recent_logs = logs[-window_size:]

    # Check error rate
    errors = sum(1 for log in recent_logs if log.get("error"))
    error_rate = errors / len(recent_logs)

    if error_rate > 0.05:  # More than 5% errors
        print(f"⚠️  Alert: Error rate is {error_rate:.1%} (threshold: 5%)")

    # Check response times
    response_times = [log.get("response_time_ms", 0) for log in recent_logs]
    avg_time = sum(response_times) / len(response_times)

    if avg_time > 8000:  # Slower than 8 seconds
        print(
            f"⚠️  Alert: Average response time is {avg_time:.0f}ms (threshold: 8000ms)"
        )

import re
from datetime import datetime


def redact_sensitive_info(text):
    """Remove sensitive information from logs."""
    # Redact email addresses
    text = re.sub(
        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[EMAIL]", text
    )

    # Redact phone numbers
    text = re.sub(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "[PHONE]", text)

    # Redact credit card numbers
    text = re.sub(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b", "[CARD]", text)

    return text


## Use this when logging user queries
log_entry = {
    "query": redact_sensitive_info(user_query),
    "response": redact_sensitive_info(agent_response),
    "timestamp": datetime.now().isoformat(),
}

def print_weekly_summary(logs):
    """Print a summary of the week's agent performance."""
    print("=== Weekly Agent Performance Summary ===\n")

    print(f"Total queries: {len(logs)}")

    # Tool usage
    tool_calls = [
        log.get("tool_call", {}).get("name")
        for log in logs
        if "tool_call" in log
    ]
    print(f"Tool calls: {len(tool_calls)}")

    # Success rate
    errors = sum(1 for log in logs if log.get("error"))
    success_rate = (len(logs) - errors) / len(logs) if logs else 0
    print(f"Success rate: {success_rate:.1%}")

    # Average response time
    times = [log.get("response_time_ms", 0) for log in logs]
    avg_time = sum(times) / len(times) if times else 0
    print(f"Avg response time: {avg_time:.0f}ms")

    print("\n" + "=" * 40)


## Run this every Monday morning
print_weekly_summary(last_week_logs)