InteractiveRefining AI Agents Using Observability: Continuous Improvement Through Log Analysis

1## Example: Analyzing a week's worth of logs
2import json
3from collections import Counter
4
5def analyze_logs(log_file):
6    """Analyze patterns in agent logs."""
7    tool_calls = []
8    unknown_responses = []
9    reasoning_lengths = []
10    
11    with open(log_file, 'r') as f:
12        for line in f:
13            log = json.loads(line)
14            
15            # Track tool usage
16            if 'tool_call' in log:
17                tool_calls.append(log['tool_call']['name'])
18            
19            # Track "I don't know" responses
20            if "I don't know" in log.get('response', ''):
21                unknown_responses.append(log['query'])
22            
23            # Track reasoning chain length
24            if 'reasoning_steps' in log:
25                reasoning_lengths.append(len(log['reasoning_steps']))
26    
27    return {
28        'tool_usage': Counter(tool_calls),
29        'unknown_queries': unknown_responses,
30        'avg_reasoning_length': sum(reasoning_lengths) / len(reasoning_lengths) if reasoning_lengths else 0
31    }
32
33## Analyze the logs
34patterns = analyze_logs('assistant_logs.jsonl')
35
36print("Tool Usage:")
37for tool, count in patterns['tool_usage'].most_common():
38    print(f"  {tool}: {count} times")
39
40print(f"\nAverage reasoning chain length: {patterns['avg_reasoning_length']:.1f} steps")
41
42print(f"\nQueries resulting in 'I don't know': {len(patterns['unknown_queries'])}")
43for query in patterns['unknown_queries'][:5]:  # Show first 5
44    print(f"  - {query}")

1## Using Claude Sonnet 4.5 for its superior tool-use capabilities
2from anthropic import Anthropic
3
4client = Anthropic(api_key="your-api-key")
5
6## Define a news search tool
7tools = [
8    {
9        "name": "search_news",
10        "description": "Search for recent news articles about a topic. Use this when the user asks about current events or recent developments.",
11        "input_schema": {
12            "type": "object",
13            "properties": {
14                "query": {
15                    "type": "string",
16                    "description": "The search query for news articles"
17                }
18            },
19            "required": ["query"]
20        }
21    }
22]
23
24def search_news(query):
25    """Simulate a news search (in practice, call a real news API)."""
26    # This would call a real news API
27    return f"Recent articles about {query}: [Article 1], [Article 2], [Article 3]"
28
29## Now when the agent encounters a question about recent events,
30## it has a tool to help instead of saying "I don't know"
31response = client.messages.create(
32    model="claude-sonnet-4.5",
33    max_tokens=1024,
34    tools=tools,
35    messages=[
36        {"role": "user", "content": "What happened in the tech industry this week?"}
37    ]
38)
39
40print(response.content)

1## Improved system prompt that emphasizes checking memory first
2system_prompt = """You are a helpful personal assistant.
3
4Before using any tools, always check if you already have the information you need:
51. Review the conversation history
62. Check if the user has already provided this information
73. Only use tools if you genuinely need new information
8
9This saves time and provides faster responses."""
10
11## The agent will now be more thoughtful about when to use tools

1system_prompt = """You are a helpful personal assistant.
2
3When reasoning through problems, be concise and direct:
4
5Good example:
6User asks: "What's 15% of 80?"
7Reasoning: Need to calculate 15% of 80. That's $0.15 \times 80 = 12$.
8Answer: 12
9
10Avoid overthinking simple problems. Break down complex problems, but keep each step clear and necessary."""

1def calculate_retry_rate(logs):
2    """Calculate how often tool calls need to be retried."""
3    total_tool_calls = 0
4    retried_calls = 0
5    
6    for log in logs:
7        if 'tool_call' in log:
8            total_tool_calls += 1
9            if log.get('retry_count', 0) > 0:
10                retried_calls += 1
11    
12    return retried_calls / total_tool_calls if total_tool_calls > 0 else 0
13
14retry_rate = calculate_retry_rate(logs)
15print(f"Tool retry rate: {retry_rate:.1%}")

1def calculate_avg_reasoning_length(logs):
2    """Calculate average number of reasoning steps."""
3    lengths = [len(log.get('reasoning_steps', [])) for log in logs if 'reasoning_steps' in log]
4    return sum(lengths) / len(lengths) if lengths else 0
5
6avg_length = calculate_avg_reasoning_length(logs)
7print(f"Average reasoning chain length: {avg_length:.1f} steps")

