๐ Token Usage
@@ -1040,6 +1125,120 @@
`;
}
+ function generateToolIterationsSection(req) {
+ const iterations = req.tool_iterations || [];
+ if (iterations.length === 0) {
+ return '';
+ }
+
+ const iterationsHTML = iterations.map(iter => {
+ // Parse tool results
+ const toolCalls = parseToolResults(iter.tool_results);
+
+ const toolCallsHTML = toolCalls.map(tool => `
+
+ `).join('');
+
+ return `
+
+
+ ${toolCallsHTML}
+
+ `;
+ }).join('');
+
+ return `
+
+
๐ ๏ธ Tool Calls (${iterations.length} iteration${iterations.length > 1 ? 's' : ''})
+
+ ${iterationsHTML}
+
+
+ `;
+ }
+
+ function parseToolResults(toolResultsStr) {
+ const tools = [];
+ if (!toolResultsStr) return tools;
+
+ // Split by "Tool: " to get each tool call
+ const parts = toolResultsStr.split(/Tool:\s+/);
+
+ for (let i = 1; i < parts.length; i++) {
+ const part = parts[i];
+
+ // Extract tool name (first line)
+ const lines = part.split('\n');
+ const name = lines[0].trim();
+
+ // Extract parameters JSON
+ let parameters = '';
+ const paramsMatch = part.match(/Parameters:\s*(\{[\s\S]*?\})\s*(?=Result:|$)/);
+ if (paramsMatch) {
+ try {
+ // Parse and re-stringify for nice formatting
+ const parsed = JSON.parse(paramsMatch[1]);
+ // Remove reasoning from display params (it's shown separately)
+ const displayParams = {...parsed};
+ delete displayParams.reasoning;
+ parameters = JSON.stringify(displayParams, null, 2);
+ } catch (e) {
+ parameters = paramsMatch[1].trim();
+ }
+ }
+
+ // Extract reasoning from llm_reasoning field in result or parameters
+ let reasoning = '';
+ const reasoningMatch = part.match(/"(?:reasoning|llm_reasoning)":\s*"([^"]+)"/);
+ if (reasoningMatch) {
+ reasoning = reasoningMatch[1];
+ }
+
+ // Extract result JSON
+ const resultMatch = part.match(/Result:\s*([\s\S]*?)(?=---|\Z)/);
+ let result = resultMatch ? resultMatch[1].trim() : '';
+
+ // Create preview
+ let resultPreview = 'click to expand';
+ try {
+ const parsed = JSON.parse(result);
+ if (parsed.total_found !== undefined) {
+ resultPreview = `${parsed.total_found} nodes found`;
+ } else if (parsed.node_id !== undefined) {
+ resultPreview = `Node ${parsed.node_id}`;
+ } else if (parsed.from_node !== undefined) {
+ resultPreview = `Path from ${parsed.from_node}`;
+ }
+ } catch (e) {}
+
+ tools.push({ name, parameters, reasoning, result, resultPreview });
+ }
+
+ return tools;
+ }
+
function toggleRequest(requestId) {
const card = document.getElementById(requestId);
if (!card) return;
@@ -1141,15 +1340,19 @@
.map(r => r.tokens?.completion || 0)
.reduce((a, b) => a + b, 0);
- // Gemini pricing (approximate)
- const inputCost = (promptTokens / 1000000) * 0.075;
- const outputCost = (completionTokens / 1000000) * 0.30;
+ // Gemini 3 Flash Preview pricing
+ const inputCost = (promptTokens / 1000000) * 0.50; // $0.50 per 1M input
+ const outputCost = (completionTokens / 1000000) * 3.00; // $3.00 per 1M output
const totalCost = (inputCost + outputCost).toFixed(4);
// Calculate success rate
const successCount = requests.filter(r => r.success).length;
const successRate = total > 0 ? Math.round((successCount / total) * 100) : 0;
+ // Count tool iterations
+ const toolIterations = requests.reduce((sum, r) =>
+ sum + (r.tool_iterations ? r.tool_iterations.length : 0), 0);
+
return {
total,
new: newCount,
@@ -1157,7 +1360,8 @@
avgTokens,
totalTokens,
totalCost,
- successRate
+ successRate,
+ toolIterations
};
}
diff --git a/examples/ai_testing/test_tools_integration.py b/examples/ai_testing/test_tools_integration.py
new file mode 100644
index 0000000000000000000000000000000000000000..8e67a5faed69484867a470de0f78583242be76fc
--- /dev/null
+++ b/examples/ai_testing/test_tools_integration.py
@@ -0,0 +1,288 @@
+"""
+Test Tool Integration for AI Agents
+
+This script tests the full tool calling flow:
+1. AgentTools - The tools themselves
+2. ToolExecutor - Execution and logging
+3. LLMClient - Function calling support
+4. AIManager - Full integration
+
+Run this to verify tools work end-to-end.
+"""
+
+import json
+from pathlib import Path
+from pycatan.ai.agent_tools import AgentTools
+from pycatan.ai.tool_executor import ToolExecutor, ToolCall
+
+
+def test_basic_tools():
+ """Test basic tool functionality."""
+ print("=" * 60)
+ print("TEST 1: Basic Tool Operations")
+ print("=" * 60)
+
+ # Load a sample game state
+ sample_state_path = Path("examples/ai_testing/sample_states/captured_game.json")
+ if not sample_state_path.exists():
+ print(f"โ ๏ธ Sample state not found at {sample_state_path}")
+ print("Using minimal test state instead")
+ game_state = {
+ "board": {
+ "nodes": [
+ {
+ "id": 10,
+ "adjacent_tiles": [0, 1, 2],
+ "neighbors": [8, 11, 14],
+ "port": None,
+ "building": None
+ },
+ {
+ "id": 14,
+ "adjacent_tiles": [2, 3, 4],
+ "neighbors": [10, 11, 18],
+ "port": "3:1",
+ "building": None
+ }
+ ],
+ "tiles": [
+ {"id": 0, "type": "Wood", "number": 6},
+ {"id": 1, "type": "Brick", "number": 8},
+ {"id": 2, "type": "Wheat", "number": 5},
+ {"id": 3, "type": "Ore", "number": 10},
+ {"id": 4, "type": "Sheep", "number": 9}
+ ]
+ }
+ }
+ else:
+ with open(sample_state_path, 'r', encoding='utf-8') as f:
+ game_state = json.load(f)
+
+ # Initialize tools
+ tools = AgentTools(game_state)
+ print(f"โ
Initialized AgentTools with {len(tools.node_lookup)} nodes")
+ print()
+
+ # Test 1: inspect_node
+ print("๐ง Testing: inspect_node(10)")
+ result1 = tools.inspect_node(10)
+ print(json.dumps(result1, indent=2))
+ print()
+
+ # Test 2: inspect_node on another node
+ print("๐ง Testing: inspect_node(14)")
+ result2 = tools.inspect_node(14)
+ print(json.dumps(result2, indent=2))
+ print()
+
+ # Test 3: find_best_nodes
+ print("๐ง Testing: find_best_nodes(min_pips=8)")
+ result3 = tools.find_best_nodes(min_pips=8, limit=5)
+ print(f"Found {result3['total_found']} nodes:")
+ for node in result3['nodes']:
+ print(f" - Node {node['node_id']}: {node['total_pips']} pips, score={node['score']}")
+ print()
+
+ # Test 4: analyze_path_potential
+ if len(tools.node_lookup) > 10:
+ print("๐ง Testing: analyze_path_potential(from_node=10)")
+ result4 = tools.analyze_path_potential(from_node=10, max_depth=2)
+ print(f"Analyzed {result4['total_directions']} directions from node 10")
+ if result4['paths']:
+ best_path = result4['paths'][0]
+ print(f" Best direction: node {best_path['direction']} (score={best_path['score']})")
+ if best_path['highlights']:
+ for highlight in best_path['highlights']:
+ print(f" - {highlight}")
+ print()
+
+
+def test_tool_executor():
+ """Test ToolExecutor with multiple tool calls."""
+ print("=" * 60)
+ print("TEST 2: Tool Executor (Multiple Calls)")
+ print("=" * 60)
+
+ # Load sample state
+ sample_state_path = Path("examples/ai_testing/sample_states/captured_game.json")
+ if sample_state_path.exists():
+ with open(sample_state_path, 'r', encoding='utf-8') as f:
+ game_state = json.load(f)
+ else:
+ # Minimal state
+ game_state = {
+ "board": {
+ "nodes": [
+ {
+ "id": 10,
+ "adjacent_tiles": [0, 1],
+ "neighbors": [8, 14],
+ "port": None,
+ "building": None
+ }
+ ],
+ "tiles": [
+ {"id": 0, "type": "Wood", "number": 6},
+ {"id": 1, "type": "Brick", "number": 8}
+ ]
+ }
+ }
+
+ # Initialize
+ tools = AgentTools(game_state)
+ executor = ToolExecutor(tools)
+ print(f"โ
Initialized ToolExecutor")
+ print()
+
+ # Simulate multiple tool calls (as if from LLM)
+ tool_calls = [
+ {
+ "id": "call_1",
+ "name": "inspect_node",
+ "parameters": {"node_id": 10}
+ },
+ {
+ "id": "call_2",
+ "name": "find_best_nodes",
+ "parameters": {"min_pips": 8, "limit": 3}
+ },
+ ]
+
+ # Add path analysis if we have enough nodes
+ if len(tools.node_lookup) > 10:
+ tool_calls.append({
+ "id": "call_3",
+ "name": "analyze_path_potential",
+ "parameters": {"from_node": 10, "max_depth": 1}
+ })
+
+ print(f"๐ง Executing {len(tool_calls)} tool calls...")
+ print()
+
+ # Execute
+ batch = executor.execute_tool_calls(tool_calls)
+
+ # Display results
+ print()
+ print("๐ Execution Summary:")
+ print(f" Total calls: {len(batch.tool_calls)}")
+ print(f" Successful: {batch.success_count}")
+ print(f" Failed: {batch.failure_count}")
+ print(f" Total time: {batch.total_time*1000:.1f}ms")
+ print(f" Total tokens: {batch.total_tokens}")
+ print(f" - Input: {batch.total_input_tokens}")
+ print(f" - Output: {batch.total_output_tokens}")
+ print()
+
+ # Show individual call details
+ print("๐ Individual Call Details:")
+ for call in batch.tool_calls:
+ status = "โ
" if call.success else "โ"
+ print(f"\n {status} {call.name}({call.parameters})")
+ print(f" Time: {call.execution_time*1000:.1f}ms")
+ print(f" Tokens: {call.input_tokens} in + {call.output_tokens} out")
+ if call.success:
+ result_preview = str(call.result)[:100]
+ print(f" Result: {result_preview}...")
+ else:
+ print(f" Error: {call.error}")
+ print()
+
+ # Format for LLM
+ print("๐ Formatted for LLM:")
+ print("-" * 60)
+ formatted = executor.format_tool_results_for_llm(batch)
+ print(formatted)
+ print("-" * 60)
+ print()
+
+
+def test_tool_schemas():
+ """Test tool schema generation for LLM."""
+ print("=" * 60)
+ print("TEST 3: Tool Schemas for LLM")
+ print("=" * 60)
+
+ tools = AgentTools()
+ schemas = tools.get_tools_schema()
+
+ print(f"โ
Generated {len(schemas)} tool schemas")
+ print()
+
+ for schema in schemas:
+ print(f"๐ Tool: {schema['name']}")
+ print(f" Description: {schema['description'][:80]}...")
+ print(f" Parameters:")
+ for param_name, param_info in schema['parameters']['properties'].items():
+ required = "required" if param_name in schema['parameters'].get('required', []) else "optional"
+ print(f" - {param_name} ({required}): {param_info.get('description', 'N/A')[:60]}...")
+ print()
+
+
+def test_execution_history():
+ """Test execution history and statistics."""
+ print("=" * 60)
+ print("TEST 4: Execution History & Statistics")
+ print("=" * 60)
+
+ # Create minimal state
+ game_state = {
+ "board": {
+ "nodes": [{"id": 10, "adjacent_tiles": [], "neighbors": [], "port": None, "building": None}],
+ "tiles": []
+ }
+ }
+
+ tools = AgentTools(game_state)
+ executor = ToolExecutor(tools)
+
+ # Execute multiple batches
+ print("๐ง Executing 3 batches of tool calls...")
+ for i in range(3):
+ tool_calls = [
+ {"id": f"batch{i}_call1", "name": "inspect_node", "parameters": {"node_id": 10}}
+ ]
+ executor.execute_tool_calls(tool_calls)
+
+ print()
+
+ # Get summary
+ summary = executor.get_execution_summary()
+
+ print("๐ Execution Summary:")
+ print(f" Total batches: {summary['total_batches']}")
+ print(f" Total calls: {summary['total_calls']}")
+ print(f" Success rate: {summary['success_rate']}")
+ print(f" Total tokens: {summary['total_tokens']}")
+ print()
+
+ print("๐ง Tool Usage:")
+ for tool_name, count in summary['tool_usage'].items():
+ print(f" - {tool_name}: {count} times")
+ print()
+
+
+if __name__ == "__main__":
+ print("\n๐งช Testing Tool Integration for AI Agents\n")
+
+ try:
+ test_basic_tools()
+ test_tool_executor()
+ test_tool_schemas()
+ test_execution_history()
+
+ print("=" * 60)
+ print("โ
All Tests Passed!")
+ print("=" * 60)
+ print()
+ print("Next steps:")
+ print("1. Tools are ready to use with AI agents")
+ print("2. LLM can call these tools via function calling")
+ print("3. All executions are logged with token counts")
+ print("4. Check tool_executions.json in session logs for details")
+ print()
+
+ except Exception as e:
+ print(f"\nโ Test failed: {e}")
+ import traceback
+ traceback.print_exc()
diff --git a/examples/ai_testing/web_viewer.py b/examples/ai_testing/web_viewer.py
index 9fd047a8ae9c27786e0fd5029f1a78f7368c1fa4..6769ffdc28cc04c3b8653a12c311a21ae44d4d71 100644
--- a/examples/ai_testing/web_viewer.py
+++ b/examples/ai_testing/web_viewer.py
@@ -99,6 +99,10 @@ def get_session_data(session_path):
# Load all prompts for this player
for prompt_file in sorted(prompts_dir.glob("prompt_*.json")):
+ # Skip iteration files - they're in iterations subfolder
+ if "iter" in prompt_file.name:
+ continue
+
try:
with open(prompt_file, 'r', encoding='utf-8') as f:
prompt_data = json.load(f)
@@ -112,6 +116,22 @@ def get_session_data(session_path):
with open(response_file, 'r', encoding='utf-8') as f:
response_data = json.load(f)
+ # Load tool iterations for this prompt
+ tool_iterations = []
+ iterations_dir = prompts_dir / "iterations"
+ if iterations_dir.exists():
+ for iter_file in sorted(iterations_dir.glob(f"prompt_{req_num}_iter*.json")):
+ try:
+ with open(iter_file, 'r', encoding='utf-8') as f:
+ iter_data = json.load(f)
+ tool_iterations.append({
+ "iteration": iter_data.get("iteration", 0),
+ "tool_results": iter_data.get("tool_results", ""),
+ "timestamp": iter_data.get("timestamp", "")
+ })
+ except Exception as e:
+ print(f" [!] Error loading iteration {iter_file.name}: {e}")
+
requests_data.append({
"player_name": player_name,
"request_number": req_num,
@@ -120,7 +140,8 @@ def get_session_data(session_path):
"response": response_data.get("parsed", {}) if response_data else None,
"raw_response": response_data.get("raw_content", "") if response_data else None,
"tokens": response_data.get("tokens", {}) if response_data else {},
- "success": response_data.get("success", False) if response_data else False
+ "success": response_data.get("success", False) if response_data else False,
+ "tool_iterations": tool_iterations
})
except Exception as e:
diff --git a/pycatan/ai/agent_tools.py b/pycatan/ai/agent_tools.py
index 702c4be75b52a596ca37de8f8ba3e8983d1445a6..7208c708057ba8f6fe7bfc01d801b01b5e9779b0 100644
--- a/pycatan/ai/agent_tools.py
+++ b/pycatan/ai/agent_tools.py
@@ -53,30 +53,88 @@ class AgentTools:
self._update_internal_structures()
def _update_internal_structures(self):
- """Build internal lookup structures from game state for fast queries."""
- # Extract board data
- self.board_data = self.game_state.get("board", {})
- self.nodes = self.board_data.get("nodes", [])
- self.tiles = self.board_data.get("tiles", [])
+ """
+ Build internal lookup structures from game state for fast queries.
- # Build node lookup: node_id -> node_data
- self.node_lookup: Dict[int, Dict[str, Any]] = {}
- for node in self.nodes:
- node_id = node.get("id")
- if node_id is not None:
- self.node_lookup[node_id] = node
+ Supports the COMPACT format from state_optimizer:
+ - H: Array of hex data. H[id] = "W12" (Wood, number 12)
+ - N: Array of node data. N[id] = [[neighbors], [hex_ids], port?]
+ - state: {"bld": [[node, owner, type], ...], "rds": [[[from,to], owner], ...]}
+
+ Resource codes: W=Wood, B=Brick, S=Sheep, Wh=Wheat, O=Ore, D=Desert
+ Port codes: ?3=Any 3:1, W2=Wood 2:1, etc.
+ """
+ # Resource code mapping (compact -> full name)
+ RES_DECODE = {"W": "Wood", "B": "Brick", "S": "Sheep", "Wh": "Wheat", "O": "Ore", "D": "Desert"}
+
+ # Extract compact arrays
+ self.H = self.game_state.get("H", []) # Hex array
+ self.N = self.game_state.get("N", []) # Node array
+ self.state = self.game_state.get("state", {}) # Buildings & roads
- # Build tile lookup: tile_id -> tile_data
+ # Build hex lookup: hex_id -> {type, number}
self.tile_lookup: Dict[int, Dict[str, Any]] = {}
- for tile in self.tiles:
- tile_id = tile.get("id")
- if tile_id is not None:
- self.tile_lookup[tile_id] = tile
+ for hex_id, hex_val in enumerate(self.H):
+ if hex_val: # Skip empty entries
+ # Parse "W12" -> type="Wood", number=12
+ # Or "D" -> type="Desert", number=0
+ resource_code = ""
+ number = 0
+
+ # Handle "Wh" (2 char) vs "W" (1 char)
+ if hex_val.startswith("Wh"):
+ resource_code = "Wh"
+ num_str = hex_val[2:]
+ elif len(hex_val) >= 1:
+ resource_code = hex_val[0]
+ num_str = hex_val[1:]
+
+ if num_str.isdigit():
+ number = int(num_str)
+
+ self.tile_lookup[hex_id] = {
+ "type": RES_DECODE.get(resource_code, resource_code),
+ "number": number
+ }
+
+ # Build buildings lookup from state.bld: [[node, owner, type], ...]
+ self.buildings: Dict[int, Dict[str, Any]] = {} # node_id -> building info
+ for bld in self.state.get("bld", []):
+ if len(bld) >= 3:
+ node_id, owner, bld_type = bld[0], bld[1], bld[2]
+ self.buildings[node_id] = {
+ "owner": owner,
+ "type": "settlement" if bld_type == "S" else "city"
+ }
+
+ # Build node lookup: node_id -> node_data (converted from compact format)
+ self.node_lookup: Dict[int, Dict[str, Any]] = {}
+ for node_id, node_val in enumerate(self.N):
+ if node_val is not None: # Skip null entries (index 0 is often null)
+ # N[id] = [[neighbors], [hex_ids], port?]
+ neighbors = node_val[0] if len(node_val) > 0 else []
+ hex_ids = node_val[1] if len(node_val) > 1 else []
+ port = node_val[2] if len(node_val) > 2 else None
+
+ # Check if this node has a building
+ building = self.buildings.get(node_id)
+
+ self.node_lookup[node_id] = {
+ "id": node_id,
+ "neighbors": neighbors,
+ "adjacent_tiles": hex_ids,
+ "port": port,
+ "building": building
+ }
+
+ # For backwards compatibility
+ self.nodes = list(self.node_lookup.values())
+ self.tiles = list(self.tile_lookup.values())
# ========== TOOL 1: Inspect Node (Prevents Hallucinations) ==========
- def inspect_node(self, node_id: int) -> Dict[str, Any]:
+ def inspect_node(self, node_id: int, reasoning: str = "") -> Dict[str, Any]:
"""
Get detailed, processed information about a specific node.
@@ -86,6 +144,7 @@ class AgentTools:
Args:
node_id: The node ID to inspect (e.g., 10, 18, 40)
+ reasoning: LLM's explanation for why it's inspecting this node
Returns:
Dictionary with complete node information
@@ -95,7 +154,8 @@ class AgentTools:
return {
"node_id": node_id,
"exists": False,
- "error": f"Node {node_id} does not exist on the board"
+ "error": f"Node {node_id} does not exist on the board",
+ "llm_reasoning": reasoning
}
node = self.node_lookup[node_id]
@@ -164,7 +224,8 @@ class AgentTools:
"occupied_by": occupied_by,
"building_type": building_type,
"can_build_here": can_build_here,
- "blocked_reason": blocked_reason
+ "blocked_reason": blocked_reason,
+ "llm_reasoning": reasoning
}
@@ -172,6 +233,7 @@ class AgentTools:
def find_best_nodes(
self,
+ reasoning: str = "",
min_pips: int = 0,
must_have_resource: Optional[str] = None,
exclude_blocked: bool = True,
@@ -186,6 +248,7 @@ class AgentTools:
tool to find the best positions that match specific criteria.
Args:
+ reasoning: LLM's explanation for this search strategy
min_pips: Minimum total pip value (e.g., 10 for high-probability spots)
must_have_resource: Required resource type (e.g., "Wheat", "Ore")
exclude_blocked: Skip nodes that can't be built on
@@ -261,6 +324,7 @@ class AgentTools:
matching_nodes = matching_nodes[:limit]
return {
+ "llm_reasoning": reasoning,
"query": {
"min_pips": min_pips,
"must_have_resource": must_have_resource,
@@ -277,6 +341,7 @@ class AgentTools:
def analyze_path_potential(
self,
from_node: int,
+ reasoning: str = "",
direction_node: Optional[int] = None,
max_depth: int = 2
) -> Dict[str, Any]:
@@ -289,6 +354,7 @@ class AgentTools:
Args:
from_node: Starting node ID
+ reasoning: LLM's explanation for analyzing this path
direction_node: Target direction (neighbor node ID), or None to check all directions
max_depth: How many steps ahead to look (1 or 2)
@@ -296,11 +362,12 @@ class AgentTools:
Dictionary with path analysis
"""
# Get starting node info
- from_info = self.inspect_node(from_node)
+ from_info = self.inspect_node(from_node, reasoning="Internal call for path analysis")
if not from_info.get("exists"):
return {
"error": f"Starting node {from_node} does not exist",
- "from_node": from_node
+ "from_node": from_node,
+ "llm_reasoning": reasoning
}
neighbors = from_info.get("neighbors", [])
@@ -311,7 +378,8 @@ class AgentTools:
return {
"error": f"Node {direction_node} is not a neighbor of {from_node}",
"from_node": from_node,
- "neighbors": neighbors
+ "neighbors": neighbors,
+ "llm_reasoning": reasoning
}
neighbors = [direction_node]
@@ -327,7 +395,7 @@ class AgentTools:
}
# Depth 1: Immediate neighbor
- depth_1_info = self.inspect_node(neighbor_id)
+ depth_1_info = self.inspect_node(neighbor_id, reasoning="Internal call for path analysis depth 1")
if depth_1_info.get("exists"):
path_analysis["depth_1"] = {
"node_id": neighbor_id,
@@ -366,7 +434,7 @@ class AgentTools:
if depth_2_id == from_node: # Don't go back
continue
- depth_2_info = self.inspect_node(depth_2_id)
+ depth_2_info = self.inspect_node(depth_2_id, reasoning="Internal call for path analysis depth 2")
if depth_2_info.get("exists"):
node_data = {
"node_id": depth_2_id,
@@ -411,6 +479,7 @@ class AgentTools:
paths.sort(key=lambda p: p["score"], reverse=True)
return {
+ "llm_reasoning": reasoning,
"from_node": from_node,
"total_directions": len(paths),
"paths": paths
@@ -436,12 +505,16 @@ class AgentTools:
"parameters": {
"type": "object",
"properties": {
+ "reasoning": {
+ "type": "string",
+ "description": "Explain WHY you're inspecting this specific node. What are you trying to verify or learn?"
+ },
"node_id": {
"type": "integer",
"description": "The node ID to inspect (e.g., 10, 18, 40)"
}
},
- "required": ["node_id"]
+ "required": ["reasoning", "node_id"]
}
},
{
@@ -450,6 +523,10 @@ class AgentTools:
"parameters": {
"type": "object",
"properties": {
+ "reasoning": {
+ "type": "string",
+ "description": "Explain your search strategy. What kind of position are you looking for and why?"
+ },
"min_pips": {
"type": "integer",
"description": "Minimum total pip value (probability). Good nodes have 10+, excellent have 12+",
@@ -476,7 +553,7 @@ class AgentTools:
"default": 10
}
},
- "required": []
+ "required": ["reasoning"]
}
},
{
@@ -485,6 +562,10 @@ class AgentTools:
"parameters": {
"type": "object",
"properties": {
+ "reasoning": {
+ "type": "string",
+ "description": "Explain your road-building strategy. Why analyze this path? What are you hoping to reach?"
+ },
"from_node": {
"type": "integer",
"description": "Starting node ID (where you currently have a settlement/road)"
@@ -500,7 +581,7 @@ class AgentTools:
"default": 2
}
},
- "required": ["from_node"]
+ "required": ["reasoning", "from_node"]
}
}
]
diff --git a/pycatan/ai/ai_logger.py b/pycatan/ai/ai_logger.py
index 05c6f098538f8693b5a1461cc34017673b06c4e5..8f513b8a6049963f6e4e515349f019ee08defbb9 100644
--- a/pycatan/ai/ai_logger.py
+++ b/pycatan/ai/ai_logger.py
@@ -19,6 +19,10 @@ from typing import Dict, Any, Optional, List
from pycatan.ai.llm_client import LLMResponse
+# Gemini 3 Flash Preview pricing (per million tokens)
+GEMINI_FLASH_INPUT_PRICE = 0.50 # $0.50 per 1M tokens
+GEMINI_FLASH_OUTPUT_PRICE = 3.00 # $3.00 per 1M tokens
+
class AILogger:
"""
@@ -65,6 +69,17 @@ class AILogger:
# Track request numbers per player
self.request_counters: Dict[str, int] = {}
+ # Global API call index (across all players/tool iterations)
+ self.api_call_index: int = 0
+
+ # Track pending API calls (sent but not yet received)
+ self.pending_api_calls: Dict[int, Dict[str, Any]] = {}
+
+ # Track cumulative costs
+ self.cumulative_input_tokens: int = 0
+ self.cumulative_output_tokens: int = 0
+ self.cumulative_cost: float = 0.0
+
# Session start time
self.start_time = datetime.now()
@@ -102,13 +117,117 @@ class AILogger:
with open(current_session_file, 'w', encoding='utf-8') as f:
f.write(self.session_dir.name)
+ def log_api_call_start(
+ self,
+ player_name: str,
+ prompt_number: int,
+ iteration: int,
+ prompt_content: Optional[str] = None,
+ tools_schema: Optional[List[Dict]] = None,
+ is_tool_followup: bool = False
+ ) -> int:
+ """
+ Log the start of an API call to LLM. Returns the API call index.
+
+ Args:
+ player_name: Name of the player making the call
+ prompt_number: The prompt number (within this player's session)
+ iteration: Tool iteration number (1, 2, 3...)
+ prompt_content: The prompt content being sent (for logging)
+ tools_schema: The tools schema being sent with the call
+ is_tool_followup: Whether this is a follow-up call after tool execution
+
+ Returns:
+ The global API call index assigned to this call
+ """
+ self.api_call_index += 1
+ call_id = self.api_call_index
+
+ # Track pending call
+ self.pending_api_calls[call_id] = {
+ "player_name": player_name,
+ "prompt_number": prompt_number,
+ "iteration": iteration,
+ "start_time": datetime.now().isoformat(),
+ "is_tool_followup": is_tool_followup
+ }
+
+ # Log to communication file
+ call_type = "TOOL_FOLLOWUP" if is_tool_followup else "INITIAL"
+ msg = f"๐ค API Call #{call_id} SENT [{call_type}] - Player: {player_name}, Prompt: {prompt_number}, Iteration: {iteration}"
+ self.log_llm_communication(msg, "API_SEND")
+
+ # Log tools being sent
+ if tools_schema:
+ tool_names = [t.get("name", "unknown") for t in tools_schema]
+ self.log_llm_communication(f" ๐ง Tools enabled: {tool_names}", "API_SEND")
+
+ return call_id
+
+ def log_api_call_end(
+ self,
+ call_id: int,
+ success: bool,
+ tokens: int = 0,
+ prompt_tokens: int = 0,
+ completion_tokens: int = 0,
+ has_tool_calls: bool = False,
+ tool_calls_count: int = 0,
+ error: Optional[str] = None
+ ) -> None:
+ """
+ Log the completion of an API call.
+
+ Args:
+ call_id: The API call index returned by log_api_call_start
+ success: Whether the call succeeded
+ tokens: Total tokens used
+ prompt_tokens: Input tokens (for cost calculation)
+ completion_tokens: Output tokens (for cost calculation)
+ has_tool_calls: Whether the response includes tool calls
+ tool_calls_count: Number of tool calls requested
+ error: Error message if failed
+ """
+ # Get pending call info
+ call_info = self.pending_api_calls.pop(call_id, {})
+ player_name = call_info.get("player_name", "unknown")
+ iteration = call_info.get("iteration", 0)
+
+ if success:
+ # Calculate cost
+ input_cost = (prompt_tokens / 1_000_000) * GEMINI_FLASH_INPUT_PRICE
+ output_cost = (completion_tokens / 1_000_000) * GEMINI_FLASH_OUTPUT_PRICE
+ total_cost = input_cost + output_cost
+
+ # Update cumulative totals
+ self.cumulative_input_tokens += prompt_tokens
+ self.cumulative_output_tokens += completion_tokens
+ self.cumulative_cost += total_cost
+
+ if has_tool_calls:
+ msg = f"๐ฅ API Call #{call_id} RECEIVED โ
- {tokens} tokens (in:{prompt_tokens} out:{completion_tokens}), {tool_calls_count} tool request(s) | ๐ฐ ${total_cost:.6f}"
+ else:
+ msg = f"๐ฅ API Call #{call_id} RECEIVED โ
- {tokens} tokens (in:{prompt_tokens} out:{completion_tokens}) (final response) | ๐ฐ ${total_cost:.6f}"
+ self.log_llm_communication(msg, "API_RECV")
+
+ # Update cumulative summary at top of log file
+ self._update_cumulative_header()
+ else:
+ msg = f"๐ฅ API Call #{call_id} FAILED โ - Error: {error}"
+ self.log_llm_communication(msg, "API_RECV")
+
+ # Show pending calls status
+ if self.pending_api_calls:
+ pending_ids = list(self.pending_api_calls.keys())
+ self.log_llm_communication(f" โณ Still pending: API Call(s) {pending_ids}", "API_STATUS")
+
def log_llm_communication(self, message: str, msg_type: str = "INFO") -> None:
"""
Log a message to the LLM communication log file.
Args:
message: The message to log
- msg_type: Type of message (SEND, RECV, ERROR, INFO)
+ msg_type: Type of message (SEND, RECV, ERROR, INFO, API_SEND, API_RECV, API_STATUS)
"""
timestamp = datetime.now().strftime("%H:%M:%S")
line = f"[{timestamp}] [{msg_type}] {message}\n"
@@ -117,6 +236,42 @@ class AILogger:
f.write(line)
f.flush() # Ensure immediate write
+ def _update_cumulative_header(self) -> None:
+ """
+ Update the cumulative cost summary at the top of the log file.
+ Rewrites the header section with updated totals.
+ """
+ import re
+
+ # Read current content
+ with open(self.llm_log_file, 'r', encoding='utf-8') as f:
+ content = f.read()
+
+ # Find where the actual log entries start (lines starting with [HH:MM:SS])
+ # Everything before that is header
+ log_entry_match = re.search(r'^\[\d{2}:\d{2}:\d{2}\]', content, re.MULTILINE)
+ if log_entry_match:
+ log_content = content[log_entry_match.start():]
+ else:
+ log_content = ""
+
+ # Create new header with cumulative totals
+ new_header = f"""=== LLM Communication Log ===
+Session: {self.session_dir.name}
+Started: {self.start_time.isoformat()}
+{'=' * 50}
+
+๐ฐ CUMULATIVE COST: ${self.cumulative_cost:.6f}
+ Input: {self.cumulative_input_tokens:,} tokens (${(self.cumulative_input_tokens / 1_000_000) * GEMINI_FLASH_INPUT_PRICE:.6f})
+ Output: {self.cumulative_output_tokens:,} tokens (${(self.cumulative_output_tokens / 1_000_000) * GEMINI_FLASH_OUTPUT_PRICE:.6f})
+{'=' * 50}
+
+"""
+
+ # Write updated file
+ with open(self.llm_log_file, 'w', encoding='utf-8') as f:
+ f.write(new_header + log_content)
+
def _ensure_player_dirs(self, player_name: str) -> Dict[str, Path]:
"""
Ensure directories exist for a player.
@@ -152,7 +307,8 @@ class AILogger:
schema: Optional[Dict[str, Any]] = None,
is_active: bool = True,
what_happened: str = "",
- allowed_actions: Optional[List[Dict]] = None
+ allowed_actions: Optional[List[Dict]] = None,
+ tools_schema: Optional[List[Dict[str, Any]]] = None
) -> Dict[str, Any]:
"""
Log a prompt for a player.
@@ -164,6 +320,7 @@ class AILogger:
is_active: Whether this is an active turn prompt
what_happened: Description of what happened
allowed_actions: List of allowed actions
+ tools_schema: List of tool schemas being sent with this prompt
Returns:
Dict with 'number', 'json_path', 'txt_path'
@@ -186,6 +343,8 @@ class AILogger:
prompt_doc["what_happened"] = what_happened
if allowed_actions:
prompt_doc["allowed_actions"] = allowed_actions
+ if tools_schema:
+ prompt_doc["tools_schema"] = tools_schema
# Save JSON file
json_path = dirs["prompts"] / f"prompt_{num}.json"
@@ -196,16 +355,28 @@ class AILogger:
txt_path = dirs["prompts"] / f"prompt_{num}.txt"
prompt_str = json.dumps(prompt, indent=2, ensure_ascii=False)
schema_str = json.dumps(schema, indent=2, ensure_ascii=False) if schema else "N/A"
+
+ # Format tools schema for TXT
+ tools_str = "N/A"
+ if tools_schema:
+ tools_summary = []
+ for tool in tools_schema:
+ tool_name = tool.get("name", "unknown")
+ tool_desc = tool.get("description", "")[:80]
+ tools_summary.append(f" - {tool_name}: {tool_desc}...")
+ tools_str = "\n".join(tools_summary)
+
with open(txt_path, 'w', encoding='utf-8') as f:
f.write(f"=== Prompt #{num} for {player_name} ===\n")
f.write(f"Timestamp: {datetime.now().isoformat()}\n")
f.write(f"Active Turn: {is_active}\n")
f.write(f"\n--- What Happened ---\n{what_happened}\n")
+ f.write(f"\n--- Tools Available ---\n{tools_str}\n")
f.write(f"\n--- Response Schema ---\n{schema_str}\n")
f.write(f"\n--- Prompt Content ---\n{prompt_str}\n")
# Update MD log
- self._append_prompt_to_md(player_name, num, prompt, is_active, what_happened)
+ self._append_prompt_to_md(player_name, num, prompt, is_active, what_happened, tools_schema)
return {
"number": num,
@@ -213,6 +384,90 @@ class AILogger:
"txt_path": txt_path
}
+ def log_tool_followup_prompt(
+ self,
+ player_name: str,
+ original_prompt_number: int,
+ iteration: int,
+ conversation_context: str,
+ tool_results: str,
+ tools_schema: Optional[List[Dict[str, Any]]] = None,
+ schema: Optional[Dict[str, Any]] = None
+ ) -> Dict[str, Any]:
+ """
+ Log a follow-up prompt sent after tool execution.
+
+ This creates a new prompt file that shows what was sent back to LLM
+ after tool execution, including the tool results.
+
+ Args:
+ player_name: Name of the player
+ original_prompt_number: The original request number
+ iteration: Tool iteration number (2, 3, etc.)
+ conversation_context: Full conversation context being sent
+ tool_results: Formatted tool results string
+ tools_schema: Tools still available (may be empty on last iteration)
+ schema: Response schema
+
+ Returns:
+ Dict with 'path' to the saved file
+ """
+ dirs = self._ensure_player_dirs(player_name)
+
+ # Create iterations subdirectory
+ iterations_dir = dirs["prompts"] / "iterations"
+ iterations_dir.mkdir(exist_ok=True)
+
+ # Use naming like prompt_1_iter2.json
+ filename = f"prompt_{original_prompt_number}_iter{iteration}"
+
+ prompt_doc = {
+ "original_request_number": original_prompt_number,
+ "iteration": iteration,
+ "timestamp": datetime.now().isoformat(),
+ "player_name": player_name,
+ "type": "tool_followup",
+ "tool_results": tool_results,
+ "full_context_sent": conversation_context,
+ }
+
+ if tools_schema:
+ prompt_doc["tools_schema"] = tools_schema
+ prompt_doc["tools_enabled"] = True
+ else:
+ prompt_doc["tools_enabled"] = False
+
+ if schema:
+ prompt_doc["response_schema"] = schema
+
+ # Save JSON in iterations folder
+ json_path = iterations_dir / f"{filename}.json"
+ with open(json_path, 'w', encoding='utf-8') as f:
+ json.dump(prompt_doc, f, indent=2, ensure_ascii=False)
+
+ # Save TXT in iterations folder
+ txt_path = iterations_dir / f"{filename}.txt"
+ tools_enabled = "Yes" if tools_schema else "No (final iteration)"
+ with open(txt_path, 'w', encoding='utf-8') as f:
+ f.write(f"=== Tool Follow-up #{iteration} for Prompt #{original_prompt_number} ===\n")
+ f.write(f"Player: {player_name}\n")
+ f.write(f"Timestamp: {datetime.now().isoformat()}\n")
+ f.write(f"Tools Enabled: {tools_enabled}\n")
+ f.write(f"\n--- Tool Results Appended ---\n")
+ f.write(tool_results)
+ f.write(f"\n\n--- Full Context Sent to LLM ---\n")
+ f.write(conversation_context[:5000]) # Truncate for readability
+ if len(conversation_context) > 5000:
+ f.write(f"\n... (truncated, full length: {len(conversation_context)} chars)")
+
+ # Update MD log
+ self._append_tool_followup_to_md(player_name, original_prompt_number, iteration, tool_results)
+
+ return {
+ "path": json_path,
+ "txt_path": txt_path
+ }
+
def log_response(
self,
player_name: str,
@@ -288,7 +543,8 @@ class AILogger:
num: int,
prompt: Dict[str, Any],
is_active: bool,
- what_happened: str
+ what_happened: str,
+ tools_schema: Optional[List[Dict[str, Any]]] = None
) -> None:
"""Append prompt section to MD file."""
dirs = self._ensure_player_dirs(player_name)
@@ -300,10 +556,16 @@ class AILogger:
turn_type = "๐ฏ ACTIVE TURN" if is_active else "๐ OBSERVING"
+ # Format tools info
+ tools_info = ""
+ if tools_schema:
+ tool_names = [t.get("name", "unknown") for t in tools_schema]
+ tools_info = f"\n**Tools:** {', '.join(tool_names)}\n"
+
section = f"""
## Request #{num} - {turn_type}
-**Time:** {datetime.now().strftime('%H:%M:%S')}
+**Time:** {datetime.now().strftime('%H:%M:%S')}{tools_info}
### What Happened
{what_happened if what_happened else "(No events)"}
@@ -315,6 +577,33 @@ See: [prompt_{num}.json](prompts/prompt_{num}.json)
with open(md_path, 'a', encoding='utf-8') as f:
f.write(section)
+ def _append_tool_followup_to_md(
+ self,
+ player_name: str,
+ original_prompt_number: int,
+ iteration: int,
+ tool_results: str
+ ) -> None:
+ """Append tool follow-up section to MD file."""
+ dirs = self._ensure_player_dirs(player_name)
+ md_path = dirs["root"] / f"{player_name}.md"
+
+ section = f"""
+### ๐ง Tool Follow-up (Iteration {iteration})
+
+**Time:** {datetime.now().strftime('%H:%M:%S')}
+
+**Tool Results:**
+```
+{tool_results}
+```
+
+See: [prompt_{original_prompt_number}_iter{iteration}.json](prompts/iterations/prompt_{original_prompt_number}_iter{iteration}.json)
+
+"""
+ with open(md_path, 'a', encoding='utf-8') as f:
+ f.write(section)
+
def _append_response_to_md(
self,
player_name: str,
@@ -331,20 +620,27 @@ See: [prompt_{num}.json](prompts/prompt_{num}.json)
thinking = parsed.get("internal_thinking", "N/A")
note = parsed.get("note_to_self", "")
say = parsed.get("say_outloud", "")
- action = parsed.get("action", {})
+
+ # Handle both action formats: old (action object) and new (action_type + parameters)
+ action_type = parsed.get("action_type") or (parsed.get("action", {}).get("type") if parsed.get("action") else None)
+ action_params = parsed.get("parameters") or (parsed.get("action", {}).get("parameters") if parsed.get("action") else None)
action_str = ""
- if action:
- action_str = f"**{action.get('type', 'unknown')}**"
- params = action.get("parameters", {})
- if params:
- action_str += f" - {json.dumps(params)}"
+ if action_type:
+ action_str = f"**{action_type}**"
+ if action_params:
+ action_str += f" - {json.dumps(action_params)}"
+
+ # Calculate cost
+ input_cost = (response.prompt_tokens / 1_000_000) * GEMINI_FLASH_INPUT_PRICE
+ output_cost = (response.completion_tokens / 1_000_000) * GEMINI_FLASH_OUTPUT_PRICE
+ total_cost = input_cost + output_cost
section = f"""### Response Received โ
-**Latency:** {response.latency_seconds:.2f}s | **Tokens:** {response.total_tokens}
+**Latency:** {response.latency_seconds:.2f}s | **Tokens:** {response.total_tokens} (in:{response.prompt_tokens} out:{response.completion_tokens}) | **Cost:** ${total_cost:.6f}
-**Thinking:** {thinking[:200]}{'...' if len(thinking) > 200 else ''}
+**Thinking:** {thinking}
"""
if note:
@@ -485,3 +781,76 @@ See: [prompt_{num}.json](prompts/prompt_{num}.json)
def get_player_log_path(self, player_name: str) -> Path:
"""Get the MD log path for a player."""
return self.session_dir / player_name / f"{player_name}.md"
+
+ def log_tool_execution(self, batch) -> None:
+ """
+ Log tool execution batch with detailed information.
+
+ Args:
+ batch: ToolExecutionBatch object with tool call results
+ """
+ # Log to communication file
+ self.log_llm_communication(
+ f"=== Tool Execution Batch ({batch.total_calls} calls) ===",
+ "TOOL"
+ )
+
+ for call in batch.tool_calls:
+ status = "โ
" if call.success else "โ"
+
+ # Extract reasoning if present
+ reasoning = call.parameters.get("reasoning", "")
+ params_without_reasoning = {k: v for k, v in call.parameters.items() if k != "reasoning"}
+
+ self.log_llm_communication(
+ f" {status} {call.name}({json.dumps(params_without_reasoning)})",
+ "TOOL"
+ )
+
+ if reasoning:
+ self.log_llm_communication(
+ f" ๐ญ Reasoning: {reasoning}",
+ "TOOL"
+ )
+
+ self.log_llm_communication(
+ f" Time: {call.execution_time*1000:.1f}ms | "
+ f"Tokens: {call.input_tokens} in + {call.output_tokens} out = {call.input_tokens + call.output_tokens} total",
+ "TOOL"
+ )
+
+ if call.success:
+ # Log abbreviated result (without reasoning field)
+ result = call.result.copy() if isinstance(call.result, dict) else call.result
+ if isinstance(result, dict) and "llm_reasoning" in result:
+ result.pop("llm_reasoning")
+ result_str = json.dumps(result, ensure_ascii=False)
+ if len(result_str) > 200:
+ result_str = result_str[:200] + "..."
+ self.log_llm_communication(f" Result: {result_str}", "TOOL")
+ else:
+ self.log_llm_communication(f" Error: {call.error}", "TOOL")
+
+ # Summary
+ self.log_llm_communication(
+ f" Total: {batch.success_count}/{batch.total_calls} successful | "
+ f"{batch.total_tokens} tokens | {batch.total_time*1000:.1f}ms",
+ "TOOL"
+ )
+ self.log_llm_communication("=" * 50, "TOOL")
+
+ # Save detailed tool execution to separate file
+ tool_log_file = self.session_dir / "tool_executions.json"
+
+ # Load existing logs
+ tool_logs = []
+ if tool_log_file.exists():
+ with open(tool_log_file, 'r', encoding='utf-8') as f:
+ tool_logs = json.load(f)
+
+ # Add new batch
+ tool_logs.append(batch.to_dict())
+
+ # Save updated logs
+ with open(tool_log_file, 'w', encoding='utf-8') as f:
+ json.dump(tool_logs, f, indent=2, ensure_ascii=False)
diff --git a/pycatan/ai/ai_manager.py b/pycatan/ai/ai_manager.py
index e46ba02375fede99fd0a7774ee6396d3f15651e9..fa456ff5d11175653696cbbb44335d4623dc2375 100644
--- a/pycatan/ai/ai_manager.py
+++ b/pycatan/ai/ai_manager.py
@@ -23,6 +23,8 @@ from pycatan.ai.response_parser import ResponseParser, ParseResult
from pycatan.ai.schemas import ResponseType, ACTIVE_TURN_RESPONSE_SCHEMA, OBSERVING_RESPONSE_SCHEMA
from pycatan.ai.agent_state import AgentState, compute_state_hash
from pycatan.ai.ai_logger import AILogger
+from pycatan.ai.agent_tools import AgentTools
+from pycatan.ai.tool_executor import ToolExecutor
from pycatan.management.actions import Action, ActionType
@@ -73,6 +75,10 @@ class AIManager:
self.response_parser = ResponseParser()
self.logger = AILogger(session_dir=session_dir)
+ # Agent tools and executor
+ self.agent_tools = AgentTools()
+ self.tool_executor = ToolExecutor(self.agent_tools)
+
# LLM client (created lazily when needed)
self._llm_client: Optional[GeminiClient] = None
@@ -202,6 +208,9 @@ class AIManager:
self._current_game_state = game_state
self._current_allowed_actions = allowed_actions
+ # Update agent tools with current game state
+ self.agent_tools.update_game_state(game_state)
+
# Build "what happened" from recent events
what_happened = self._build_what_happened(agent)
@@ -214,14 +223,18 @@ class AIManager:
is_active_turn=True
)
- # Log the prompt
+ # Get tool schemas for logging
+ tool_schemas = self.agent_tools.get_tools_schema()
+
+ # Log the prompt (with tools schema)
log_info = self.logger.log_prompt(
player_name=player_name,
prompt=prompt,
schema=schema,
is_active=True,
what_happened=what_happened,
- allowed_actions=self._format_allowed_actions(allowed_actions)
+ allowed_actions=self._format_allowed_actions(allowed_actions),
+ tools_schema=tool_schemas
)
# Mark request sent
@@ -234,7 +247,11 @@ class AIManager:
if self.send_to_llm:
try:
self.logger.log_llm_communication(f"Sending prompt #{log_info['number']} for {player_name}", "SEND")
- response = self._send_to_llm(prompt, schema, ResponseType.ACTIVE_TURN)
+ response = self._send_to_llm(
+ prompt, schema, ResponseType.ACTIVE_TURN,
+ player_name=player_name,
+ prompt_number=log_info["number"]
+ )
if response and response.success and response.content:
self.logger.log_llm_communication(f"Received response for {player_name} ({response.total_tokens} tokens)", "RECV")
@@ -245,10 +262,7 @@ class AIManager:
if llm_suggestion:
action = llm_suggestion.get("action_type", "unknown")
params = llm_suggestion.get("parameters", {})
- thinking = llm_suggestion.get("internal_thinking", "")[:100]
self.logger.log_llm_communication(f"LLM suggests: {action} {params}", "RECV")
- if thinking:
- self.logger.log_llm_communication(f" Thinking: {thinking}...", "RECV")
else:
error_msg = response.error if response else "No response"
self.logger.log_llm_communication(f"LLM error: {error_msg}", "ERROR")
@@ -687,27 +701,232 @@ class AIManager:
self,
prompt: Dict[str, Any],
schema: Dict[str, Any],
- response_type: ResponseType
+ response_type: ResponseType,
+ player_name: str = "unknown",
+ prompt_number: int = 0
) -> LLMResponse:
- """Send prompt to LLM and get response."""
+ """
+ Send prompt to LLM and get response.
+
+ Handles tool calling loop:
+ 1. Send prompt with tools
+ 2. If LLM requests tools, execute them
+ 3. Send results back to LLM
+ 4. Repeat until final answer
+
+ Args:
+ prompt: The prompt dictionary
+ schema: Response schema
+ response_type: Type of response expected
+ player_name: Name of the player (for logging)
+ prompt_number: The prompt number (for logging)
+ """
# Convert prompt to string
prompt_str = json.dumps(prompt, indent=2, ensure_ascii=False)
- # Send to LLM with schema and thinking mode (if enabled)
+ # Get tool schemas
+ tool_schemas = self.agent_tools.get_tools_schema()
+
+ # Build generation kwargs
kwargs = {
"response_schema": schema,
+ "tools": tool_schemas, # Enable function calling
"enable_thinking": self.config.llm.enable_thinking,
"max_tokens": self.config.llm.max_tokens,
}
- if self.config.llm.enable_thinking:
- kwargs["thinking_budget"] = self.config.llm.thinking_budget
-
- response = self.llm_client.generate(
- prompt_str,
- **kwargs
+ # Determine thinking budgets and max iterations
+ if self.config.llm.thinking_budgets:
+ # Use dynamic budgets from config
+ thinking_budgets = self.config.llm.thinking_budgets
+ max_tool_iterations = len(thinking_budgets)
+ else:
+ # Fallback to single budget, default 3 iterations
+ thinking_budgets = [self.config.llm.thinking_budget] * 3
+ max_tool_iterations = 3
+
+ # Tool calling loop
+ iteration = 0
+ conversation_context = prompt_str
+
+ # Accumulated tokens across all iterations
+ accumulated_prompt_tokens = 0
+ accumulated_completion_tokens = 0
+ accumulated_thinking_tokens = 0
+ accumulated_tool_tokens = 0
+ final_response = None
+
+ while iteration < max_tool_iterations:
+ iteration += 1
+ is_tool_followup = iteration > 1
+
+ # Set thinking budget for this iteration
+ if self.config.llm.enable_thinking:
+ current_budget = thinking_budgets[iteration - 1] if iteration <= len(thinking_budgets) else thinking_budgets[-1]
+ kwargs["thinking_budget"] = current_budget
+ self.logger.log_llm_communication(
+ f"๐ญ Thinking budget for iteration {iteration}: {current_budget} tokens",
+ "INFO"
+ )
+
+ # Log API call start with running index
+ current_tools = kwargs.get("tools", [])
+ api_call_id = self.logger.log_api_call_start(
+ player_name=player_name,
+ prompt_number=prompt_number,
+ iteration=iteration,
+ tools_schema=current_tools if current_tools else None,
+ is_tool_followup=is_tool_followup
+ )
+
+ # Send request to LLM
+ response = self.llm_client.generate(
+ conversation_context,
+ **kwargs
+ )
+
+ # Accumulate tokens from this iteration
+ accumulated_prompt_tokens += response.prompt_tokens
+ accumulated_completion_tokens += response.completion_tokens
+ accumulated_thinking_tokens += response.thinking_tokens
+
+ # Log API call end
+ self.logger.log_api_call_end(
+ call_id=api_call_id,
+ success=response.success,
+ tokens=response.total_tokens,
+ prompt_tokens=response.prompt_tokens,
+ completion_tokens=response.completion_tokens,
+ has_tool_calls=bool(response.tool_calls),
+ tool_calls_count=len(response.tool_calls) if response.tool_calls else 0,
+ error=response.error
+ )
+
+ if not response.success:
+ # Update response with accumulated tokens before returning
+ response.prompt_tokens = accumulated_prompt_tokens
+ response.completion_tokens = accumulated_completion_tokens
+ response.thinking_tokens = accumulated_thinking_tokens
+ response.total_tokens = accumulated_prompt_tokens + accumulated_completion_tokens + accumulated_thinking_tokens + accumulated_tool_tokens
+ return response
+
+ # Check if LLM requested tools
+ if response.tool_calls:
+ self.logger.log_llm_communication(
+ f"๐ง LLM requested {len(response.tool_calls)} tool(s) (iteration {iteration})",
+ "TOOL_REQUEST"
+ )
+
+ # Execute tools
+ batch = self.tool_executor.execute_tool_calls(response.tool_calls)
+
+ # Log tool execution
+ self.logger.log_tool_execution(batch)
+
+ # Add tool tokens to accumulated count
+ accumulated_tool_tokens += batch.total_tokens
+
+ # Add tool tokens to stats
+ self.llm_client.stats.add_tool_tokens(batch.total_tokens)
+
+ # Format results for LLM
+ tool_results = self.tool_executor.format_tool_results_for_llm(batch)
+
+ # Add tool results to conversation
+ conversation_context = f"{conversation_context}\n\n{tool_results}\n\nNow provide your final answer based on the tool results:"
+
+ # Check if this is the last iteration
+ if iteration >= max_tool_iterations:
+ # Remove tools and send ONE FINAL request for structured answer
+ kwargs["tools"] = []
+ self.logger.log_llm_communication(
+ f"๐ Tools disabled - sending final request for structured answer",
+ "INFO"
+ )
+
+ # Log the final prompt
+ self.logger.log_tool_followup_prompt(
+ player_name=player_name,
+ original_prompt_number=prompt_number,
+ iteration=iteration + 1,
+ conversation_context=conversation_context,
+ tool_results=tool_results,
+ tools_schema=None, # No tools
+ schema=schema
+ )
+
+ # Send final request (without tools)
+ final_api_call_id = self.logger.log_api_call_start(
+ player_name=player_name,
+ prompt_number=prompt_number,
+ iteration=iteration + 1,
+ tools_schema=None,
+ is_tool_followup=True
+ )
+
+ final_response = self.llm_client.generate(
+ conversation_context,
+ **kwargs
+ )
+
+ # Accumulate tokens from final request
+ accumulated_prompt_tokens += final_response.prompt_tokens
+ accumulated_completion_tokens += final_response.completion_tokens
+ accumulated_thinking_tokens += final_response.thinking_tokens
+
+ self.logger.log_api_call_end(
+ call_id=final_api_call_id,
+ success=final_response.success,
+ tokens=final_response.total_tokens,
+ prompt_tokens=final_response.prompt_tokens,
+ completion_tokens=final_response.completion_tokens,
+ has_tool_calls=bool(final_response.tool_calls),
+ tool_calls_count=0,
+ error=final_response.error
+ )
+
+ # Update final response with accumulated tokens
+ final_response.prompt_tokens = accumulated_prompt_tokens
+ final_response.completion_tokens = accumulated_completion_tokens
+ final_response.thinking_tokens = accumulated_thinking_tokens
+ final_response.total_tokens = accumulated_prompt_tokens + accumulated_completion_tokens + accumulated_thinking_tokens + accumulated_tool_tokens
+
+ return final_response
+ else:
+ # More iterations available - continue the loop
+ next_tools = tool_schemas
+
+ # Log the tool follow-up prompt that will be sent
+ self.logger.log_tool_followup_prompt(
+ player_name=player_name,
+ original_prompt_number=prompt_number,
+ iteration=iteration + 1,
+ conversation_context=conversation_context,
+ tool_results=tool_results,
+ tools_schema=next_tools,
+ schema=schema
+ )
+
+ self.logger.log_llm_communication(
+ f"โ
Tool results sent back to LLM ({batch.total_tokens} tokens)",
+ "TOOL_RESULTS"
+ )
+
+ else:
+ # No tool calls - this is the final structured answer
+ # Gemini 3 supports tools + JSON schema together, so response is already structured
+ # Update response with accumulated tokens before returning
+ response.prompt_tokens = accumulated_prompt_tokens
+ response.completion_tokens = accumulated_completion_tokens
+ response.thinking_tokens = accumulated_thinking_tokens
+ response.total_tokens = accumulated_prompt_tokens + accumulated_completion_tokens + accumulated_thinking_tokens + accumulated_tool_tokens
+ return response
+
+ # Should not reach here normally, but return last response as fallback
+ self.logger.log_llm_communication(
+ f"โ ๏ธ Loop ended unexpectedly after {iteration} iterations",
+ "WARNING"
)
-
return response
def _parse_response(
diff --git a/pycatan/ai/config.py b/pycatan/ai/config.py
index 598a63a78bffeaf6d5f744309d067397a5a3466b..bbc138d70630fc768e5140cec64e6db54dec590d 100644
--- a/pycatan/ai/config.py
+++ b/pycatan/ai/config.py
@@ -39,7 +39,7 @@ class LLMConfig:
# Provider settings
provider: str = "gemini" # "gemini", "openai", "anthropic", "azure"
- model_name: str = "gemini-2.5-flash" # Current stable model (Jan 2026)
+ model_name: str = "gemini-3-flash-preview" # Gemini 3 supports tools + JSON together
# Generation parameters
temperature: float = 0.7
@@ -51,7 +51,14 @@ class LLMConfig:
# Uses new google-genai SDK
# Note: Requires gemini-2.0-flash-thinking-exp model
enable_thinking: bool = False # Disabled - thinking tokens eat into output budget
- thinking_budget: int = 3000 # Max tokens for thinking (3k default)
+ thinking_budget: int = 3000 # Max tokens for thinking (3k default) - used if thinking_budgets is empty
+
+ # Dynamic thinking budgets per tool iteration
+ # Each element = budget for that iteration (1st, 2nd, 3rd, etc.)
+ # List length determines max tool iterations
+ # Last iteration automatically disables tools
+ # Example: [8000, 4000, 2000] = 3 iterations, tools disabled on iteration 3
+ thinking_budgets: list = field(default_factory=lambda: [])
# API settings
api_key_env_var: str = "GEMINI_API_KEY" # Environment variable name
diff --git a/pycatan/ai/config_example.yaml b/pycatan/ai/config_example.yaml
index 2cb6aa4d976ba7a99b3829af41df67777e241446..8c1bb2a9faeaf54d4cb89101a6975df140111cb0 100644
--- a/pycatan/ai/config_example.yaml
+++ b/pycatan/ai/config_example.yaml
@@ -36,7 +36,17 @@ llm:
# Enables the model to "think" before responding
# Use with: gemini-2.0-flash-thinking-exp
enable_thinking: false # Enable thinking mode
- thinking_budget: 16000 # Max tokens for thinking (8000-32000)
+ thinking_budget: 16000 # Max tokens for thinking (fallback if thinking_budgets is empty)
+
+ # Dynamic thinking budgets per tool iteration
+ # Each element = budget for that tool call iteration
+ # List length determines max tool iterations
+ # Last iteration automatically disables tools (forces final answer)
+ # Examples:
+ # [8000, 4000, 2000] = 3 iterations, tools disabled on iteration 3
+ # [10000, 5000, 3000, 1000] = 4 iterations, tools disabled on iteration 4
+ # [] = use thinking_budget for all iterations (default 3 iterations)
+ thinking_budgets: []
# API settings
# The API key is read from environment variables for security
diff --git a/pycatan/ai/llm_client.py b/pycatan/ai/llm_client.py
index 7447eb4682c096c1bd9ddf4168f834e872bc4800..6bad9f3fc975280c865462f92e766d4eefe6b41b 100644
--- a/pycatan/ai/llm_client.py
+++ b/pycatan/ai/llm_client.py
@@ -24,6 +24,7 @@ class LLMResponse:
content: Optional[str] = None
raw_response: Optional[Any] = None
error: Optional[str] = None
+ tool_calls: List[Dict[str, Any]] = field(default_factory=list) # Function calls from LLM
# Metadata
model: str = ""
@@ -44,7 +45,7 @@ class LLMResponse:
if self.thinking_tokens > 0:
tokens_dict["thinking"] = self.thinking_tokens
- return {
+ result = {
"success": self.success,
"content": self.content[:200] + "..." if self.content and len(self.content) > 200 else self.content,
"error": self.error,
@@ -53,6 +54,12 @@ class LLMResponse:
"latency_seconds": round(self.latency_seconds, 2),
"timestamp": self.timestamp
}
+
+ # Add tool calls if present
+ if self.tool_calls:
+ result["tool_calls"] = self.tool_calls
+
+ return result
@dataclass
@@ -62,6 +69,7 @@ class LLMStats:
successful_requests: int = 0
failed_requests: int = 0
total_tokens: int = 0
+ tool_tokens: int = 0 # Tokens from tool inputs/outputs
total_cost_usd: float = 0.0
total_latency: float = 0.0
@@ -76,6 +84,11 @@ class LLMStats:
self.total_cost_usd += cost
self.total_latency += response.latency_seconds
+ def add_tool_tokens(self, tokens: int):
+ """Add tokens from tool execution."""
+ self.tool_tokens += tokens
+ self.total_tokens += tokens
+
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return {
@@ -84,6 +97,8 @@ class LLMStats:
"failed": self.failed_requests,
"success_rate": f"{self.successful_requests / self.total_requests * 100:.1f}%" if self.total_requests > 0 else "0%",
"total_tokens": self.total_tokens,
+ "tool_tokens": self.tool_tokens,
+ "llm_tokens": self.total_tokens - self.tool_tokens,
"total_cost_usd": f"${self.total_cost_usd:.4f}",
"avg_latency": f"{self.total_latency / self.total_requests:.2f}s" if self.total_requests > 0 else "0s"
}
@@ -167,9 +182,10 @@ class GeminiClient(LLMClient):
- response_schema: JSON schema to enforce structure
- enable_thinking: Enable thinking mode
- thinking_budget: Max tokens for thinking
+ - tools: List of tool schemas for function calling
Returns:
- LLMResponse object with result
+ LLMResponse object with result (may include tool_calls)
"""
start_time = time.time()
@@ -189,13 +205,41 @@ class GeminiClient(LLMClient):
)
logger.info(f"Thinking mode enabled with budget: {thinking_budget}")
+ # Add tools for function calling if provided
+ # NOTE: Must be done BEFORE setting response format (they're mutually exclusive!)
+ tools = kwargs.get("tools", [])
+ has_tools = False
+ if tools:
+ # Convert tool schemas to Gemini Tool objects
+ try:
+ gemini_tools = []
+ for tool_dict in tools:
+ # Create FunctionDeclaration for each tool
+ func_decl = self.types.FunctionDeclaration(
+ name=tool_dict.get("name", ""),
+ description=tool_dict.get("description", ""),
+ parameters=self._remove_unsupported_fields(tool_dict.get("parameters", {}))
+ )
+ gemini_tools.append(func_decl)
+
+ # Wrap in Tool object
+ config_dict["tools"] = [self.types.Tool(function_declarations=gemini_tools)]
+ has_tools = True
+ logger.info(f"Function calling enabled with {len(tools)} tool(s)")
+ except Exception as e:
+ logger.warning(f"Failed to create tool declarations: {e}")
+ logger.warning("Tools will be disabled.")
+ # Continue without tools if conversion fails
+ pass
+
# Set response format
+ # NOTE: Gemini 3 supports tools + JSON schema together!
+ # For older models (Gemini 2.x), this was incompatible
response_format = kwargs.get("response_format", self.response_format)
if response_format == "json":
config_dict["response_mime_type"] = "application/json"
# Add response_json_schema if provided (enforces structure)
- # Note: new SDK uses response_json_schema instead of response_schema
if "response_schema" in kwargs:
schema = kwargs["response_schema"]
cleaned_schema = self._remove_unsupported_fields(schema)
@@ -217,8 +261,31 @@ class GeminiClient(LLMClient):
latency = time.time() - start_time
- # Extract content
- content = response.text
+ # Extract content and tool calls
+ content = response.text if hasattr(response, 'text') else ""
+ tool_calls = []
+
+ # Check for function calls in response
+ if hasattr(response, 'candidates') and response.candidates:
+ candidate = response.candidates[0]
+ if hasattr(candidate, 'content') and hasattr(candidate.content, 'parts'):
+ for part in candidate.content.parts:
+ if hasattr(part, 'function_call'):
+ # Extract function call
+ func_call = part.function_call
+ # Guard: func_call or func_call.name can be None sometimes
+ if func_call is None:
+ logger.warning("Skipping None function_call in response")
+ continue
+ func_name = getattr(func_call, 'name', None)
+ if func_name is None:
+ logger.warning(f"Skipping function_call with None name: {func_call}")
+ continue
+ tool_calls.append({
+ "id": f"call_{len(tool_calls)+1}",
+ "name": func_name,
+ "parameters": dict(func_call.args) if hasattr(func_call, 'args') else {}
+ })
# Token counting from usage metadata
thinking_tokens = 0
@@ -254,6 +321,7 @@ class GeminiClient(LLMClient):
success=True,
content=content,
raw_response=response,
+ tool_calls=tool_calls,
model=self.model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
@@ -264,11 +332,17 @@ class GeminiClient(LLMClient):
self.stats.add_request(llm_response, cost)
- if thinking_tokens > 0:
+ if tool_calls:
+ logger.info(f"โ
Response with {len(tool_calls)} tool call(s): {completion_tokens} tokens, {latency:.2f}s")
+ for tc in tool_calls:
+ logger.info(f" ๐ง {tc['name']}({tc['parameters']})")
+ elif thinking_tokens > 0:
logger.info(f"โ
Response received: {completion_tokens} tokens (+{thinking_tokens} thinking), {latency:.2f}s")
else:
logger.info(f"โ
Response received: {completion_tokens} tokens, {latency:.2f}s")
- logger.debug(f"Response preview: {content[:100]}...")
+
+ if content:
+ logger.debug(f"Response preview: {content[:100]}...")
return llm_response
@@ -298,6 +372,8 @@ class GeminiClient(LLMClient):
"""
return len(text) // 4
+ # Note: _convert_tool_schema removed - using FunctionDeclaration directly now
+
def _remove_unsupported_fields(self, schema: Any) -> Any:
"""
Remove fields that are not supported by the Python SDK.
diff --git a/pycatan/ai/tool_executor.py b/pycatan/ai/tool_executor.py
new file mode 100644
index 0000000000000000000000000000000000000000..e3ae3d9ebd42e7a9651fef1865b1a0d204720c88
--- /dev/null
+++ b/pycatan/ai/tool_executor.py
@@ -0,0 +1,319 @@
+"""
+Tool Executor - Manages LLM Tool/Function Calling
+
+This module handles:
+1. Tool execution with proper logging
+2. Token counting for tool inputs/outputs
+3. Support for multiple tool calls
+4. Detailed execution traces
+
+The ToolExecutor bridges between LLM responses with tool calls
+and the actual AgentTools implementation.
+"""
+
+import json
+import time
+import logging
+from typing import Dict, Any, List, Optional
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from pycatan.ai.agent_tools import AgentTools
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ToolCall:
+ """Represents a single tool call from LLM."""
+ id: str # Unique ID for this call (from LLM response)
+ name: str # Tool name (e.g., "inspect_node")
+ parameters: Dict[str, Any] # Tool parameters
+
+ # Execution results (filled after execution)
+ result: Optional[Dict[str, Any]] = None
+ success: bool = False
+ error: Optional[str] = None
+ execution_time: float = 0.0
+
+ # Token tracking
+ input_tokens: int = 0
+ output_tokens: int = 0
+
+ def to_dict(self) -> Dict[str, Any]:
+ """Convert to dictionary for logging."""
+ return {
+ "id": self.id,
+ "name": self.name,
+ "parameters": self.parameters,
+ "result": self.result,
+ "success": self.success,
+ "error": self.error,
+ "execution_time_ms": round(self.execution_time * 1000, 2),
+ "tokens": {
+ "input": self.input_tokens,
+ "output": self.output_tokens,
+ "total": self.input_tokens + self.output_tokens
+ }
+ }
+
+
+@dataclass
+class ToolExecutionBatch:
+ """Represents a batch of tool calls (can be multiple in one turn)."""
+ timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
+ tool_calls: List[ToolCall] = field(default_factory=list)
+ total_time: float = 0.0
+
+ @property
+ def total_input_tokens(self) -> int:
+ """Sum of all input tokens."""
+ return sum(call.input_tokens for call in self.tool_calls)
+
+ @property
+ def total_output_tokens(self) -> int:
+ """Sum of all output tokens."""
+ return sum(call.output_tokens for call in self.tool_calls)
+
+ @property
+ def total_tokens(self) -> int:
+ """Sum of all tokens."""
+ return self.total_input_tokens + self.total_output_tokens
+
+ @property
+ def success_count(self) -> int:
+ """Number of successful calls."""
+ return sum(1 for call in self.tool_calls if call.success)
+
+ @property
+ def failure_count(self) -> int:
+ """Number of failed calls."""
+ return sum(1 for call in self.tool_calls if not call.success)
+
+ @property
+ def total_calls(self) -> int:
+ """Total number of tool calls in this batch."""
+ return len(self.tool_calls)
+
+ def to_dict(self) -> Dict[str, Any]:
+ """Convert to dictionary for logging."""
+ return {
+ "timestamp": self.timestamp,
+ "total_calls": len(self.tool_calls),
+ "successful": self.success_count,
+ "failed": self.failure_count,
+ "total_time_ms": round(self.total_time * 1000, 2),
+ "tokens": {
+ "input": self.total_input_tokens,
+ "output": self.total_output_tokens,
+ "total": self.total_tokens
+ },
+ "calls": [call.to_dict() for call in self.tool_calls]
+ }
+
+
+class ToolExecutor:
+ """
+ Executor for AI agent tools with full logging and token tracking.
+
+ Features:
+ - Executes tool calls from LLM responses
+ - Supports multiple tool calls in parallel
+ - Tracks tokens for input (parameters) and output (results)
+ - Logs detailed execution traces
+ - Handles errors gracefully
+ """
+
+ def __init__(self, agent_tools: AgentTools):
+ """
+ Initialize the tool executor.
+
+ Args:
+ agent_tools: AgentTools instance with game state
+ """
+ self.agent_tools = agent_tools
+ self.execution_history: List[ToolExecutionBatch] = []
+
+ def execute_tool_calls(
+ self,
+ tool_calls: List[Dict[str, Any]],
+ call_id_prefix: str = "call"
+ ) -> ToolExecutionBatch:
+ """
+ Execute a batch of tool calls.
+
+ Args:
+ tool_calls: List of tool call dictionaries from LLM
+ Each should have: {id, name, parameters}
+ call_id_prefix: Prefix for generating call IDs if missing
+
+ Returns:
+ ToolExecutionBatch with all results
+ """
+ batch_start = time.time()
+ batch = ToolExecutionBatch()
+
+ logger.info(f"๐ง Executing {len(tool_calls)} tool call(s)...")
+
+ for idx, tool_call_data in enumerate(tool_calls):
+ # Parse tool call
+ call_id = tool_call_data.get("id", f"{call_id_prefix}_{idx+1}")
+ tool_name = tool_call_data.get("name", tool_call_data.get("function", ""))
+ parameters = tool_call_data.get("parameters", tool_call_data.get("arguments", {}))
+
+ # Handle parameters as JSON string
+ if isinstance(parameters, str):
+ try:
+ parameters = json.loads(parameters)
+ except json.JSONDecodeError:
+ parameters = {}
+
+ tool_call = ToolCall(
+ id=call_id,
+ name=tool_name,
+ parameters=parameters
+ )
+
+ # Execute single tool
+ self._execute_single_tool(tool_call)
+
+ batch.tool_calls.append(tool_call)
+
+ batch.total_time = time.time() - batch_start
+
+ # Log summary
+ logger.info(
+ f"โ
Tool execution complete: {batch.success_count}/{len(tool_calls)} successful, "
+ f"{batch.total_tokens} tokens, {batch.total_time:.2f}s"
+ )
+
+ # Add to history
+ self.execution_history.append(batch)
+
+ return batch
+
+ def _execute_single_tool(self, tool_call: ToolCall) -> None:
+ """
+ Execute a single tool call and populate results.
+
+ Args:
+ tool_call: ToolCall object to execute (modified in place)
+ """
+ start_time = time.time()
+
+ logger.info(f" ๐ง {tool_call.name}({tool_call.parameters})")
+
+ try:
+ # Count input tokens (parameters as JSON)
+ param_json = json.dumps(tool_call.parameters)
+ tool_call.input_tokens = self._estimate_tokens(param_json)
+
+ # Execute the tool
+ result = self.agent_tools.execute_tool(
+ tool_name=tool_call.name,
+ parameters=tool_call.parameters
+ )
+
+ # Count output tokens (result as JSON)
+ result_json = json.dumps(result)
+ tool_call.output_tokens = self._estimate_tokens(result_json)
+
+ # Store result
+ tool_call.result = result
+ tool_call.success = True
+ tool_call.execution_time = time.time() - start_time
+
+ logger.info(
+ f" โ Success: {tool_call.output_tokens} tokens, "
+ f"{tool_call.execution_time*1000:.1f}ms"
+ )
+ logger.debug(f" Result preview: {str(result)[:100]}...")
+
+ except Exception as e:
+ tool_call.success = False
+ tool_call.error = str(e)
+ tool_call.execution_time = time.time() - start_time
+
+ logger.error(f" โ Failed: {e}")
+
+ def format_tool_results_for_llm(self, batch: ToolExecutionBatch) -> str:
+ """
+ Format tool results for sending back to LLM.
+
+ Creates a formatted text response that includes:
+ - Each tool call with its result
+ - Clear separation between calls
+ - Error handling
+
+ Args:
+ batch: ToolExecutionBatch with executed calls
+
+ Returns:
+ Formatted string for LLM context
+ """
+ lines = ["=== Tool Results ===\n"]
+
+ for call in batch.tool_calls:
+ lines.append(f"Tool: {call.name}")
+ lines.append(f"Parameters: {json.dumps(call.parameters, indent=2)}")
+
+ if call.success:
+ lines.append(f"Result:")
+ lines.append(json.dumps(call.result, indent=2))
+ else:
+ lines.append(f"Error: {call.error}")
+
+ lines.append("---\n")
+
+ return "\n".join(lines)
+
+ def get_execution_summary(self) -> Dict[str, Any]:
+ """
+ Get summary statistics of all tool executions.
+
+ Returns:
+ Dictionary with summary stats
+ """
+ if not self.execution_history:
+ return {
+ "total_batches": 0,
+ "total_calls": 0,
+ "total_tokens": 0
+ }
+
+ total_calls = sum(len(batch.tool_calls) for batch in self.execution_history)
+ total_tokens = sum(batch.total_tokens for batch in self.execution_history)
+ successful_calls = sum(batch.success_count for batch in self.execution_history)
+ failed_calls = sum(batch.failure_count for batch in self.execution_history)
+
+ # Count by tool name
+ tool_usage: Dict[str, int] = {}
+ for batch in self.execution_history:
+ for call in batch.tool_calls:
+ tool_usage[call.name] = tool_usage.get(call.name, 0) + 1
+
+ return {
+ "total_batches": len(self.execution_history),
+ "total_calls": total_calls,
+ "successful_calls": successful_calls,
+ "failed_calls": failed_calls,
+ "success_rate": f"{successful_calls / total_calls * 100:.1f}%" if total_calls > 0 else "0%",
+ "total_tokens": total_tokens,
+ "tool_usage": tool_usage,
+ "recent_batches": [batch.to_dict() for batch in self.execution_history[-5:]]
+ }
+
+ def _estimate_tokens(self, text: str) -> int:
+ """
+ Estimate token count for text.
+
+ Uses rough approximation: 1 token โ 4 characters.
+ This is the same method used in llm_client.py for consistency.
+
+ Args:
+ text: Text to count tokens for
+
+ Returns:
+ Estimated token count
+ """
+ return len(text) // 4