track token usage for each tool call in ChatAgent

.gitignore

@@ -465,3 +465,4 @@ yarn-error.log*
 
 # Yarn Integrity file
 .yarn-integrity
+.env

camel/agents/chat_agent.py

@@ -2161,6 +2161,9 @@ def _step_impl(
 
             if tool_call_requests := response.tool_call_requests:
                 # Process all tool calls
+                # All tools in this batch share the same token usage from the
+                # LLM call that generated them
+                current_token_usage = response.usage_dict
                 for tool_call_request in tool_call_requests:
                     if (
                         tool_call_request.tool_name
@@ -2179,7 +2182,9 @@ def _step_impl(
                             else:
                                 while not self.pause_event.is_set():
                                     time.sleep(0.001)
-                        result = self._execute_tool(tool_call_request)
+                        result = self._execute_tool(
+                            tool_call_request, token_usage=current_token_usage
+                        )
                         tool_call_records.append(result)
 
                 # If we found external tool calls, break the loop
@@ -2380,6 +2385,9 @@ async def _astep_non_streaming_task(
 
             if tool_call_requests := response.tool_call_requests:
                 # Process all tool calls
+                # All tools in this batch share the same token usage from the
+                # LLM call that generated them
+                current_token_usage = response.usage_dict
                 for tool_call_request in tool_call_requests:
                     if (
                         tool_call_request.tool_name
@@ -2401,7 +2409,7 @@ async def _astep_non_streaming_task(
                                     None, self.pause_event.wait
                                 )
                         tool_call_record = await self._aexecute_tool(
-                            tool_call_request
+                            tool_call_request, token_usage=current_token_usage
                         )
                         tool_call_records.append(tool_call_record)
 
@@ -2952,11 +2960,15 @@ def _step_terminate(
     def _execute_tool(
         self,
         tool_call_request: ToolCallRequest,
+        token_usage: Optional[Dict[str, int]] = None,
     ) -> ToolCallingRecord:
         r"""Execute the tool with arguments following the model's response.
 
         Args:
             tool_call_request (_ToolCallRequest): The tool call request.
+            token_usage (Optional[Dict[str, int]], optional): Token usage
+                information for the LLM call that generated this tool call.
+                (default: :obj:`None`)
 
         Returns:
             FunctionCallingRecord: A struct for logging information about this
@@ -2987,12 +2999,18 @@ def _execute_tool(
             logger.warning(f"{error_msg} with result: {result}")
 
         return self._record_tool_calling(
-            func_name, args, result, tool_call_id, mask_output=mask_flag
+            func_name,
+            args,
+            result,
+            tool_call_id,
+            mask_output=mask_flag,
+            token_usage=token_usage,
         )
 
     async def _aexecute_tool(
         self,
         tool_call_request: ToolCallRequest,
+        token_usage: Optional[Dict[str, int]] = None,
     ) -> ToolCallingRecord:
         func_name = tool_call_request.tool_name
         args = tool_call_request.args
@@ -3029,7 +3047,9 @@ async def _aexecute_tool(
             error_msg = f"Error executing async tool '{func_name}': {e!s}"
             result = f"Tool execution failed: {error_msg}"
             logger.warning(error_msg)
-        return self._record_tool_calling(func_name, args, result, tool_call_id)
+        return self._record_tool_calling(
+            func_name, args, result, tool_call_id, token_usage=token_usage
+        )
 
     def _record_tool_calling(
         self,
@@ -3038,6 +3058,7 @@ def _record_tool_calling(
         result: Any,
         tool_call_id: str,
         mask_output: bool = False,
+        token_usage: Optional[Dict[str, int]] = None,
     ):
         r"""Record the tool calling information in the memory, and return the
         tool calling record.
@@ -3050,6 +3071,9 @@ def _record_tool_calling(
             mask_output (bool, optional): Whether to return a sanitized
                 placeholder instead of the raw tool output.
                 (default: :obj:`False`)
+            token_usage (Optional[Dict[str, int]], optional): Token usage
+                information for the LLM call that generated this tool call.
+                (default: :obj:`None`)
 
         Returns:
             ToolCallingRecord: A struct containing information about
@@ -3099,6 +3123,7 @@ def _record_tool_calling(
             args=args,
             result=result,
             tool_call_id=tool_call_id,
+            token_usage=token_usage,
         )
 
         return tool_record

camel/types/agents/tool_calling_record.py

@@ -26,25 +26,34 @@ class ToolCallingRecord(BaseModel):
         tool_call_id (str): The ID of the tool call, if available.
         images (Optional[List[str]]): List of base64-encoded images returned
             by the tool, if any.
+        token_usage (Optional[Dict[str, int]]): Token usage information for
+            the LLM call that generated this tool call. Contains keys like
+            'prompt_tokens', 'completion_tokens', and 'total_tokens'.
+            If multiple tools are called in a single LLM response, they share
+            the same token usage dict. (default: :obj:`None`)
     """
 
     tool_name: str
     args: Dict[str, Any]
     result: Any
     tool_call_id: str
     images: Optional[List[str]] = None
+    token_usage: Optional[Dict[str, int]] = None
 
     def __str__(self) -> str:
         r"""Overridden version of the string function.
 
         Returns:
             str: Modified string to represent the tool calling.
         """
-        return (
+        base_str = (
             f"Tool Execution: {self.tool_name}\n"
             f"\tArgs: {self.args}\n"
             f"\tResult: {self.result}\n"
         )
+        if self.token_usage:
+            base_str += f"\tToken Usage: {self.token_usage}\n"
+        return base_str
 
     def as_dict(self) -> dict[str, Any]:
         r"""Returns the tool calling record as a dictionary.

examples/agents/tool_calling_with_token_tracking.py

@@ -0,0 +1,102 @@
+# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
+
+"""
+This example demonstrates how to track token usage for each individual tool
+call in a ChatAgent step.
+
+Starting from version 0.2.79, each ToolCallingRecord includes token_usage
+information, allowing you to monitor the cost of each tool call separately.
+"""
+
+from camel.agents import ChatAgent
+from camel.messages import BaseMessage
+from camel.models import ModelFactory
+from camel.toolkits import MathToolkit
+from camel.types import ModelPlatformType, ModelType
+
+# Create a ChatAgent with math tools
+model = ModelFactory.create(
+    model_platform=ModelPlatformType.OPENAI,
+    model_type=ModelType.GPT_5_MINI,
+)
+
+agent = ChatAgent(
+    system_message="You are a helpful math assistant.",
+    model=model,
+    tools=MathToolkit().get_tools(),
+)
+
+# Send a message that will trigger multiple tool calls
+user_message = BaseMessage.make_user_message(
+    role_name="User",
+    content="Calculate the result of (5 * 3) - 10",
+)
+
+print("User:", user_message.content)
+print("\n" + "=" * 60 + "\n")
+
+# Step the agent
+response = agent.step(user_message)
+
+# Display the response
+print("Assistant:", response.msgs[0].content)
+print("\n" + "=" * 60 + "\n")
+
+# Check tool calls and their token usage
+if response.info.get('tool_calls'):
+    print("Tool Calls with Token Usage:")
+    print("-" * 60)
+
+    for i, tool_call in enumerate(response.info['tool_calls'], 1):
+        print(f"\nTool Call #{i}:")
+        print(f"  Tool Name: {tool_call.tool_name}")
+        print(f"  Arguments: {tool_call.args}")
+        print(f"  Result: {tool_call.result}")
+
+        if tool_call.token_usage:
+            print("  Token Usage:")
+            prompt_tokens = tool_call.token_usage['prompt_tokens']
+            completion_tokens = tool_call.token_usage['completion_tokens']
+            total_tokens = tool_call.token_usage['total_tokens']
+            print(f"    - Prompt Tokens: {prompt_tokens}")
+            print(f"    - Completion Tokens: {completion_tokens}")
+            print(f"    - Total Tokens: {total_tokens}")
+        else:
+            print("  Token Usage: Not available")
+
+    print("\n" + "=" * 60 + "\n")
+
+    # Calculate total tokens across all tool calls
+    total_prompt = sum(
+        tc.token_usage['prompt_tokens']
+        for tc in response.info['tool_calls']
+        if tc.token_usage
+    )
+    total_completion = sum(
+        tc.token_usage['completion_tokens']
+        for tc in response.info['tool_calls']
+        if tc.token_usage
+    )
+    total_overall = sum(
+        tc.token_usage['total_tokens']
+        for tc in response.info['tool_calls']
+        if tc.token_usage
+    )
+
+    print("Summary:")
+    print(f"  Total Tool Calls: {len(response.info['tool_calls'])}")
+    print(f"  Total Prompt Tokens: {total_prompt}")
+    print(f"  Total Completion Tokens: {total_completion}")
+    print(f"  Total Tokens (all tools): {total_overall}")

test/agents/test_chat_agent.py

@@ -1427,3 +1427,53 @@ def test_memory_setter_preserves_system_message():
     assert len(new_context) > 0
     assert new_context[0]['role'] == 'system'
     assert new_context[0]['content'] == system_content
+
+
+@pytest.mark.model_backend
+def test_tool_calling_token_usage_tracking():
+    r"""Test that token usage is tracked for each tool call."""
+    system_message = BaseMessage(
+        role_name="assistant",
+        role_type=RoleType.ASSISTANT,
+        meta_dict=None,
+        content="You are a helpful assistant.",
+    )
+    model = ModelFactory.create(
+        model_platform=ModelPlatformType.OPENAI,
+        model_type=ModelType.GPT_5_MINI,
+    )
+    agent = ChatAgent(
+        system_message=system_message,
+        model=model,
+        tools=MathToolkit().get_tools(),
+    )
+
+    user_msg = BaseMessage(
+        role_name="User",
+        role_type=RoleType.USER,
+        meta_dict=dict(),
+        content="Calculate 5 + 3",
+    )
+
+    # Step the agent
+    response = agent.step(user_msg)
+
+    # Check that we have tool calls in the response
+    if response.info.get('tool_calls'):
+        tool_calls = response.info['tool_calls']
+
+        # Each tool call should have token_usage information
+        for tool_call in tool_calls:
+            assert isinstance(tool_call, ToolCallingRecord)
+            assert tool_call.token_usage is not None
+            assert isinstance(tool_call.token_usage, dict)
+
+            # Check that the standard token usage keys exist
+            assert 'prompt_tokens' in tool_call.token_usage
+            assert 'completion_tokens' in tool_call.token_usage
+            assert 'total_tokens' in tool_call.token_usage
+
+            # Verify token counts are positive integers
+            assert tool_call.token_usage['prompt_tokens'] > 0
+            assert tool_call.token_usage['completion_tokens'] >= 0
+            assert tool_call.token_usage['total_tokens'] > 0