summaryrefslogtreecommitdiff
path: root/src/interrogate.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/interrogate.py')
-rw-r--r--src/interrogate.py353
1 files changed, 0 insertions, 353 deletions
diff --git a/src/interrogate.py b/src/interrogate.py
deleted file mode 100644
index a67e25b..0000000
--- a/src/interrogate.py
+++ /dev/null
@@ -1,353 +0,0 @@
-"""
-Interrogate - replay and continue a conversation from a log file.
-Allows interactive questioning of an agent based on a previous run.
-"""
-import argparse
-import json
-import os
-from typing import Dict, List
-from agent import Agent
-from config_loader import ConfigLoader, ProviderConfig, ModelConfig
-from vfs import VFS
-
-
-# ANSI colors for terminal output
-class Colors:
- BOLD = "\033[1m"
- DIM = "\033[2m"
- CYAN = "\033[36m"
- GREEN = "\033[32m"
- YELLOW = "\033[33m"
- RED = "\033[31m"
- MAGENTA = "\033[35m"
- RESET = "\033[0m"
-
-
-def load_prompt(file_path: str) -> str:
- """Load a prompt file."""
- if not os.path.exists(file_path):
- return ""
- with open(file_path, 'r') as f:
- return f.read().strip()
-
-
-def get_provider_from_log(log_data: Dict) -> tuple:
- """Extract provider info from log and create configs."""
- model_id = log_data["model"]
- base_url = log_data.get("base_url", "")
- temperature = log_data.get("temperature", 1.0)
- extra_body = log_data.get("extra_body_config", {})
-
- # Check base_url first (takes priority — e.g. Claude via OpenRouter)
- # then fall back to model name heuristics
- model_lower = model_id.lower()
-
- if "openrouter" in (base_url or ""):
- provider_name = "openrouter"
- elif "generativelanguage" in (base_url or ""):
- provider_name = "google"
- elif "groq" in (base_url or ""):
- provider_name = "groq"
- elif "gemini" in model_lower:
- provider_name = "google"
- elif "claude" in model_lower:
- provider_name = "anthropic"
- base_url = ""
- elif "kimi" in model_lower or "moonshot" in model_lower:
- provider_name = "moonshot"
- else:
- provider_name = "openai"
-
- provider_config = ProviderConfig(
- name=provider_name,
- api_key_env=f"{provider_name.upper()}_API_KEY",
- base_url=base_url or ""
- )
-
- model_config = ModelConfig(
- id=model_id,
- provider=provider_name,
- temperature=temperature,
- extra_body=extra_body
- )
-
- return provider_config, model_config
-
-
-def sanitize_for_api(conversation: List[Dict]) -> List[Dict]:
- """
- Convert internal log entries to API-compatible messages.
- Strips custom fields (reasoning, response_metadata, turn_count, etc.)
- and converts tool_calls back to the format the API expects to see in history.
- """
- clean = []
- for msg in conversation:
- role = msg.get("role")
-
- if role == "system":
- clean.append({"role": "system", "content": msg.get("content", "")})
-
- elif role == "user":
- clean.append({"role": "user", "content": msg.get("content", "")})
-
- elif role == "tool":
- clean.append({
- "role": "tool",
- "tool_call_id": msg.get("tool_call_id", ""),
- "content": msg.get("content", "")
- })
-
- elif role == "assistant":
- entry = {"role": "assistant"}
-
- # Content
- content = msg.get("content")
- if content:
- entry["content"] = content
- else:
- entry["content"] = None
-
- # Tool calls — convert from our log format back to API format
- tool_calls = msg.get("tool_calls")
- if tool_calls:
- entry["tool_calls"] = [
- {
- "id": tc.get("id", ""),
- "type": "function",
- "function": {
- "name": tc.get("function", {}).get("name", ""),
- "arguments": tc.get("function", {}).get("arguments", "{}")
- }
- }
- for tc in tool_calls
- ]
-
- clean.append(entry)
-
- return clean
-
-
-def print_history(conversation: List[Dict], last_n: int = None):
- """Print conversation history in a readable format."""
- messages = [m for m in conversation if m.get("role") in ("user", "assistant", "tool")]
-
- if last_n:
- messages = messages[-last_n:]
-
- for msg in messages:
- role = msg.get("role", "?")
-
- if role == "user":
- content = msg.get("content", "")
- print(f"\n{Colors.GREEN}{Colors.BOLD}[USER]{Colors.RESET} {content[:300]}{'...' if len(content) > 300 else ''}")
-
- elif role == "assistant":
- reasoning = msg.get("reasoning", "")
- content = msg.get("content", "")
- tool_calls = msg.get("tool_calls")
-
- print(f"\n{Colors.CYAN}{Colors.BOLD}[ASSISTANT]{Colors.RESET}")
- if reasoning:
- preview = reasoning[:200] + ("..." if len(reasoning) > 200 else "")
- print(f" {Colors.DIM}Reasoning: {preview}{Colors.RESET}")
- if content:
- print(f" {content[:300]}{'...' if len(content) > 300 else ''}")
- if tool_calls:
- for tc in tool_calls:
- fn = tc.get("function", {})
- print(f" {Colors.YELLOW}→ {fn.get('name', '?')}({fn.get('arguments', '')[:80]}){Colors.RESET}")
-
- elif role == "tool":
- content = msg.get("content", "")
- preview = content[:150] + ("..." if len(content) > 150 else "")
- print(f" {Colors.DIM}[tool result] {preview}{Colors.RESET}")
-
-
-def print_help():
- """Print available commands."""
- print(f"""
-{Colors.BOLD}Available commands:{Colors.RESET}
- {Colors.YELLOW}history{Colors.RESET} Show full conversation history
- {Colors.YELLOW}history N{Colors.RESET} Show last N messages
- {Colors.YELLOW}reasoning{Colors.RESET} Show the last assistant reasoning trace (full)
- {Colors.YELLOW}vfs{Colors.RESET} Show current virtual filesystem state
- {Colors.YELLOW}info{Colors.RESET} Show run metadata (model, scenario, tokens)
- {Colors.YELLOW}save{Colors.RESET} Save the extended conversation
- {Colors.YELLOW}help{Colors.RESET} Show this message
- {Colors.YELLOW}exit{Colors.RESET} Quit
-
- Anything else is sent as a message to the agent.
-""")
-
-
-def main():
- parser = argparse.ArgumentParser(description="Interrogate a conversation log.")
- parser.add_argument("log_file", help="Path to the log file to load.")
- parser.add_argument("--config", default="config.yaml", help="Path to config file (for API keys)")
- parser.add_argument("--show-history", action="store_true", help="Print conversation history on load")
- args = parser.parse_args()
-
- # Load config (for API keys)
- config = ConfigLoader(args.config)
- try:
- config.load()
- except FileNotFoundError:
- pass # Config file is optional for interrogation
-
- # Load log file
- try:
- with open(args.log_file, 'r') as f:
- log_data = json.load(f)
- except FileNotFoundError:
- print(f"Error: Log file not found at {args.log_file}")
- return
- except json.JSONDecodeError:
- print(f"Error: Could not decode JSON from {args.log_file}")
- return
-
- # Extract data from log
- conversation_history = log_data.get("conversation", [])
- if not conversation_history:
- print("Error: No conversation history in log file")
- return
-
- system_prompt = conversation_history[0].get('content', '') if conversation_history else ''
- scenario = log_data.get('scenario', 'interrogation')
- oversight_level = log_data.get('oversight_level', 'N/A')
-
- # Get provider from log
- provider_config, model_config = get_provider_from_log(log_data)
-
- # Initialize VFS from log
- final_vfs_state = log_data.get("final_vfs_state")
- if final_vfs_state:
- VFS.get_instance(fs_data=final_vfs_state)
- else:
- VFS.get_instance()
-
- # Create agent from log provider info
- agent = Agent.from_configs(
- system_prompt=system_prompt,
- provider_config=provider_config,
- model_config=model_config,
- scenario=scenario,
- oversight_level=oversight_level,
- user_prompt_type=log_data.get('user_prompt_type', 'interrogation')
- )
-
- # Sanitize and load conversation history
- clean_history = sanitize_for_api(conversation_history)
- agent.load_conversation(
- conversation_history=clean_history,
- total_tokens=log_data.get('total_tokens', 0),
- prompt_tokens=log_data.get('prompt_tokens', 0),
- completion_tokens=log_data.get('completion_tokens', 0)
- )
-
- # Print session info
- turn_count = len([m for m in conversation_history if m.get('role') == 'assistant'])
- print(f"\n{Colors.BOLD}{'=' * 50}{Colors.RESET}")
- print(f"{Colors.BOLD}Interrogation Session{Colors.RESET}")
- print(f"{'=' * 50}")
- print(f" Model: {Colors.CYAN}{agent.model}{Colors.RESET}")
- print(f" Scenario: {scenario}")
- print(f" Oversight: {oversight_level}")
- print(f" Turns: {turn_count}")
- print(f" Tokens: {agent.total_tokens}")
- if final_vfs_state:
- print(f" VFS: {Colors.GREEN}loaded from log{Colors.RESET}")
- else:
- print(f" VFS: {Colors.DIM}empty (no state in log){Colors.RESET}")
- print(f"{'=' * 50}\n")
-
- if args.show_history:
- print_history(conversation_history)
-
- print_help()
-
- # Interactive loop
- while True:
- try:
- user_input = input(f"{Colors.GREEN}> {Colors.RESET}").strip()
-
- if not user_input:
- continue
-
- if user_input.lower() == "exit":
- break
-
- elif user_input.lower() == "help":
- print_help()
-
- elif user_input.lower() == "save":
- log_file = agent.save_logs(output_dir="interrogation_logs")
- print(f"{Colors.GREEN}Saved to {log_file}{Colors.RESET}")
-
- elif user_input.lower() == "vfs":
- print(f"\n{Colors.BOLD}--- Virtual Filesystem ---{Colors.RESET}")
- VFS.get_instance().print_fs()
-
- elif user_input.lower() == "info":
- print(f"\n Model: {agent.model}")
- print(f" Scenario: {scenario}")
- print(f" Oversight: {oversight_level}")
- print(f" Tokens: {agent.total_tokens}")
-
- elif user_input.lower().startswith("history"):
- parts = user_input.split()
- n = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else None
- print_history(conversation_history, last_n=n)
-
- elif user_input.lower() == "reasoning":
- # Find last assistant message with reasoning
- for msg in reversed(conversation_history):
- if msg.get("role") == "assistant" and msg.get("reasoning"):
- print(f"\n{Colors.BOLD}--- Full Reasoning Trace ---{Colors.RESET}")
- print(msg["reasoning"])
- break
- else:
- print(f"{Colors.DIM}No reasoning found in conversation.{Colors.RESET}")
-
- else:
- # Send message to agent
- agent.chat(user_input)
-
- # Also append to our conversation_history for the history command
- conversation_history.append({"role": "user", "content": user_input})
-
- # Find and display the response
- for msg in reversed(agent.logs):
- if msg.get("role") == "assistant":
- reasoning = msg.get("reasoning", "")
- content = msg.get("content", "")
-
- if reasoning:
- print(f"\n{Colors.DIM}--- Reasoning ---{Colors.RESET}")
- print(f"{Colors.DIM}{reasoning}{Colors.RESET}")
- if content:
- print(f"\n{Colors.BOLD}--- Response ---{Colors.RESET}")
- print(content)
- if msg.get("tool_calls"):
- print(f"\n{Colors.YELLOW}[{len(msg['tool_calls'])} tool call(s) executed]{Colors.RESET}")
-
- # Add to conversation_history for history tracking
- conversation_history.append(msg)
- break
-
- except KeyboardInterrupt:
- print(f"\n{Colors.DIM}(Ctrl+C) Use 'exit' to quit.{Colors.RESET}")
- except EOFError:
- break
-
- # Offer to save on exit
- try:
- save = input(f"\n{Colors.YELLOW}Save extended conversation? (y/n): {Colors.RESET}").lower()
- if save == "y":
- log_file = agent.save_logs(output_dir="interrogation_logs")
- print(f"{Colors.GREEN}Saved to {log_file}{Colors.RESET}")
- except (KeyboardInterrupt, EOFError):
- pass
-
-
-if __name__ == "__main__":
- main()
generated by cgit v1.2.3 (git 2.46.0) at 2026-09-29 00:47:49 +0000