1def analyze_response_times(logs):
2    """Break down response times by query type."""
3    from collections import defaultdict
4    
5    times_by_type = defaultdict(list)
6    
7    for log in logs:
8        query_type = log.get('query_type', 'unknown')
9        response_time = log.get('response_time_ms', 0)
10        times_by_type[query_type].append(response_time)
11    
12    # Calculate averages
13    avg_times = {
14        qtype: sum(times) / len(times)
15        for qtype, times in times_by_type.items()
16    }
17    
18    return avg_times
19
20response_times = analyze_response_times(logs)
21for query_type, avg_time in sorted(response_times.items(), key=lambda x: x[1], reverse=True):
22    print(f"{query_type}: {avg_time:.0f}ms")

1## Week 1: Initial observation
2## Logs show: Agent uses web search 40 times, but 15 of those searches
3## are for information already in the conversation history
4
5## Week 2: Make a change
6## Update the system prompt to emphasize checking memory first
7new_prompt = """Before searching the web, always check:
81. Has the user already mentioned this in our conversation?
92. Did I already look this up earlier in this session?
10
11Only search if you genuinely need new information."""
12
13## Week 3: Measure impact
14## New logs show: Agent uses web search 28 times, with only 2 redundant searches
15## Success! Redundant searches dropped from 37.5% to 7.1%
16
17## Week 4: Move to next improvement
18## Now focus on the next pattern: reasoning chains are too long for simple questions

1def check_performance_goal(logs, max_response_time_ms=5000):
2    """Check if agent meets response time goal."""
3    response_times = [log.get('response_time_ms', 0) for log in logs]
4    
5    within_goal = sum(1 for t in response_times if t <= max_response_time_ms)
6    total = len(response_times)
7    
8    success_rate = within_goal / total if total > 0 else 0
9    
10    print(f"Response time goal: {success_rate:.1%} of responses under {max_response_time_ms}ms")
11    print(f"Average response time: {sum(response_times) / total:.0f}ms")
12    
13    return success_rate
14
15success_rate = check_performance_goal(logs)
16
17if success_rate < 0.90:  # Goal: 90% of responses under 5 seconds
18    print("Need to optimize response time!")

1def check_for_anomalies(logs, window_size=100):
2    """Simple anomaly detection for production logs."""
3    recent_logs = logs[-window_size:]
4    
5    # Check error rate
6    errors = sum(1 for log in recent_logs if log.get('error'))
7    error_rate = errors / len(recent_logs)
8    
9    if error_rate > 0.05:  # More than 5% errors
10        print(f"⚠️  Alert: Error rate is {error_rate:.1%} (threshold: 5%)")
11    
12    # Check response times
13    response_times = [log.get('response_time_ms', 0) for log in recent_logs]
14    avg_time = sum(response_times) / len(response_times)
15    
16    if avg_time > 8000:  # Slower than 8 seconds
17        print(f"⚠️  Alert: Average response time is {avg_time:.0f}ms (threshold: 8000ms)")

1import re
2from datetime import datetime
3
4def redact_sensitive_info(text):
5    """Remove sensitive information from logs."""
6    # Redact email addresses
7    text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]', text)
8    
9    # Redact phone numbers
10    text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE]', text)
11    
12    # Redact credit card numbers
13    text = re.sub(r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b', '[CARD]', text)
14    
15    return text
16
17## Use this when logging user queries
18log_entry = {
19    'query': redact_sensitive_info(user_query),
20    'response': redact_sensitive_info(agent_response),
21    'timestamp': datetime.now().isoformat()
22}

1def print_weekly_summary(logs):
2    """Print a summary of the week's agent performance."""
3    print("=== Weekly Agent Performance Summary ===\n")
4    
5    print(f"Total queries: {len(logs)}")
6    
7    # Tool usage
8    tool_calls = [log.get('tool_call', {}).get('name') for log in logs if 'tool_call' in log]
9    print(f"Tool calls: {len(tool_calls)}")
10    
11    # Success rate
12    errors = sum(1 for log in logs if log.get('error'))
13    success_rate = (len(logs) - errors) / len(logs) if logs else 0
14    print(f"Success rate: {success_rate:.1%}")
15    
16    # Average response time
17    times = [log.get('response_time_ms', 0) for log in logs]
18    avg_time = sum(times) / len(times) if times else 0
19    print(f"Avg response time: {avg_time:.0f}ms")
20    
21    print("\n" + "="*40)
22
23## Run this every Monday morning
24print_weekly_summary(last_week_logs)