import os import json import logging import httpx import base64 import mimetypes from typing import Dict, Any, List, Optional from fastapi import HTTPException import configload as configload # Logger konfigurieren logger = logging.getLogger(__name__) # Konfigurationsdaten laden def load_config_data(): config = configload.load_config() return { "api_key": config.get('Connector_AiAnthropic', 'API_KEY'), "api_url": config.get('Connector_AiAnthropic', 'API_URL', fallback="https://api.anthropic.com/v1/messages"), "model_name": config.get('Connector_AiAnthropic', 'MODEL_NAME', fallback="claude-3-opus-20240229"), "temperature": float(config.get('Connector_AiAnthropic', 'TEMPERATURE', fallback="0.2")), "max_tokens": int(config.get('Connector_AiAnthropic', 'MAX_TOKENS', fallback="2000")) } class ChatService: """ Connector für die Kommunikation mit der Anthropic API. """ def __init__(self): # Konfiguration laden self.config = load_config_data() self.api_key = self.config["api_key"] self.api_url = self.config["api_url"] self.model_name = self.config["model_name"] # HttpClient für API-Aufrufe self.http_client = httpx.AsyncClient( timeout=120.0, # Längeres Timeout für komplexe Anfragen headers={ "x-api-key": self.api_key, "anthropic-version": "2023-06-01", # Anthropic API Version "Content-Type": "application/json" } ) logger.info(f"Anthropic Connector initialisiert mit Modell: {self.model_name}") async def call_api(self, messages: List[Dict[str, Any]], temperature: float = None, max_tokens: int = None) -> Dict[str, Any]: """ Ruft die Anthropic API mit den gegebenen Nachrichten auf. Args: messages: Liste von Nachrichten im OpenAI-Format (role, content) temperature: Temperatur für die Antwortgenerierung (0.0-1.0) max_tokens: Maximale Anzahl der Token in der Antwort Returns: Die Antwort umgewandelt ins OpenAI-Format Raises: HTTPException: Bei Fehlern in der API-Kommunikation """ try: # OpenAI-Format in Anthropic-Format umwandeln formatted_messages = self._convert_to_anthropic_format(messages) # Verwende Parameter aus der Konfiguration, falls keine überschrieben wurden if temperature is None: temperature = self.config.get("temperature", 0.2) if max_tokens is None: max_tokens = self.config.get("max_tokens", 2000) # Anthropic API Payload erstellen payload = { "model": self.model_name, "messages": formatted_messages, "temperature": temperature, "max_tokens": max_tokens } response = await self.http_client.post( self.api_url, json=payload ) if response.status_code != 200: logger.error(f"Anthropic API-Fehler: {response.status_code} - {response.text}") raise HTTPException(status_code=500, detail="Fehler bei der Kommunikation mit Anthropic API") # Antwort im Anthropic-Format in OpenAI-Format umwandeln anthropic_response = response.json() openai_formatted_response = self._convert_to_openai_format(anthropic_response) return openai_formatted_response except Exception as e: logger.error(f"Fehler beim Aufruf der Anthropic API: {str(e)}") raise HTTPException(status_code=500, detail=f"Fehler beim Aufruf der Anthropic API: {str(e)}") def _convert_to_anthropic_format(self, openai_messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Konvertiert Nachrichten vom OpenAI-Format ins Anthropic-Format. OpenAI verwendet: [{"role": "system", "content": "..."}, {"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}] Anthropic verwendet: [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}] Anmerkung: Anthropic hat kein direktes System-Message-Äquivalent, daher fügen wir System-Nachrichten in die erste User-Nachricht ein. """ anthropic_messages = [] system_content = "" # Extrahiere zuerst alle System-Nachrichten for msg in openai_messages: if msg.get("role") == "system": system_content += msg.get("content", "") + "\n\n" # Konvertiere die restlichen Nachrichten for i, msg in enumerate(openai_messages): role = msg.get("role") content = msg.get("content", "") # System-Nachrichten überspringen (bereits extrahiert) if role == "system": continue # Für die erste User-Nachricht: System-Inhalte voranstellen, falls vorhanden if role == "user" and system_content and not any(m.get("role") == "user" for m in anthropic_messages): if isinstance(content, str): content = system_content + content elif isinstance(content, list): # Wenn content ein Array ist (für Multimodal-Nachrichten) text_parts = [] for part in content: if part.get("type") == "text": text_parts.append(part) if text_parts: text_parts[0]["text"] = system_content + text_parts[0].get("text", "") # Anthropic unterstützt nur "user" und "assistant" als Rollen if role not in ["user", "assistant"]: role = "user" anthropic_messages.append({"role": role, "content": content}) return anthropic_messages def _convert_to_openai_format(self, anthropic_response: Dict[str, Any]) -> Dict[str, Any]: """ Konvertiert eine Antwort vom Anthropic-Format ins OpenAI-Format. Anthropic gibt zurück: { "id": "msg_...", "content": [{"type": "text", "text": "Antworttext"}], "model": "claude-...", ... } OpenAI gibt zurück: { "id": "chatcmpl-...", "object": "chat.completion", "choices": [ { "message": { "role": "assistant", "content": "Antworttext" }, "index": 0, "finish_reason": "stop" } ], "model": "gpt-...", ... } """ # Extrahiere Inhalt aus Anthropic-Antwort content = "" if "content" in anthropic_response: if isinstance(anthropic_response["content"], list): # Inhalt ist eine Liste von Teilen (bei neueren API-Versionen) for part in anthropic_response["content"]: if part.get("type") == "text": content += part.get("text", "") else: # Direkter Inhalt als String (bei älteren API-Versionen) content = anthropic_response["content"] # Erstelle OpenAI-formatierte Antwort return { "id": anthropic_response.get("id", ""), "object": "chat.completion", "created": anthropic_response.get("created", 0), "model": anthropic_response.get("model", self.model_name), "choices": [ { "message": { "role": "assistant", "content": content }, "index": 0, "finish_reason": "stop" } ] } def prepare_file_message_content(self, prompt_text: str, file_paths: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Bereitet eine Nachricht mit Dateien für Anthropic API vor. Args: prompt_text: Der Text-Prompt file_paths: Liste von Dateipfaden mit Metadaten (Dict mit id, name, type, path) Returns: Eine für Anthropic-API formatierte content-Liste """ message_content = [ { "type": "text", "text": prompt_text } ] # Füge Dateien als Anhänge hinzu for file_info in file_paths: file_path = file_info.get("path", "") if file_path and os.path.exists(file_path): try: # Datei als Base64 codieren with open(file_path, "rb") as f: file_data = f.read() base64_data = base64.b64encode(file_data).decode('utf-8') # MIME-Typ bestimmen mime_type, _ = mimetypes.guess_type(file_path) if not mime_type: mime_type = "application/octet-stream" # Content-Type bestimmen (image oder document) content_type = "image" if mime_type.startswith("image/") else "document" # Füge die Datei als Anhang hinzu message_content.append({ "type": content_type, "source": { "type": "base64", "media_type": mime_type, "data": base64_data } }) logger.info(f"Datei {file_info.get('name', 'Unbekannt')} als {content_type} hinzugefügt") except Exception as e: logger.error(f"Fehler beim Hinzufügen der Datei {file_info.get('name', 'Unbekannt')}: {str(e)}") return message_content async def close(self): """Schließt den HTTP-Client beim Beenden der Anwendung""" await self.http_client.aclose()