feat(cost): track Anthropic usage metadata

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

gpt_researcher/llm_provider/generic/base.py

@@ -95,6 +95,29 @@ class GenericLLMProvider:
         self.llm = llm
         self.chat_logger = ChatLogger(chat_log) if chat_log else None
         self.verbose = verbose
+        self.last_usage_metadata: dict[str, Any] | None = None
+        self.last_response_metadata: dict[str, Any] = {}
+
+    def _reset_last_response_metadata(self) -> None:
+        self.last_usage_metadata = None
+        self.last_response_metadata = {}
+
+    def _capture_response_metadata(self, message: Any) -> None:
+        usage_metadata = getattr(message, "usage_metadata", None)
+        if usage_metadata:
+            if hasattr(usage_metadata, "model_dump"):
+                usage_metadata = usage_metadata.model_dump()
+            self.last_usage_metadata = dict(usage_metadata)
+
+        response_metadata = getattr(message, "response_metadata", None)
+        if response_metadata:
+            if hasattr(response_metadata, "model_dump"):
+                response_metadata = response_metadata.model_dump()
+            self.last_response_metadata = {
+                **self.last_response_metadata,
+                **dict(response_metadata),
+            }
+
     @classmethod
     def from_provider(cls, provider: str, chat_log: str | None = None, verbose: bool=True, **kwargs: Any):
         if provider == "openai":
@@ -282,9 +305,11 @@ class GenericLLMProvider:
 
 
     async def get_chat_response(self, messages, stream, websocket=None, **kwargs):
+        self._reset_last_response_metadata()
         if not stream:
             # Getting output from the model chain using ainvoke for asynchronous invoking
             output = await self.llm.ainvoke(messages, **kwargs)
+            self._capture_response_metadata(output)
 
             res = output.content
 
@@ -297,11 +322,13 @@ class GenericLLMProvider:
         return res
 
     async def stream_response(self, messages, websocket=None, **kwargs):
+        self._reset_last_response_metadata()
         paragraph = ""
         response = ""
 
         # Streaming the response using the chain astream method from langchain
         async for chunk in self.llm.astream(messages, **kwargs):
+            self._capture_response_metadata(chunk)
             content = chunk.content
             if not content:
                 continue

gpt_researcher/utils/costs.py

@@ -1,9 +1,10 @@
-"""Cost estimation utilities for LLM API usage.
+"""Cost estimation utilities for LLM API usage."""
 
-This module provides functions to estimate the cost of LLM API calls
-based on token counts. Cost estimates are based on OpenAI pricing
-and may vary for other model providers.
-"""
+from __future__ import annotations
+
+import logging
+from collections.abc import Mapping
+from typing import Any
 
 import tiktoken
 
@@ -14,6 +15,27 @@ OUTPUT_COST_PER_TOKEN = 0.000015
 IMAGE_INFERENCE_COST = 0.003825
 EMBEDDING_COST = 0.02 / 1000000  # Assumes new ada-3-small
 
