mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-11 02:22:48 +00:00
refactor: reduce statement count in langsmith and anthropic methods
- Extract helper methods in langsmith._prepare_log_data to reduce from 51 to <50 statements - Extract helper methods in anthropic.transform_parsed_response to reduce from 57 to <50 statements - Fixes PLR0915 linter errors - All existing tests pass (10 langsmith tests, 126 anthropic tests) Made-with: Cursor
This commit is contained in:
Sameer Kankute
@@ -5,7 +5,6 @@ import os
|
||||
import random
|
||||
import traceback
|
||||
import types
|
||||
from litellm._uuid import uuid
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
@@ -14,10 +13,11 @@ from pydantic import BaseModel # type: ignore
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm._uuid import uuid
|
||||
from litellm.integrations.custom_batch_logger import CustomBatchLogger
|
||||
from litellm.integrations.langsmith_mock_client import (
|
||||
should_use_langsmith_mock,
|
||||
create_mock_langsmith_client,
|
||||
should_use_langsmith_mock,
|
||||
)
|
||||
from litellm.llms.custom_httpx.http_handler import (
|
||||
get_async_httpx_client,
|
||||
@@ -110,6 +110,56 @@ class LangsmithLogger(CustomBatchLogger):
|
||||
LANGSMITH_TENANT_ID=_credentials_tenant_id,
|
||||
)
|
||||
|
||||
def _extract_metadata_fields(
|
||||
self, metadata: dict, credentials: LangsmithCredentialsObject
|
||||
):
|
||||
return {
|
||||
"project_name": metadata.get("project_name", credentials["LANGSMITH_PROJECT"]),
|
||||
"run_name": metadata.get("run_name", self.langsmith_default_run_name),
|
||||
"run_id": metadata.get("id", metadata.get("run_id", None)),
|
||||
"parent_run_id": metadata.get("parent_run_id", None),
|
||||
"trace_id": metadata.get("trace_id", None),
|
||||
"session_id": metadata.get("session_id", None),
|
||||
"dotted_order": metadata.get("dotted_order", None),
|
||||
}
|
||||
|
||||
def _build_extra_metadata(self, metadata: Dict):
|
||||
extra_metadata = dict(metadata)
|
||||
requester_metadata = extra_metadata.get("requester_metadata")
|
||||
if requester_metadata and isinstance(requester_metadata, dict):
|
||||
for key in ("session_id", "thread_id", "conversation_id"):
|
||||
if key in requester_metadata and key not in extra_metadata:
|
||||
extra_metadata[key] = requester_metadata[key]
|
||||
return extra_metadata
|
||||
|
||||
def _build_outputs_with_usage(self, payload: StandardLoggingPayload) -> Dict[str, Any]:
|
||||
response = payload["response"]
|
||||
outputs: Dict[str, Any]
|
||||
if isinstance(response, dict):
|
||||
outputs = {**response}
|
||||
else:
|
||||
outputs = {"output": response}
|
||||
outputs["usage_metadata"] = {
|
||||
"input_tokens": payload.get("prompt_tokens", 0),
|
||||
"output_tokens": payload.get("completion_tokens", 0),
|
||||
"total_tokens": payload.get("total_tokens", 0),
|
||||
"total_cost": payload.get("response_cost", 0),
|
||||
}
|
||||
return outputs
|
||||
|
||||
def _ensure_required_ids(self, data: dict, run_id: Optional[str]):
|
||||
if "id" not in data or data["id"] is None:
|
||||
run_id = str(uuid.uuid4())
|
||||
data["id"] = run_id
|
||||
|
||||
if "trace_id" not in data or data["trace_id"] is None:
|
||||
if run_id is not None and isinstance(run_id, str):
|
||||
data["trace_id"] = run_id
|
||||
|
||||
if "dotted_order" not in data or data["dotted_order"] is None:
|
||||
if run_id is not None and isinstance(run_id, str):
|
||||
data["dotted_order"] = self.make_dot_order(run_id=run_id)
|
||||
|
||||
def _prepare_log_data(
|
||||
self,
|
||||
kwargs,
|
||||
@@ -121,56 +171,28 @@ class LangsmithLogger(CustomBatchLogger):
|
||||
try:
|
||||
_litellm_params = kwargs.get("litellm_params", {}) or {}
|
||||
metadata = _litellm_params.get("metadata", {}) or {}
|
||||
project_name = metadata.get(
|
||||
"project_name", credentials["LANGSMITH_PROJECT"]
|
||||
)
|
||||
run_name = metadata.get("run_name", self.langsmith_default_run_name)
|
||||
run_id = metadata.get("id", metadata.get("run_id", None))
|
||||
parent_run_id = metadata.get("parent_run_id", None)
|
||||
trace_id = metadata.get("trace_id", None)
|
||||
session_id = metadata.get("session_id", None)
|
||||
dotted_order = metadata.get("dotted_order", None)
|
||||
|
||||
fields = self._extract_metadata_fields(metadata, credentials)
|
||||
verbose_logger.debug(
|
||||
f"Langsmith Logging - project_name: {project_name}, run_name {run_name}"
|
||||
f"Langsmith Logging - project_name: {fields['project_name']}, run_name {fields['run_name']}"
|
||||
)
|
||||
|
||||
# Ensure everything in the payload is converted to str
|
||||
payload: Optional[StandardLoggingPayload] = kwargs.get(
|
||||
"standard_logging_object", None
|
||||
)
|
||||
|
||||
if payload is None:
|
||||
raise Exception("Error logging request payload. Payload=none.")
|
||||
|
||||
metadata = payload[
|
||||
"metadata"
|
||||
] # ensure logged metadata is json serializable
|
||||
|
||||
extra_metadata = dict(metadata)
|
||||
requester_metadata = extra_metadata.get("requester_metadata")
|
||||
if requester_metadata and isinstance(requester_metadata, dict):
|
||||
for key in ("session_id", "thread_id", "conversation_id"):
|
||||
if key in requester_metadata and key not in extra_metadata:
|
||||
extra_metadata[key] = requester_metadata[key]
|
||||
|
||||
outputs = payload["response"]
|
||||
if isinstance(outputs, dict):
|
||||
outputs = {**outputs}
|
||||
else:
|
||||
outputs = {"output": outputs}
|
||||
outputs["usage_metadata"] = {
|
||||
"input_tokens": payload.get("prompt_tokens", 0),
|
||||
"output_tokens": payload.get("completion_tokens", 0),
|
||||
"total_tokens": payload.get("total_tokens", 0),
|
||||
"total_cost": payload.get("response_cost", 0),
|
||||
}
|
||||
metadata = payload["metadata"]
|
||||
extra_metadata = self._build_extra_metadata(dict(metadata))
|
||||
outputs = self._build_outputs_with_usage(payload)
|
||||
|
||||
data = {
|
||||
"name": run_name,
|
||||
"run_type": "llm", # this should always be llm, since litellm always logs llm calls. Langsmith allow us to log "chain"
|
||||
"name": fields["run_name"],
|
||||
"run_type": "llm",
|
||||
"inputs": payload,
|
||||
"outputs": outputs,
|
||||
"session_name": project_name,
|
||||
"session_name": fields["project_name"],
|
||||
"start_time": payload["startTime"],
|
||||
"end_time": payload["endTime"],
|
||||
"tags": payload["request_tags"],
|
||||
@@ -180,46 +202,13 @@ class LangsmithLogger(CustomBatchLogger):
|
||||
if payload["error_str"] is not None and payload["status"] == "failure":
|
||||
data["error"] = payload["error_str"]
|
||||
|
||||
if run_id:
|
||||
data["id"] = run_id
|
||||
|
||||
if parent_run_id:
|
||||
data["parent_run_id"] = parent_run_id
|
||||
|
||||
if trace_id:
|
||||
data["trace_id"] = trace_id
|
||||
|
||||
if session_id:
|
||||
data["session_id"] = session_id
|
||||
|
||||
if dotted_order:
|
||||
data["dotted_order"] = dotted_order
|
||||
|
||||
run_id: Optional[str] = data.get("id") # type: ignore
|
||||
if "id" not in data or data["id"] is None:
|
||||
"""
|
||||
for /batch langsmith requires id, trace_id and dotted_order passed as params
|
||||
"""
|
||||
run_id = str(uuid.uuid4())
|
||||
|
||||
data["id"] = run_id
|
||||
|
||||
if (
|
||||
"trace_id" not in data
|
||||
or data["trace_id"] is None
|
||||
and (run_id is not None and isinstance(run_id, str))
|
||||
):
|
||||
data["trace_id"] = run_id
|
||||
|
||||
if (
|
||||
"dotted_order" not in data
|
||||
or data["dotted_order"] is None
|
||||
and (run_id is not None and isinstance(run_id, str))
|
||||
):
|
||||
data["dotted_order"] = self.make_dot_order(run_id=run_id) # type: ignore
|
||||
for key in ("id", "parent_run_id", "trace_id", "session_id", "dotted_order"):
|
||||
field_key = "run_id" if key == "id" else key
|
||||
if fields[field_key]:
|
||||
data[key] = fields[field_key]
|
||||
|
||||
self._ensure_required_ids(data, fields["run_id"])
|
||||
verbose_logger.debug("Langsmith Logging data on langsmith: %s", data)
|
||||
|
||||
return data
|
||||
except Exception:
|
||||
raise
|
||||
|
||||
@@ -50,6 +50,10 @@ from litellm.types.llms.openai import (
|
||||
OpenAIMcpServerTool,
|
||||
OpenAIWebSearchOptions,
|
||||
)
|
||||
from litellm.types.responses.main import (
|
||||
OutputCodeInterpreterCall,
|
||||
build_code_interpreter_log_outputs,
|
||||
)
|
||||
from litellm.types.utils import (
|
||||
CacheCreationTokenDetails,
|
||||
CompletionTokensDetailsWrapper,
|
||||
@@ -59,10 +63,6 @@ from litellm.types.utils import (
|
||||
PromptTokensDetailsWrapper,
|
||||
ServerToolUse,
|
||||
)
|
||||
from litellm.types.responses.main import (
|
||||
OutputCodeInterpreterCall,
|
||||
build_code_interpreter_log_outputs,
|
||||
)
|
||||
from litellm.utils import (
|
||||
ModelResponse,
|
||||
Usage,
|
||||
@@ -1684,6 +1684,85 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
|
||||
)
|
||||
return usage
|
||||
|
||||
def _build_code_by_id_map(self, tool_calls: List[ChatCompletionToolCallChunk]) -> Dict[str, str]:
|
||||
code_by_id: Dict[str, str] = {}
|
||||
for tc in tool_calls:
|
||||
try:
|
||||
args = json.loads(tc.get("function", {}).get("arguments", "{}"))
|
||||
call_id = tc.get("id")
|
||||
command = args.get("command", "")
|
||||
if isinstance(call_id, str):
|
||||
code_by_id[call_id] = command if isinstance(command, str) else ""
|
||||
except Exception:
|
||||
pass
|
||||
return code_by_id
|
||||
|
||||
def _build_code_interpreter_results(
|
||||
self, tool_results: List[Any], code_by_id: Dict[str, str], container_id: Optional[str]
|
||||
) -> List[OutputCodeInterpreterCall]:
|
||||
code_interpreter_results = []
|
||||
for tr in tool_results:
|
||||
if tr.get("type") != "bash_code_execution_tool_result":
|
||||
continue
|
||||
call_id = tr.get("tool_use_id", "")
|
||||
content = tr.get("content", {})
|
||||
log_outputs = build_code_interpreter_log_outputs(content)
|
||||
code_interpreter_results.append(
|
||||
OutputCodeInterpreterCall(
|
||||
type="code_interpreter_call",
|
||||
id=call_id,
|
||||
code=code_by_id.get(call_id, ""),
|
||||
container_id=container_id,
|
||||
status="completed",
|
||||
outputs=log_outputs,
|
||||
)
|
||||
)
|
||||
return code_interpreter_results
|
||||
|
||||
def _build_provider_specific_fields(
|
||||
self,
|
||||
completion_response: dict,
|
||||
citations: Optional[List[Any]],
|
||||
thinking_blocks: Optional[List[Union[ChatCompletionThinkingBlock, ChatCompletionRedactedThinkingBlock]]],
|
||||
web_search_results: Optional[List[Any]],
|
||||
tool_results: Optional[List[Any]],
|
||||
compaction_blocks: Optional[List[Any]],
|
||||
tool_calls: List[ChatCompletionToolCallChunk],
|
||||
) -> Dict[str, Any]:
|
||||
provider_specific_fields: Dict[str, Any] = {
|
||||
"citations": citations,
|
||||
"thinking_blocks": thinking_blocks,
|
||||
}
|
||||
|
||||
context_management = completion_response.get("context_management")
|
||||
if context_management is not None:
|
||||
provider_specific_fields["context_management"] = context_management
|
||||
|
||||
if web_search_results is not None:
|
||||
provider_specific_fields["web_search_results"] = web_search_results
|
||||
|
||||
if tool_results is not None:
|
||||
provider_specific_fields["tool_results"] = tool_results
|
||||
container_id = (
|
||||
completion_response.get("container", {}).get("id")
|
||||
if isinstance(completion_response.get("container"), dict)
|
||||
else None
|
||||
)
|
||||
code_by_id = self._build_code_by_id_map(tool_calls)
|
||||
code_interpreter_results = self._build_code_interpreter_results(
|
||||
tool_results, code_by_id, container_id
|
||||
)
|
||||
provider_specific_fields["code_interpreter_results"] = code_interpreter_results
|
||||
|
||||
container = completion_response.get("container")
|
||||
if container is not None:
|
||||
provider_specific_fields["container"] = container
|
||||
|
||||
if compaction_blocks is not None:
|
||||
provider_specific_fields["compaction_blocks"] = compaction_blocks
|
||||
|
||||
return provider_specific_fields
|
||||
|
||||
def transform_parsed_response(
|
||||
self,
|
||||
completion_response: dict,
|
||||
@@ -1704,128 +1783,66 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
|
||||
status_code=raw_response.status_code,
|
||||
headers=response_headers,
|
||||
)
|
||||
else:
|
||||
text_content = ""
|
||||
citations: Optional[List[Any]] = None
|
||||
thinking_blocks: Optional[
|
||||
List[
|
||||
Union[
|
||||
ChatCompletionThinkingBlock, ChatCompletionRedactedThinkingBlock
|
||||
]
|
||||
]
|
||||
] = None
|
||||
reasoning_content: Optional[str] = None
|
||||
tool_calls: List[ChatCompletionToolCallChunk] = []
|
||||
|
||||
(
|
||||
text_content,
|
||||
citations,
|
||||
thinking_blocks,
|
||||
reasoning_content,
|
||||
tool_calls,
|
||||
web_search_results,
|
||||
tool_results,
|
||||
compaction_blocks,
|
||||
) = self.extract_response_content(completion_response=completion_response)
|
||||
(
|
||||
text_content,
|
||||
citations,
|
||||
thinking_blocks,
|
||||
reasoning_content,
|
||||
tool_calls,
|
||||
web_search_results,
|
||||
tool_results,
|
||||
compaction_blocks,
|
||||
) = self.extract_response_content(completion_response=completion_response)
|
||||
|
||||
if (
|
||||
prefix_prompt is not None
|
||||
and not text_content.startswith(prefix_prompt)
|
||||
and not litellm.disable_add_prefix_to_prompt
|
||||
):
|
||||
text_content = prefix_prompt + text_content
|
||||
if (
|
||||
prefix_prompt is not None
|
||||
and not text_content.startswith(prefix_prompt)
|
||||
and not litellm.disable_add_prefix_to_prompt
|
||||
):
|
||||
text_content = prefix_prompt + text_content
|
||||
|
||||
context_management: Optional[Dict] = completion_response.get(
|
||||
"context_management"
|
||||
)
|
||||
provider_specific_fields = self._build_provider_specific_fields(
|
||||
completion_response,
|
||||
citations,
|
||||
thinking_blocks,
|
||||
web_search_results,
|
||||
tool_results,
|
||||
compaction_blocks,
|
||||
tool_calls,
|
||||
)
|
||||
|
||||
container: Optional[Dict] = completion_response.get("container")
|
||||
_message = litellm.Message(
|
||||
tool_calls=tool_calls,
|
||||
content=text_content or None,
|
||||
provider_specific_fields=provider_specific_fields,
|
||||
thinking_blocks=thinking_blocks,
|
||||
reasoning_content=reasoning_content,
|
||||
)
|
||||
_message.provider_specific_fields = provider_specific_fields
|
||||
|
||||
provider_specific_fields: Dict[str, Any] = {
|
||||
"citations": citations,
|
||||
"thinking_blocks": thinking_blocks,
|
||||
}
|
||||
if context_management is not None:
|
||||
provider_specific_fields["context_management"] = context_management
|
||||
if web_search_results is not None:
|
||||
provider_specific_fields["web_search_results"] = web_search_results
|
||||
if tool_results is not None:
|
||||
provider_specific_fields["tool_results"] = tool_results
|
||||
# Convert to provider-neutral OutputCodeInterpreterCall objects
|
||||
# so the Responses API layer can use them without Anthropic-specific knowledge.
|
||||
container_id = (
|
||||
completion_response.get("container", {}).get("id")
|
||||
if isinstance(completion_response.get("container"), dict)
|
||||
else None
|
||||
)
|
||||
code_by_id: Dict[str, str] = {}
|
||||
for tc in tool_calls:
|
||||
try:
|
||||
args = json.loads(tc.get("function", {}).get("arguments", "{}"))
|
||||
code_by_id[tc.get("id", "")] = args.get("command", "")
|
||||
except Exception:
|
||||
pass
|
||||
code_interpreter_results = []
|
||||
for tr in tool_results:
|
||||
if tr.get("type") != "bash_code_execution_tool_result":
|
||||
continue
|
||||
call_id = tr.get("tool_use_id", "")
|
||||
content = tr.get("content", {})
|
||||
log_outputs = build_code_interpreter_log_outputs(content)
|
||||
code_interpreter_results.append(
|
||||
OutputCodeInterpreterCall(
|
||||
type="code_interpreter_call",
|
||||
id=call_id,
|
||||
code=code_by_id.get(call_id, ""),
|
||||
container_id=container_id,
|
||||
status="completed",
|
||||
outputs=log_outputs,
|
||||
)
|
||||
)
|
||||
provider_specific_fields["code_interpreter_results"] = (
|
||||
code_interpreter_results
|
||||
)
|
||||
if container is not None:
|
||||
provider_specific_fields["container"] = container
|
||||
if compaction_blocks is not None:
|
||||
provider_specific_fields["compaction_blocks"] = compaction_blocks
|
||||
json_mode_message = self._transform_response_for_json_mode(
|
||||
json_mode=json_mode,
|
||||
tool_calls=tool_calls,
|
||||
)
|
||||
if json_mode_message is not None:
|
||||
completion_response["stop_reason"] = "stop"
|
||||
_message = json_mode_message
|
||||
|
||||
_message = litellm.Message(
|
||||
tool_calls=tool_calls,
|
||||
content=text_content or None,
|
||||
provider_specific_fields=provider_specific_fields,
|
||||
thinking_blocks=thinking_blocks,
|
||||
reasoning_content=reasoning_content,
|
||||
)
|
||||
_message.provider_specific_fields = provider_specific_fields
|
||||
model_response.choices[0].message = _message
|
||||
model_response._hidden_params["original_response"] = completion_response["content"]
|
||||
model_response.choices[0].finish_reason = cast(
|
||||
OpenAIChatCompletionFinishReason,
|
||||
map_finish_reason(completion_response["stop_reason"]),
|
||||
)
|
||||
|
||||
## HANDLE JSON MODE - anthropic returns single function call
|
||||
json_mode_message = self._transform_response_for_json_mode(
|
||||
json_mode=json_mode,
|
||||
tool_calls=tool_calls,
|
||||
)
|
||||
if json_mode_message is not None:
|
||||
completion_response["stop_reason"] = "stop"
|
||||
_message = json_mode_message
|
||||
|
||||
model_response.choices[0].message = _message # type: ignore
|
||||
model_response._hidden_params["original_response"] = completion_response[
|
||||
"content"
|
||||
] # allow user to access raw anthropic tool calling response
|
||||
|
||||
model_response.choices[0].finish_reason = cast(
|
||||
OpenAIChatCompletionFinishReason,
|
||||
map_finish_reason(completion_response["stop_reason"]),
|
||||
)
|
||||
|
||||
## CALCULATING USAGE
|
||||
usage = self.calculate_usage(
|
||||
usage_object=completion_response["usage"],
|
||||
reasoning_content=reasoning_content,
|
||||
completion_response=completion_response,
|
||||
speed=speed,
|
||||
)
|
||||
setattr(model_response, "usage", usage) # type: ignore
|
||||
setattr(model_response, "usage", usage)
|
||||
|
||||
model_response.created = int(time.time())
|
||||
model_response.model = completion_response["model"]
|
||||
|
||||
Reference in New Issue
Block a user