File size: 5,364 Bytes
dc893fb | 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 | """Agent run logger"""
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from .schema import Message, ToolCall
class AgentLogger:
"""Agent run logger
Responsible for recording the complete interaction process of each agent run, including:
- LLM requests and responses
- Tool calls and results
"""
def __init__(self):
"""Initialize logger
Logs are stored in ~/.mini-agent/log/ directory
"""
# Use ~/.mini-agent/log/ directory for logs
self.log_dir = Path.home() / ".mini-agent" / "log"
self.log_dir.mkdir(parents=True, exist_ok=True)
self.log_file = None
self.log_index = 0
def start_new_run(self):
"""Start new run, create new log file"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_filename = f"agent_run_{timestamp}.log"
self.log_file = self.log_dir / log_filename
self.log_index = 0
# Write log header
with open(self.log_file, "w", encoding="utf-8") as f:
f.write("=" * 80 + "\n")
f.write(f"Agent Run Log - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("=" * 80 + "\n\n")
def log_request(self, messages: list[Message], tools: list[Any] | None = None):
"""Log LLM request
Args:
messages: Message list
tools: Tool list (optional)
"""
self.log_index += 1
# Build complete request data structure
request_data = {
"messages": [],
"tools": [],
}
# Convert messages to JSON serializable format
for msg in messages:
msg_dict = {
"role": msg.role,
"content": msg.content,
}
if msg.thinking:
msg_dict["thinking"] = msg.thinking
if msg.tool_calls:
msg_dict["tool_calls"] = [tc.model_dump() for tc in msg.tool_calls]
if msg.tool_call_id:
msg_dict["tool_call_id"] = msg.tool_call_id
if msg.name:
msg_dict["name"] = msg.name
request_data["messages"].append(msg_dict)
# Only record tool names
if tools:
request_data["tools"] = [tool.name for tool in tools]
# Format as JSON
content = "LLM Request:\n\n"
content += json.dumps(request_data, indent=2, ensure_ascii=False)
self._write_log("REQUEST", content)
def log_response(
self,
content: str,
thinking: str | None = None,
tool_calls: list[ToolCall] | None = None,
finish_reason: str | None = None,
):
"""Log LLM response
Args:
content: Response content
thinking: Thinking content (optional)
tool_calls: Tool call list (optional)
finish_reason: Finish reason (optional)
"""
self.log_index += 1
# Build complete response data structure
response_data = {
"content": content,
}
if thinking:
response_data["thinking"] = thinking
if tool_calls:
response_data["tool_calls"] = [tc.model_dump() for tc in tool_calls]
if finish_reason:
response_data["finish_reason"] = finish_reason
# Format as JSON
log_content = "LLM Response:\n\n"
log_content += json.dumps(response_data, indent=2, ensure_ascii=False)
self._write_log("RESPONSE", log_content)
def log_tool_result(
self,
tool_name: str,
arguments: dict[str, Any],
result_success: bool,
result_content: str | None = None,
result_error: str | None = None,
):
"""Log tool execution result
Args:
tool_name: Tool name
arguments: Tool arguments
result_success: Whether successful
result_content: Result content (on success)
result_error: Error message (on failure)
"""
self.log_index += 1
# Build complete tool execution result data structure
tool_result_data = {
"tool_name": tool_name,
"arguments": arguments,
"success": result_success,
}
if result_success:
tool_result_data["result"] = result_content
else:
tool_result_data["error"] = result_error
# Format as JSON
content = "Tool Execution:\n\n"
content += json.dumps(tool_result_data, indent=2, ensure_ascii=False)
self._write_log("TOOL_RESULT", content)
def _write_log(self, log_type: str, content: str):
"""Write log entry
Args:
log_type: Log type (REQUEST, RESPONSE, TOOL_RESULT)
content: Log content
"""
if self.log_file is None:
return
with open(self.log_file, "a", encoding="utf-8") as f:
f.write("\n" + "-" * 80 + "\n")
f.write(f"[{self.log_index}] {log_type}\n")
f.write(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]}\n")
f.write("-" * 80 + "\n")
f.write(content + "\n")
def get_log_file_path(self) -> Path:
"""Get current log file path"""
return self.log_file
|