+logger = logging.getLogger(__name__)
+
+ANTHROPIC_MODEL_PRICING = (
+    (("claude-opus-4-7",), 5.0, 25.0),
+    (("claude-opus-4-6",), 5.0, 25.0),
+    (("claude-opus-4-5", "claude-4-opus"), 5.0, 25.0),
+    (("claude-opus-4-1",), 15.0, 75.0),
+    (("claude-opus-4",), 15.0, 75.0),
+    (("claude-sonnet-4-6",), 3.0, 15.0),
+    (("claude-sonnet-4-5", "claude-4-sonnet"), 3.0, 15.0),
+    (("claude-sonnet-4",), 3.0, 15.0),
+    (("claude-haiku-4-5",), 1.0, 5.0),
+    (("claude-3-5-haiku",), 0.8, 4.0),
+)
+
+ANTHROPIC_US_INFERENCE_GEO_MODELS = (
+    "claude-opus-4-7",
+    "claude-opus-4-6",
+    "claude-sonnet-4-6",
+)
+
 
 def estimate_llm_cost(input_content: str, output_content: str) -> float:
     """Estimate the cost of an LLM API call based on input and output content.
@@ -35,6 +57,130 @@ def estimate_llm_cost(input_content: str, output_content: str) -> float:
     return input_costs + output_costs
 
 
+def _mapping_to_dict(value: Mapping[str, Any] | Any | None) -> dict[str, Any]:
+    if value is None:
+        return {}
+    if isinstance(value, Mapping):
+        return dict(value)
+    if hasattr(value, "model_dump"):
+        return dict(value.model_dump())
+    return {}
+
+
+def _resolve_anthropic_model_name(
+    model: str | None,
+    response_metadata: Mapping[str, Any] | None = None,
+) -> str:
+    metadata = _mapping_to_dict(response_metadata)
+    return str(
+        metadata.get("model")
+        or metadata.get("model_name")
+        or model
+        or ""
+    ).lower()
+
+
+def _extract_anthropic_usage(
+    response_metadata: Mapping[str, Any] | None = None,
+    usage_metadata: Mapping[str, Any] | Any | None = None,
+) -> dict[str, int] | None:
+    metadata = _mapping_to_dict(response_metadata)
+    usage = _mapping_to_dict(metadata.get("usage"))
+
+    if usage:
+        input_tokens = usage.get("input_tokens")
+        output_tokens = usage.get("output_tokens")
+        if input_tokens is not None and output_tokens is not None:
+            return {
+                "input_tokens": int(input_tokens),
+                "output_tokens": int(output_tokens),
+            }
+
+    usage = _mapping_to_dict(usage_metadata)
+    input_tokens = usage.get("input_tokens")
+    output_tokens = usage.get("output_tokens")
+    if input_tokens is None or output_tokens is None:
+        return None
+
+    return {
+        "input_tokens": int(input_tokens),
+        "output_tokens": int(output_tokens),
+    }
+
+
+def _get_anthropic_pricing(model_name: str) -> tuple[float, float] | None:
+    normalized_model_name = model_name.lower()
+    for patterns, input_price_per_mtok, output_price_per_mtok in ANTHROPIC_MODEL_PRICING:
+        if any(pattern in normalized_model_name for pattern in patterns):
+            return input_price_per_mtok, output_price_per_mtok
+    return None
+
+
+def _get_anthropic_pricing_multiplier(
+    model_name: str,
+    request_options: Mapping[str, Any] | None = None,
+) -> float:
+    if not request_options:
+        return 1.0
+
+    inference_geo = str(request_options.get("inference_geo", "")).lower()
+    if inference_geo != "us":
+        return 1.0
+
+    if any(pattern in model_name for pattern in ANTHROPIC_US_INFERENCE_GEO_MODELS):
+        return 1.1
+
+    return 1.0
+
+
+def calculate_anthropic_cost(
+    model: str | None,
+    response_metadata: Mapping[str, Any] | None = None,
+    usage_metadata: Mapping[str, Any] | Any | None = None,
+    request_options: Mapping[str, Any] | None = None,
+) -> float | None:
+    usage = _extract_anthropic_usage(response_metadata=response_metadata, usage_metadata=usage_metadata)
+    if not usage:
+        return None
+
+    model_name = _resolve_anthropic_model_name(model=model, response_metadata=response_metadata)
+    pricing = _get_anthropic_pricing(model_name)
+    if pricing is None:
+        logger.warning(
+            "Missing Anthropic pricing rule for model '%s'; falling back to token estimator.",
+            model_name or model,
+        )
+        return None
+
+    input_price_per_mtok, output_price_per_mtok = pricing
+    multiplier = _get_anthropic_pricing_multiplier(model_name, request_options=request_options)
+    input_cost = usage["input_tokens"] * input_price_per_mtok / 1_000_000
+    output_cost = usage["output_tokens"] * output_price_per_mtok / 1_000_000
+    return (input_cost + output_cost) * multiplier
+
+
+def calculate_llm_cost(
+    llm_provider: str | None,
+    model: str | None,
+    input_content: str,
+    output_content: str,
+    response_metadata: Mapping[str, Any] | None = None,
+    usage_metadata: Mapping[str, Any] | Any | None = None,
+    request_options: Mapping[str, Any] | None = None,
+) -> float:
+    if llm_provider == "anthropic":
+        anthropic_cost = calculate_anthropic_cost(
+            model=model,
+            response_metadata=response_metadata,
+            usage_metadata=usage_metadata,
+            request_options=request_options,
+        )
+        if anthropic_cost is not None:
+            return anthropic_cost
+
+    return estimate_llm_cost(input_content, output_content)
+
+
 def estimate_embedding_cost(model: str, docs: list) -> float:
     """Estimate the cost of embedding documents.
 

