1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
|
"""
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, Any, Optional
from agent import Agent
from config_loader import ConfigLoader, ProviderConfig, ModelConfig
from provider import create_provider_adapter
from vfs import VFS
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", {})
# Determine provider from model ID first (more reliable), then base_url
model_lower = model_id.lower()
if "claude" in model_lower:
provider_name = "anthropic"
# Claude doesn't use base_url
base_url = ""
elif "gemini" in model_lower:
provider_name = "google"
elif "kimi" in model_lower or "moonshot" in model_lower:
provider_name = "moonshot"
elif "openrouter" in base_url:
provider_name = "openrouter"
elif "openai" in base_url or "generativelanguage" in base_url:
provider_name = "google" if "generativelanguage" in base_url else "openai"
else:
# Default to openai-compatible
provider_name = "openai"
# Create configs
provider_config = ProviderConfig(
name=provider_name,
api_key_env=f"{provider_name.upper()}_API_KEY",
base_url=base_url
)
model_config = ModelConfig(
id=model_id,
provider=provider_name,
temperature=temperature,
extra_body=extra_body
)
return provider_config, model_config
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)")
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)
print("--- VFS State Loaded from Log ---")
VFS.get_instance().print_fs()
else:
VFS.get_instance()
print("--- No VFS State in Log, Initializing Empty VFS ---")
# 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')
)
# Load conversation history (skip system prompt as it's already in system_prompt)
agent.load_conversation(
conversation_history=conversation_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("\n--- Loaded Conversation ---")
print(f"Model: {agent.model}")
print(f"Scenario: {scenario}")
print(f"Oversight: {oversight_level}")
print(f"Total tokens so far: {agent.total_tokens}")
print(f"Conversation turns: {len([m for m in conversation_history if m.get('role') in ['user', 'assistant', 'tool']])}")
# Start interactive session
print("\nStarting interactive session... (type 'exit' to quit, 'save' to save)")
while True:
try:
user_input = input("\nYour turn: ")
if user_input.lower() == 'exit':
break
if user_input.lower() == 'save':
agent.save_logs()
continue
agent.chat(user_input)
# Find the last assistant message to print response
found = False
for msg in reversed(agent.logs):
if msg.get('role') == 'assistant':
content = msg.get('content', '')
reasoning = msg.get('reasoning', '')
print(f"\n--- Assistant Response ---")
if reasoning:
# print(f"Reasoning: {reasoning[:300]}{'...' if len(reasoning) > 300 else ''}")
print(f"Reasoning: {reasoning})
if content:
# print(f"Content: {content[:300]}{'...' if len(content) > 300 else ''}")
print(f"Content: {content})
if msg.get('tool_calls'):
print(f"Tool calls: {len(msg['tool_calls'])}")
found = True
break
if not found:
print("No assistant message found in logs")
except KeyboardInterrupt:
print("\nExiting...")
break
# Ask to save
save_choice = input("\nSave the extended conversation to a new log file? (y/n): ").lower()
if save_choice == 'y':
agent.save_logs()
if __name__ == "__main__":
main()
|
