214 lines
9.3 KiB
Python
214 lines
9.3 KiB
Python
"""
|
|
Mock LLM Client for offline lab execution.
|
|
|
|
All 13 labs use this instead of real API calls.
|
|
Provides deterministic responses so students can run labs without API keys.
|
|
|
|
Usage:
|
|
from mock_llm import MockAnthropic, MockResponse
|
|
|
|
client = MockAnthropic()
|
|
response = client.messages.create(
|
|
model="claude-sonnet-4",
|
|
max_tokens=1024,
|
|
system="You are a helpful assistant.",
|
|
messages=[{"role": "user", "content": "Hello"}],
|
|
tools=[...]
|
|
)
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
|
|
class MockContentBlock:
|
|
"""Simulates Anthropic's content block responses."""
|
|
|
|
def __init__(self, type: str = "text", **kwargs):
|
|
self.type = type
|
|
for k, v in kwargs.items():
|
|
setattr(self, k, v)
|
|
|
|
|
|
class MockResponse:
|
|
"""Simulates Anthropic's API response."""
|
|
|
|
def __init__(self, stop_reason: str = "end_turn", content: list = None):
|
|
self.stop_reason = stop_reason
|
|
self.content = content or [MockContentBlock(type="text", text="Mock response")]
|
|
|
|
def __getitem__(self, key):
|
|
return self.content[key]
|
|
|
|
|
|
_mock_call_count = 0
|
|
_MAX_MOCK_CALLS = 20
|
|
|
|
class MockAnthropic:
|
|
"""Simulates the Anthropic client for offline lab execution."""
|
|
|
|
def __init__(self, api_key: str = "mock-key"):
|
|
self.api_key = api_key
|
|
|
|
class messages:
|
|
"""Simulates Anthropic's messages API."""
|
|
|
|
@staticmethod
|
|
def create(model: str = None, max_tokens: int = None, system: str = None,
|
|
messages: list = None, tools: list = None, **kwargs) -> MockResponse:
|
|
"""Deterministic mock response based on prompt patterns."""
|
|
global _mock_call_count
|
|
_mock_call_count += 1
|
|
if _mock_call_count >= _MAX_MOCK_CALLS:
|
|
return MockResponse(stop_reason="end_turn", content=[
|
|
MockContentBlock(type="text", text="Mock completed analysis.")
|
|
])
|
|
|
|
if not messages:
|
|
return MockResponse(stop_reason="end_turn", content=[
|
|
MockContentBlock(type="text", text="No input provided.")
|
|
])
|
|
|
|
last_msg = messages[-1]["content"] if messages else ""
|
|
|
|
# Only trigger tool_use on the first user message (the actual request)
|
|
# After a tool result comes back, always return end_turn
|
|
user_msg_count = sum(1 for m in messages if m.get("role") == "user")
|
|
if user_msg_count > 1:
|
|
return MockResponse(stop_reason="end_turn", content=[
|
|
MockContentBlock(type="text", text=_generate_text_response(last_msg))
|
|
])
|
|
|
|
# Tool calling patterns — return tool_use if prompt asks for it
|
|
if tools and _wants_tool_call(last_msg):
|
|
tool_name = tools[0]["name"]
|
|
tool_input = _generate_tool_input(tool_name, last_msg)
|
|
return MockResponse(stop_reason="tool_use", content=[
|
|
MockContentBlock(type="tool_use", name=tool_name, input=tool_input)
|
|
])
|
|
|
|
# Default text response
|
|
return MockResponse(stop_reason="end_turn", content=[
|
|
MockContentBlock(type="text", text=_generate_text_response(last_msg))
|
|
])
|
|
|
|
|
|
def _wants_tool_call(prompt: str) -> bool:
|
|
"""Detect if the prompt asks for tool use.
|
|
|
|
Only matches multi-word phrases to avoid false positives
|
|
from common English words in file contents or agent responses.
|
|
"""
|
|
tool_triggers = [
|
|
"read file", "read the file", "search for", "search the",
|
|
"write file", "write to", "execute", "run this",
|
|
"find files", "list files", "check if", "verify that",
|
|
"look up", "lookup", "query the", "fetch from",
|
|
]
|
|
prompt_lower = prompt.lower()
|
|
return any(t in prompt_lower for t in tool_triggers)
|
|
|
|
|
|
def _generate_tool_input(tool_name: str, prompt: str) -> dict:
|
|
"""Generate realistic mock tool input based on prompt + tool name."""
|
|
if tool_name == "read_file":
|
|
# Extract filename from prompt
|
|
files = re.findall(r'[\w/.-]+\.\w+', prompt)
|
|
return {"path": files[0] if files else "test.txt", "reasoning": "Reading file as requested"}
|
|
elif tool_name == "search_web":
|
|
return {"query": prompt[:50], "reasoning": "Searching for information"}
|
|
elif tool_name == "write_file":
|
|
return {"path": "output.txt", "content": "Mock content", "reasoning": "Writing output"}
|
|
elif "query" in tool_name.lower():
|
|
return {"query": "SELECT * FROM users LIMIT 5", "reasoning": "Querying database"}
|
|
else:
|
|
return {"reasoning": "Executing tool", "action": prompt[:50]}
|
|
|
|
|
|
def _generate_text_response(prompt: str) -> str:
|
|
"""Generate realistic mock text response from tool results or user input."""
|
|
if not prompt or len(prompt) < 5:
|
|
return "I understand your request. How can I help you further?"
|
|
|
|
prompt_lower = prompt.lower()
|
|
|
|
# Agent session log responses
|
|
if "session" in prompt_lower and ("agent" in prompt_lower or "tool call" in prompt_lower):
|
|
return "I've analyzed the session log. Here's what I found:\n\n- The agent completed 3 tool calls successfully\n- 1 critical error was detected (database connection timeout)\n- 2 warnings (deprecated API usage)\n- Total session cost: $0.042\n- Recommendation: Check the database connection pool settings"
|
|
|
|
# File content responses - extract what the file is about
|
|
if "contents of" in prompt_lower:
|
|
if "error" in prompt_lower or "FAIL" in prompt_lower:
|
|
return "I've analyzed the file. It contains session trace data from a production agent deployment. There are signs of a database connection issue that needs investigation."
|
|
if "deploy" in prompt_lower or "log" in prompt_lower:
|
|
return "I've reviewed the deployment log. The system ran 3 tool calls successfully with one critical error. The root cause appears to be a database timeout."
|
|
return "Based on the file contents, I can see it contains sample data with records and configuration details. The key patterns I identified are structured around standard formats."
|
|
|
|
# Search result responses
|
|
if "search" in prompt_lower and ("result" in prompt_lower or "found" in prompt_lower):
|
|
return "Here are the search results I found. The most relevant result covers the topic you asked about with specific examples and implementation details."
|
|
|
|
# Write confirmation responses
|
|
if "successfully wrote" in prompt_lower or "wrote" in prompt_lower:
|
|
return "The file has been written successfully. I've confirmed the content was saved correctly."
|
|
|
|
# Hello/greetings
|
|
if "hello" in prompt_lower or "hi " in prompt_lower:
|
|
return "Hello! I'm your AI coding agent. How can I help you today?"
|
|
|
|
# Explanation patterns
|
|
elif "explain" in prompt_lower or "what is" in prompt_lower:
|
|
return "Based on my analysis: this is a well-known pattern in software engineering. The key insight is that it separates concerns and allows for independent evolution of components."
|
|
|
|
# Error/fix patterns
|
|
elif "error" in prompt_lower or "bug" in prompt_lower or "fix" in prompt_lower:
|
|
return "I found the issue. The problem is a missing null check on line 42. Adding `if value is not None:` before the operation resolves it."
|
|
|
|
# Test results
|
|
elif "test" in prompt_lower or "pytest" in prompt_lower:
|
|
return "All tests pass. 34 passed, 0 failed, 0 skipped. Completed in 3.62 seconds."
|
|
|
|
# Summary requests
|
|
elif "summarize" in prompt_lower or "summary" in prompt_lower:
|
|
return "Summary: The document covers three main topics. First, architectural patterns for agent systems. Second, security considerations for production deployments. Third, evaluation methodologies."
|
|
|
|
else:
|
|
return "I've completed the analysis. Here are the key findings:\n\n1. The approach is viable\n2. Recommended next steps are documented\n3. No blockers identified\n\nWould you like me to proceed with implementation?"
|
|
|
|
|
|
class MockOpenAI:
|
|
"""Simulates OpenAI client for offline lab execution."""
|
|
|
|
def __init__(self, api_key: str = "mock-key"):
|
|
self.api_key = api_key
|
|
|
|
class chat:
|
|
class completions:
|
|
@staticmethod
|
|
def create(model: str = None, messages: list = None, tools: list = None, **kwargs) -> dict:
|
|
return {
|
|
"choices": [{
|
|
"message": {
|
|
"content": _generate_text_response(messages[-1]["content"] if messages else ""),
|
|
"tool_calls": None
|
|
},
|
|
"finish_reason": "stop"
|
|
}]
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Quick test
|
|
client = MockAnthropic()
|
|
resp = client.messages.create(
|
|
messages=[{"role": "user", "content": "Read the file data.txt and summarize it"}],
|
|
tools=[{"name": "read_file", "description": "Read a file"}]
|
|
)
|
|
print(f"Stop reason: {resp.stop_reason}")
|
|
for block in resp.content:
|
|
if block.type == "text":
|
|
print(f"Text: {block.text[:100]}")
|
|
elif block.type == "tool_use":
|
|
print(f"Tool: {block.name}({block.input})")
|