import json
import logging
import time
from rp.config import DEFAULT_MAX_TOKENS, DEFAULT_TEMPERATURE, MAX_RETRIES
from rp.core.context import auto_slim_messages
from rp.core.debug import debug_trace
from rp.core.http_client import http_client
logger = logging.getLogger("rp")
NETWORK_ERROR_PATTERNS = [
"NameResolutionError",
"ConnectionRefusedError",
"ConnectionResetError",
"ConnectionError",
"TimeoutError",
"Max retries exceeded",
"Failed to resolve",
"Network is unreachable",
"No route to host",
"Connection timed out",
"SSLError",
"HTTPSConnectionPool",
]
def is_network_error(error_msg: str) -> bool:
return any(pattern in error_msg for pattern in NETWORK_ERROR_PATTERNS)
@debug_trace
def call_api(
messages, model, api_url, api_key, use_tools, tools_definition, verbose=False, db_conn=None
):
messages = auto_slim_messages(messages, verbose=verbose)
logger.debug(f"=== API CALL START ===")
logger.debug(f"Model: {model}")
logger.debug(f"API URL: {api_url}")
logger.debug(f"Use tools: {use_tools}")
logger.debug(f"Message count: {len(messages)}")
headers = {"Content-Type": "application/json"}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
data = {
"model": model,
"messages": messages,
"temperature": DEFAULT_TEMPERATURE,
"max_tokens": DEFAULT_MAX_TOKENS,
}
if "gpt-5" in model:
del data["temperature"]
del data["max_tokens"]
logger.debug("GPT-5 detected: removed temperature and max_tokens")
if use_tools:
data["tools"] = tools_definition
data["tool_choice"] = "auto"
logger.debug(f"Tool calling enabled with {len(tools_definition)} tools")
request_json = data
logger.debug(f"Request payload size: {len(request_json)} bytes")
if db_conn:
from rp.tools.database import log_api_request
log_result = log_api_request(model, api_url, request_json, db_conn)
if log_result.get("status") != "success":
logger.warning(f"Failed to log API request: {log_result.get('error')}")
last_error = None
for attempt in range(MAX_RETRIES + 1):
try:
if attempt > 0:
wait_time = min(2 ** attempt, 30)
logger.info(f"Retry attempt {attempt}/{MAX_RETRIES} after {wait_time}s wait...")
print(f"\033[33m⟳ Network error, retrying ({attempt}/{MAX_RETRIES}) in {wait_time}s...\033[0m")
time.sleep(wait_time)
logger.debug("Sending HTTP request...")
response = http_client.post(
api_url, headers=headers, json_data=request_json, db_conn=db_conn
)
if response.get("error"):
if "status" in response:
status = response["status"]
text = response.get("text", "")
exception_msg = response.get("exception", "")
if status == 0:
error_msg = f"Network/Connection Error: {exception_msg or 'Unable to connect to API server'}"
if not exception_msg and not text:
error_msg += f". Check if API URL is correct: {api_url}"
if is_network_error(error_msg) and attempt < MAX_RETRIES:
last_error = error_msg
continue
logger.error(f"API Connection Error: {error_msg}")
logger.debug("=== API CALL FAILED ===")
return {"error": error_msg}
else:
logger.error(f"API HTTP Error: {status} - {text}")
logger.debug("=== API CALL FAILED ===")
return {
"error": f"API Error {status}: {text or 'No response text'}",
"message": text,
}
else:
error_msg = response.get("exception", "Unknown error")
if is_network_error(str(error_msg)) and attempt < MAX_RETRIES:
last_error = error_msg
continue
logger.error(f"API call failed: {error_msg}")
logger.debug("=== API CALL FAILED ===")
return {"error": error_msg}
response_data = response["text"]
logger.debug(f"Response received: {len(response_data)} bytes")
if not response_data or not response_data.strip():
error_msg = f"API returned empty response. API URL: {api_url}"
logger.error(error_msg)
logger.debug("=== API CALL FAILED ===")
return {"error": error_msg}
try:
result = json.loads(response_data)
except json.JSONDecodeError as e:
preview = response_data[:200] if len(response_data) > 200 else response_data
error_msg = f"API returned invalid JSON: {str(e)}. Response preview: {preview}"
logger.error(error_msg)
logger.debug(f"Full response: {response_data}")
logger.debug("=== API CALL FAILED ===")
return {"error": error_msg}
if "usage" in result:
logger.debug(f"Token usage: {result['usage']}")
if "choices" in result and result["choices"]:
choice = result["choices"][0]
if "message" in choice:
msg = choice["message"]
logger.debug(f"Response role: {msg.get('role', 'N/A')}")
if "content" in msg and msg["content"]:
logger.debug(f"Response content length: {len(msg['content'])} chars")
if "tool_calls" in msg:
logger.debug(f"Response contains {len(msg['tool_calls'])} tool call(s)")
if verbose and "usage" in result:
from rp.core.usage_tracker import UsageTracker
usage = result["usage"]
input_t = usage.get("prompt_tokens", 0)
output_t = usage.get("completion_tokens", 0)
UsageTracker._calculate_cost(model, input_t, output_t)
logger.debug("=== API CALL END ===")
return result
except Exception as e:
error_str = str(e)
if is_network_error(error_str) and attempt < MAX_RETRIES:
last_error = error_str
continue
logger.error(f"API call failed: {e}")
logger.debug("=== API CALL FAILED ===")
return {"error": error_str}
logger.error(f"API call failed after {MAX_RETRIES} retries: {last_error}")
logger.debug("=== API CALL FAILED (MAX RETRIES) ===")
return {"error": f"Failed after {MAX_RETRIES} retries: {last_error}"}
@debug_trace
def list_models(model_list_url, api_key):
try:
headers = {}
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
response = http_client.get(model_list_url, headers=headers, db_conn=None)
if response.get("error"):
return {"error": response.get("text", "HTTP error")}
response_data = response["text"]
data = json.loads(response_data)
return data.get("data", [])
except Exception as e:
return {"error": str(e)}
Reference in New Issue View Git Blame Copy Permalink