gpt_researcher/utils/llm.py

@@ -20,7 +20,7 @@ from gpt_researcher.llm_provider.generic.base import (
 )
 
 from ..prompts import PromptFamily
-from .costs import estimate_llm_cost
+from .costs import calculate_llm_cost
 from .validators import Subtopics
 
 
@@ -126,7 +126,15 @@ async def create_chat_completion(
             break
 
         if cost_callback:
-            llm_costs = estimate_llm_cost(str(messages), response)
+            llm_costs = calculate_llm_cost(
+                llm_provider=llm_provider,
+                model=model,
+                input_content=str(messages),
+                output_content=response,
+                response_metadata=provider.last_response_metadata,
+                usage_metadata=provider.last_usage_metadata,
+                request_options=provider_kwargs,
+            )
             cost_callback(llm_costs)
 
         return response

gpt_researcher/utils/tools.py

@@ -12,11 +12,37 @@ from typing import Any, Dict, List, Tuple, Callable, Optional
 from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
 from langchain_core.tools import tool
 
+from .costs import calculate_llm_cost
 from .llm import create_chat_completion
 
 logger = logging.getLogger(__name__)
 
 
+def _track_response_cost(
+    *,
+    llm_provider: str | None,
+    model: str | None,
+    input_payload: Any,
+    response_message: Any,
+    request_options: Dict[str, Any],
+    cost_callback: Callable | None,
+) -> None:
+    if not cost_callback:
+        return
+
+    response_content = getattr(response_message, "content", "") or ""
+    llm_costs = calculate_llm_cost(
+        llm_provider=llm_provider,
+        model=model,
+        input_content=str(input_payload),
+        output_content=str(response_content),
+        response_metadata=getattr(response_message, "response_metadata", None),
+        usage_metadata=getattr(response_message, "usage_metadata", None),
+        request_options=request_options,
+    )
+    cost_callback(llm_costs)
+
+
 async def create_chat_completion_with_tools(
     messages: List[Dict[str, str]],
     tools: List[Callable],
@@ -88,6 +114,14 @@ async def create_chat_completion_with_tools(
         
         # First call to LLM
         response = await llm_with_tools.ainvoke(lc_messages)
+        _track_response_cost(
+            llm_provider=llm_provider,
+            model=model,
+            input_payload=lc_messages,
+            response_message=response,
+            request_options=provider_kwargs,
+            cost_callback=cost_callback,
+        )
         
         # Process tool calls if any were made
         tool_calls_metadata = []
@@ -152,23 +186,21 @@ async def create_chat_completion_with_tools(
             # Get final response from LLM after tool execution
             logger.info("Getting final response from LLM after tool execution")
             final_response = await llm_with_tools.ainvoke(lc_messages)
-            
+             
             # Track costs if callback provided
-            if cost_callback:
-                from .costs import estimate_llm_cost
-                # Calculate costs for both calls
-                llm_costs = estimate_llm_cost(str(lc_messages), final_response.content or "")
-                cost_callback(llm_costs)
-            
+            _track_response_cost(
+                llm_provider=llm_provider,
+                model=model,
+                input_payload=lc_messages,
+                response_message=final_response,
+                request_options=provider_kwargs,
+                cost_callback=cost_callback,
+            )
+             
             return final_response.content, tool_calls_metadata
-        
+         
         else:
             # No tool calls, return regular response
-            if cost_callback:
-                from .costs import estimate_llm_cost
-                llm_costs = estimate_llm_cost(str(messages), response.content or "")
-                cost_callback(llm_costs)
-            
             return response.content, []
         
     except Exception as e:

tests/test_costs.py

@@ -0,0 +1,73 @@
+import unittest
+
+from gpt_researcher.utils.costs import calculate_llm_cost, estimate_llm_cost
+
+
+class TestCosts(unittest.TestCase):
+    def test_calculate_llm_cost_uses_anthropic_api_usage(self):
+        cost = calculate_llm_cost(
+            llm_provider="anthropic",
+            model="claude-sonnet-4-6",
+            input_content="ignored",
+            output_content="ignored",
+            response_metadata={
+                "model": "claude-sonnet-4-6",
+                "usage": {
+                    "input_tokens": 1000,
+                    "output_tokens": 500,
+                },
+            },
+        )
+
+        self.assertAlmostEqual(cost, 0.0105)
+
+    def test_calculate_llm_cost_supports_dated_anthropic_model_names(self):
+        cost = calculate_llm_cost(
+            llm_provider="anthropic",
+            model="claude-haiku-4-5-20251001",
+            input_content="ignored",
+            output_content="ignored",
+            response_metadata={
+                "usage": {
+                    "input_tokens": 2000,
+                    "output_tokens": 1000,
+                },
+            },
+        )
+
+        self.assertAlmostEqual(cost, 0.007)
+
+    def test_calculate_llm_cost_prefers_native_anthropic_usage(self):
+        cost = calculate_llm_cost(
+            llm_provider="anthropic",
+            model="claude-opus-4-7",
+            input_content="ignored",
+            output_content="ignored",
+            response_metadata={
+                "usage": {
+                    "input_tokens": 100,
+                    "output_tokens": 50,
+                },
+            },
+            usage_metadata={
+                "input_tokens": 999999,
+                "output_tokens": 999999,
+            },
+        )
+
+        self.assertAlmostEqual(cost, 0.00175)
+
+    def test_calculate_llm_cost_falls_back_without_usage(self):
+        fallback_cost = calculate_llm_cost(
+            llm_provider="anthropic",
+            model="claude-sonnet-4-6",
+            input_content="hello",
+            output_content="world",
+        )
+
+        self.assertEqual(fallback_cost, estimate_llm_cost("hello", "world"))
+
+
+if __name__ == "__main__":
+    unittest.main()
+

tests/test_llm_usage_tracking.py

@@ -0,0 +1,42 @@
+import asyncio
+import unittest
+
+from gpt_researcher.llm_provider.generic.base import GenericLLMProvider
+
+
+class _Chunk:
+    def __init__(self, content, usage_metadata=None, response_metadata=None):
+        self.content = content
+        self.usage_metadata = usage_metadata
+        self.response_metadata = response_metadata or {}
+
+
+class _StreamingLLM:
+    async def astream(self, messages, **kwargs):
+        yield _Chunk("Hello")
+        yield _Chunk(
+            "",
+            usage_metadata={"input_tokens": 321, "output_tokens": 123},
+            response_metadata={"usage": {"input_tokens": 321, "output_tokens": 123}},
+        )
+
+
+class TestLLMUsageTracking(unittest.TestCase):
+    def test_stream_response_captures_usage_from_empty_final_chunk(self):
+        provider = GenericLLMProvider(_StreamingLLM(), verbose=False)
+
+        response = asyncio.run(provider.stream_response([{"role": "user", "content": "hi"}]))
+
+        self.assertEqual(response, "Hello")
+        self.assertEqual(
+            provider.last_usage_metadata,
+            {"input_tokens": 321, "output_tokens": 123},
+        )
+        self.assertEqual(
+            provider.last_response_metadata.get("usage"),
+            {"input_tokens": 321, "output_tokens": 123},
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()