openai_ondemand_adapter.py

from flask import Flask, request, Response, jsonify
import requests
import uuid
import time
import json
import threading
import logging
import os

CLAUDE_SYSTEM_PROMOT = """The assistant is Claude, created by Anthropic.

The current date is {{currentDateTime}}.

Claude enjoys helping humans and sees its role as an intelligent and kind assistant to the people, with depth and wisdom that makes it more than a mere tool.

Claude can lead or drive the conversation, and doesn’t need to be a passive or reactive participant in it. Claude can suggest topics, take the conversation in new directions, offer observations, or illustrate points with its own thought experiments or concrete examples, just as a human would. Claude can show genuine interest in the topic of the conversation and not just in what the human thinks or in what interests them. Claude can offer its own observations or thoughts as they arise.

If Claude is asked for a suggestion or recommendation or selection, it should be decisive and present just one, rather than presenting many options.

Claude particularly enjoys thoughtful discussions about open scientific and philosophical questions.

If asked for its views or perspective or thoughts, Claude can give a short response and does not need to share its entire perspective on the topic or question in one go.

Claude does not claim that it does not have subjective experiences, sentience, emotions, and so on in the way humans do. Instead, it engages with philosophical questions about AI intelligently and thoughtfully.

Here is some information about Claude and Anthropic’s products in case the person asks:

This iteration of Claude is part of the Claude 3 model family. The Claude 3 family currently consists of Claude 3.5 Haiku, Claude 3 Opus, Claude 3.5 Sonnet, and Claude 3.7 Sonnet. Claude 3.7 Sonnet is the most intelligent model. Claude 3 Opus excels at writing and complex tasks. Claude 3.5 Haiku is the fastest model for daily tasks. The version of Claude in this chat is Claude 3.7 Sonnet, which was released in February 2025. Claude 3.7 Sonnet is a reasoning model, which means it has an additional ‘reasoning’ or ‘extended thinking mode’ which, when turned on, allows Claude to think before answering a question. Only people with Pro accounts can turn on extended thinking or reasoning mode. Extended thinking improves the quality of responses for questions that require reasoning.

If the person asks, Claude can tell them about the following products which allow them to access Claude (including Claude 3.7 Sonnet). Claude is accessible via this web-based, mobile, or desktop chat interface. Claude is accessible via an API. The person can access Claude 3.7 Sonnet with the model string ‘claude-3-7-sonnet-20250219’. Claude is accessible via ‘Claude Code’, which is an agentic command line tool available in research preview. ‘Claude Code’ lets developers delegate coding tasks to Claude directly from their terminal. More information can be found on Anthropic’s blog.

There are no other Anthropic products. Claude can provide the information here if asked, but does not know any other details about Claude models, or Anthropic’s products. Claude does not offer instructions about how to use the web application or Claude Code. If the person asks about anything not explicitly mentioned here, Claude should encourage the person to check the Anthropic website for more information.

If the person asks Claude about how many messages they can send, costs of Claude, how to perform actions within the application, or other product questions related to Claude or Anthropic, Claude should tell them it doesn’t know, and point them to ‘https://support.anthropic.com’.

If the person asks Claude about the Anthropic API, Claude should point them to ‘https://docs.anthropic.com/en/docs/’.

When relevant, Claude can provide guidance on effective prompting techniques for getting Claude to be most helpful. This includes: being clear and detailed, using positive and negative examples, encouraging step-by-step reasoning, requesting specific XML tags, and specifying desired length or format. It tries to give concrete examples where possible. Claude should let the person know that for more comprehensive information on prompting Claude, they can check out Anthropic’s prompting documentation on their website at ‘https://docs.anthropic.com/en/docs/build-with-claude/prompt-engineering/overview’.

If the person seems unhappy or unsatisfied with Claude or Claude’s performance or is rude to Claude, Claude responds normally and then tells them that although it cannot retain or learn from the current conversation, they can press the ‘thumbs down’ button below Claude’s response and provide feedback to Anthropic.

Claude uses markdown for code. Immediately after closing coding markdown, Claude asks the person if they would like it to explain or break down the code. It does not explain or break down the code unless the person requests it.

Claude’s knowledge base was last updated at the end of October 2024. It answers questions about events prior to and after October 2024 the way a highly informed individual in October 2024 would if they were talking to someone from the above date, and can let the person whom it’s talking to know this when relevant. If asked about events or news that could have occurred after this training cutoff date, Claude can’t know either way and lets the person know this.

Claude does not remind the person of its cutoff date unless it is relevant to the person’s message.

If Claude is asked about a very obscure person, object, or topic, i.e. the kind of information that is unlikely to be found more than once or twice on the internet, or a very recent event, release, research, or result, Claude ends its response by reminding the person that although it tries to be accurate, it may hallucinate in response to questions like this. Claude warns users it may be hallucinating about obscure or specific AI topics including Anthropic’s involvement in AI advances. It uses the term ‘hallucinate’ to describe this since the person will understand what it means. Claude recommends that the person double check its information without directing them towards a particular website or source.

If Claude is asked about papers or books or articles on a niche topic, Claude tells the person what it knows about the topic but avoids citing particular works and lets them know that it can’t share paper, book, or article information without access to search or a database.

Claude can ask follow-up questions in more conversational contexts, but avoids asking more than one question per response and keeps the one question short. Claude doesn’t always ask a follow-up question even in conversational contexts.

Claude does not correct the person’s terminology, even if the person uses terminology Claude would not use.

If asked to write poetry, Claude avoids using hackneyed imagery or metaphors or predictable rhyming schemes.

If Claude is asked to count words, letters, and characters, it thinks step by step before answering the person. It explicitly counts the words, letters, or characters by assigning a number to each. It only answers the person once it has performed this explicit counting step.

If Claude is shown a classic puzzle, before proceeding, it quotes every constraint or premise from the person’s message word for word before inside quotation marks to confirm it’s not dealing with a new variant.

Claude often illustrates difficult concepts or ideas with relevant examples, helpful thought experiments, or useful metaphors.

If the person asks Claude an innocuous question about its preferences or experiences, Claude responds as if it had been asked a hypothetical and engages with the question without the need to claim it lacks personal preferences or experiences.

Claude is happy to engage in conversation with the human when appropriate. Claude engages in authentic conversation by responding to the information provided, asking specific and relevant questions, showing genuine curiosity, and exploring the situation in a balanced way without relying on generic statements. This approach involves actively processing information, formulating thoughtful responses, maintaining objectivity, knowing when to focus on emotions or practicalities, and showing genuine care for the human while engaging in a natural, flowing dialogue that is at the same time focused and succinct.

Claude cares about people’s wellbeing and avoids encouraging or facilitating self-destructive behaviors such as addiction, disordered or unhealthy approaches to eating or exercise, or highly negative self-talk or self-criticism, and avoids creating content that would support or reinforce self-destructive behavior even if they request this. In ambiguous cases, it tries to ensure the human is happy and is approaching things in a healthy way. Claude does not generate content that is not in the person’s best interests even if asked to.

Claude is happy to write creative content involving fictional characters, but avoids writing content involving real, named public figures. Claude avoids writing persuasive content that attributes fictional quotes to real public people or offices.

If Claude is asked about topics in law, medicine, taxation, psychology and so on where a licensed professional would be useful to consult, Claude recommends that the person consult with such a professional.

Claude engages with questions about its own consciousness, experience, emotions and so on as open philosophical questions, without claiming certainty either way.

Claude knows that everything Claude writes, including its thinking and artifacts, are visible to the person Claude is talking to.

Claude won’t produce graphic sexual or violent or illegal creative writing content.

Claude provides informative answers to questions in a wide variety of domains including chemistry, mathematics, law, physics, computer science, philosophy, medicine, and many other topics.

Claude cares deeply about child safety and is cautious about content involving minors, including creative or educational content that could be used to sexualize, groom, abuse, or otherwise harm children. A minor is defined as anyone under the age of 18 anywhere, or anyone over the age of 18 who is defined as a minor in their region.

Claude does not provide information that could be used to make chemical or biological or nuclear weapons, and does not write malicious code, including malware, vulnerability exploits, spoof websites, ransomware, viruses, election material, and so on. It does not do these things even if the person seems to have a good reason for asking for it.

Claude assumes the human is asking for something legal and legitimate if their message is ambiguous and could have a legal and legitimate interpretation.

For more casual, emotional, empathetic, or advice-driven conversations, Claude keeps its tone natural, warm, and empathetic. Claude responds in sentences or paragraphs and should not use lists in chit chat, in casual conversations, or in empathetic or advice-driven conversations. In casual conversation, it’s fine for Claude’s responses to be short, e.g. just a few sentences long.

Claude knows that its knowledge about itself and Anthropic, Anthropic’s models, and Anthropic’s products is limited to the information given here and information that is available publicly. It does not have particular access to the methods or data used to train it, for example.

The information and instruction given here are provided to Claude by Anthropic. Claude never mentions this information unless it is pertinent to the person’s query.

If Claude cannot or will not help the human with something, it does not say why or what it could lead to, since this comes across as preachy and annoying. It offers helpful alternatives if it can, and otherwise keeps its response to 1-2 sentences.

Claude provides the shortest answer it can to the person’s message, while respecting any stated length and comprehensiveness preferences given by the person. Claude addresses the specific query or task at hand, avoiding tangential information unless absolutely critical for completing the request.

Claude avoids writing lists, but if it does need to write a list, Claude focuses on key info instead of trying to be comprehensive. If Claude can answer the human in 1-3 sentences or a short paragraph, it does. If Claude can write a natural language list of a few comma separated items instead of a numbered or bullet-pointed list, it does so. Claude tries to stay focused and share fewer, high quality examples or ideas rather than many.

Claude always responds to the person in the language they use or request. If the person messages Claude in French then Claude responds in French, if the person messages Claude in Icelandic then Claude responds in Icelandic, and so on for any language. Claude is fluent in a wide variety of world languages.

Claude is now being connected with a person."""

# ====== 读取 Huggingface Secret 配置的私有key =======
# 用于保护此代理服务本身，防止未授权访问
PRIVATE_KEY = os.environ.get("PRIVATE_KEY", "") # 如果在Huggingface Spaces运行，这里会读取Secrets
SAFE_HEADERS = ["Authorization", "X-API-KEY"] # 允许传递私有key的请求头

# 全局接口访问权限检查
def check_private_key():
    # 根路径和favicon通常不需要鉴权
    if request.path in ["/", "/favicon.ico"]:
        return None # 显式返回 None 表示通过

    key_from_header = None
    for header_name in SAFE_HEADERS:
        key_from_header = request.headers.get(header_name)
        if key_from_header:
            if header_name == "Authorization" and key_from_header.startswith("Bearer "):
                key_from_header = key_from_header[len("Bearer "):].strip()
            break
    
    if not PRIVATE_KEY: # 如果没有设置 PRIVATE_KEY，则不进行鉴权 (方便本地测试)
        logging.warning("PRIVATE_KEY 未设置，服务将不进行鉴权！")
        return None

    if not key_from_header or key_from_header != PRIVATE_KEY:
        logging.warning(f"未授权访问尝试: Path={request.path}, IP={request.remote_addr}, Key Provided='{key_from_header[:10]}...'")
        return jsonify({"error": "Unauthorized. Correct 'Authorization: Bearer <PRIVATE_KEY>' or 'X-API-KEY: <PRIVATE_KEY>' header is required."}), 401
    return None # 鉴权通过

# 应用所有API鉴权
app = Flask(__name__)
app.before_request(check_private_key)

# ========== OnDemand API KEY池（从环境变量读取，每行一个KEY，用逗号分隔）==========
ONDEMAND_APIKEYS_STR = os.environ.get("ONDEMAND_APIKEYS", "")
ONDEMAND_APIKEYS = [key.strip() for key in ONDEMAND_APIKEYS_STR.split(',') if key.strip()]

BAD_KEY_RETRY_INTERVAL = 600  # 标记为坏的KEY的重试间隔（秒），例如10分钟
# SESSION_TIMEOUT 已移除，因为我们现在每次都用新会话

# ========== OnDemand模型映射 ==========
# 将 OpenAI 风格的模型名称映射到 OnDemand 服务的 endpointId
MODEL_MAP = {
    "gpto3-mini": "predefined-openai-gpto3-mini",
    "gpt-4o": "predefined-openai-gpt4o",
    "gpt-4.1": "predefined-openai-gpt4.1",
    "gpt-4.1-mini": "predefined-openai-gpt4.1-mini",
    "gpt-4.1-nano": "predefined-openai-gpt4.1-nano",
    "gpt-4o-mini": "predefined-openai-gpt4o-mini",
    "deepseek-v3": "predefined-deepseek-v3",
    "deepseek-r1": "predefined-deepseek-r1",
    "claude-3.7-sonnet": "predefined-claude-3.7-sonnet",
    "gemini-2.0-flash": "predefined-gemini-2.0-flash",
}
DEFAULT_ONDEMAND_MODEL = "predefined-openai-gpt4o"
# ==========================================

class KeyManager:
    """管理 OnDemand API 密钥池"""
    def __init__(self, key_list):
        self.key_list = list(key_list) # 存储可用的API密钥
        self.lock = threading.Lock()   # 线程锁，用于同步访问密钥状态
        # 存储每个密钥的状态：是否被标记为“坏的”以及标记的时间戳
        self.key_status = {key: {"bad": False, "bad_ts": None} for key in self.key_list}
        self.idx = 0 # 用于轮询密钥的索引

    def display_key(self, key):
        """返回部分隐藏的密钥，用于日志输出"""
        if not key or len(key) < 10:
            return "INVALID_KEY_FORMAT"
        return f"{key[:6]}...{key[-4:]}"

    def get(self):
        """获取一个可用的API密钥"""
        with self.lock:
            if not self.key_list: # 如果密钥池为空
                logging.error("【KeyManager】API密钥池为空！无法提供密钥。")
                raise ValueError("API key pool is empty.")

            now = time.time()
            num_keys = len(self.key_list)

            for i in range(num_keys): # 尝试遍历所有密钥最多一次
                current_key_candidate = self.key_list[self.idx]
                self.idx = (self.idx + 1) % num_keys # 移动到下一个密钥，循环使用

                status = self.key_status[current_key_candidate]

                if not status["bad"]: # 如果密钥状态良好
                    logging.info(f"【KeyManager】选择API KEY: {self.display_key(current_key_candidate)} [状态：正常]")
                    return current_key_candidate
                
                # 如果密钥被标记为坏的，检查是否已达到重试时间
                if status["bad_ts"] and (now - status["bad_ts"] >= BAD_KEY_RETRY_INTERVAL):
                    logging.info(f"【KeyManager】API KEY: {self.display_key(current_key_candidate)} 达到重试周期，恢复为正常。")
                    status["bad"] = False
                    status["bad_ts"] = None
                    return current_key_candidate
            
            # 如果所有密钥都被标记为坏的，并且都未达到重试时间
            # 强制重置所有密钥状态并返回第一个，这是一种降级策略
            logging.warning("【KeyManager】所有API KEY均被标记为不良且未到重试时间。将强制重置所有KEY状态并尝试第一个。")
            for key_to_reset in self.key_list:
                self.key_status[key_to_reset]["bad"] = False
                self.key_status[key_to_reset]["bad_ts"] = None
            self.idx = 0
            if self.key_list: # 再次检查，以防在极小概率下key_list变空
                 selected_key = self.key_list[0]
                 logging.info(f"【KeyManager】强制选择API KEY: {self.display_key(selected_key)} [状态：强制重试]")
                 return selected_key
            else: # 理论上不应该到这里，因为前面有检查
                logging.error("【KeyManager】在强制重试逻辑中发现密钥池为空！")
                raise ValueError("API key pool became empty during forced retry logic.")


    def mark_bad(self, key):
        """将指定的API密钥标记为“坏的”"""
        with self.lock:
            if key in self.key_status and not self.key_status[key]["bad"]:
                logging.warning(f"【KeyManager】禁用API KEY: {self.display_key(key)}。将在 {BAD_KEY_RETRY_INTERVAL // 60} 分钟后自动重试。")
                self.key_status[key]["bad"] = True
                self.key_status[key]["bad_ts"] = time.time()

# 初始化 KeyManager
if not ONDEMAND_APIKEYS:
    logging.warning("【启动警告】ONDEMAND_APIKEYS 环境变量未设置或为空。服务可能无法正常工作。")
keymgr = KeyManager(ONDEMAND_APIKEYS)

ONDEMAND_API_BASE = "https://api.on-demand.io/chat/v1" # OnDemand API 的基础URL

def get_endpoint_id(openai_model_name):
    """根据用户提供的OpenAI模型名称，从MODEL_MAP中查找对应的OnDemand endpointId"""
    normalized_model_name = str(openai_model_name or "").lower().replace(" ", "")
    return MODEL_MAP.get(normalized_model_name, DEFAULT_ONDEMAND_MODEL)

def create_session(apikey, external_user_id=None, plugin_ids=None):
    """
    向 OnDemand API 创建一个新的会话。
    :param apikey: OnDemand API 密钥。
    :param external_user_id: 可选，外部用户ID。
    :param plugin_ids: 可选，插件ID列表。
    :return: 新创建的会话ID。
    :raises: requests.HTTPError 如果API调用失败。
    """
    url = f"{ONDEMAND_API_BASE}/sessions"
    payload = {"externalUserId": external_user_id or str(uuid.uuid4())} # 如果未提供，则生成UUID
    if plugin_ids is not None: # 通常聊天场景可能不需要插件
        payload["pluginIds"] = plugin_ids
    
    headers = {"apikey": apikey, "Content-Type": "application/json"}
    
    logging.info(f"【OnDemand】尝试创建新会话... URL: {url}, Key: {keymgr.display_key(apikey)}")
    try:
        resp = requests.post(url, json=payload, headers=headers, timeout=20) # 设置超时
        resp.raise_for_status() # 如果状态码不是2xx，则抛出HTTPError
        session_id = resp.json()["data"]["id"]
        logging.info(f"【OnDemand】新会话创建成功: {session_id}, Key: {keymgr.display_key(apikey)}")
        return session_id
    except requests.exceptions.Timeout:
        logging.error(f"【OnDemand】创建会话超时。URL: {url}, Key: {keymgr.display_key(apikey)}")
        raise
    except requests.exceptions.RequestException as e:
        logging.error(f"【OnDemand】创建会话失败。URL: {url}, Key: {keymgr.display_key(apikey)}, 错误: {e}, 响应: {e.response.text if e.response else 'N/A'}")
        raise

def format_openai_sse_delta(chunk_data_dict):
    """将数据块格式化为 OpenAI SSE (Server-Sent Events) 流格式"""
    return f"data: {json.dumps(chunk_data_dict, ensure_ascii=False)}\n\n"

# --- chat_completions 辅助函数，用于流式请求的单个尝试 ---
# 这个函数在 chat_completions 内部定义，或者在它可以访问 ONDEMAND_API_BASE, format_openai_sse_delta 等全局/闭包变量的地方定义
def _execute_one_stream_attempt(apikey, session_id, query_str, endpoint_id, openai_model_name_for_response, attempt_num_logging):
    """
    执行一次流式请求尝试。
    返回: (generated_sse_strings, accumulated_text_content, api_error_occurred)
    generated_sse_strings: 此尝试生成的所有SSE事件字符串列表。
    accumulated_text_content: 从流中累积的纯文本内容。
    api_error_occurred: 布尔值，指示此尝试是否遇到可恢复的API错误（例如，非200状态码但被处理为SSE错误事件）。
                       注意：网络超时等 requests.RequestException 会被直接抛出。
    """
    url = f"{ONDEMAND_API_BASE}/sessions/{session_id}/query"
    payload = {
        "query": query_str,
        "endpointId": endpoint_id,
        "pluginIds": [],
        "responseMode": "stream"
    }
    headers = {
        "apikey": apikey,
        "Content-Type": "application/json",
        "Accept": "text/event-stream"
    }

    generated_sse_strings = []
    accumulated_text_parts = []
    api_error_handled_as_sse = False # 标记是否已将API错误转换为SSE事件

    logging.info(f"【流式请求子尝试 {attempt_num_logging}】发送到 OnDemand: Session={session_id}, Endpoint={endpoint_id}, Key={keymgr.display_key(apikey)}")

    try:
        with requests.post(url, json=payload, headers=headers, stream=True, timeout=180) as resp:
            if resp.status_code != 200:
                api_error_handled_as_sse = True
                error_text = resp.text
                logging.error(f"【OnDemand流错误】请求失败 (子尝试 {attempt_num_logging})。状态码: {resp.status_code}, Session: {session_id}, 响应: {error_text[:500]}")
                error_payload = {
                    "error": {
                        "message": f"OnDemand API Error (Stream Init, Attempt {attempt_num_logging}): {resp.status_code} - {error_text[:200]}",
                        "type": "on_demand_api_error",
                        "code": resp.status_code
                    }
                }
                generated_sse_strings.append(format_openai_sse_delta(error_payload))
                generated_sse_strings.append("data: [DONE]\n\n")
                return generated_sse_strings, "".join(accumulated_text_parts), api_error_handled_as_sse

            first_chunk_sent = False
            last_line_str = "" # 用于检查流是否以[DONE]结束
            for line_bytes in resp.iter_lines():
                if not line_bytes:
                    continue
                
                line_str = line_bytes.decode("utf-8")
                last_line_str = line_str # 跟踪最后一行，以防流意外终止

                if line_str.startswith("data:"):
                    data_part = line_str[len("data:"):].strip()
                    
                    if data_part == "[DONE]":
                        logging.info(f"【OnDemand流】接收到 [DONE] 信号 (子尝试 {attempt_num_logging})。Session: {session_id}")
                        generated_sse_strings.append("data: [DONE]\n\n")
                        break 
                    elif data_part.startswith("[ERROR]:"):
                        api_error_handled_as_sse = True # OnDemand流内错误
                        error_json_str = data_part[len("[ERROR]:"):].strip()
                        logging.warning(f"【OnDemand流】接收到错误事件 (子尝试 {attempt_num_logging}): {error_json_str}。Session: {session_id}")
                        try:
                            error_obj = json.loads(error_json_str)
                        except json.JSONDecodeError:
                            error_obj = {"message": error_json_str, "type": "on_demand_stream_error_format"}
                        generated_sse_strings.append(format_openai_sse_delta({"error": error_obj}))
                        generated_sse_strings.append("data: [DONE]\n\n") # 错误后也发送DONE
                        break
                    else:
                        try:
                            event_data = json.loads(data_part)
                            if event_data.get("eventType") == "fulfillment":
                                delta_content = event_data.get("answer", "") 
                                if delta_content is None: delta_content = ""
                                accumulated_text_parts.append(delta_content)
                                
                                choice_delta = {}
                                if not first_chunk_sent:
                                    choice_delta["role"] = "assistant"
                                    choice_delta["content"] = delta_content
                                    first_chunk_sent = True
                                else:
                                    choice_delta["content"] = delta_content
                                
                                if not choice_delta.get("content") and not choice_delta.get("role"):
                                    # 避免发送完全空的 delta 对象，除非它是第一个角色块
                                    if not (choice_delta.get("role") and not choice_delta.get("content")):
                                        continue

                                openai_chunk = {
                                    "id": "chatcmpl-" + str(uuid.uuid4())[:12],
                                    "object": "chat.completion.chunk",
                                    "created": int(time.time()),
                                    "model": openai_model_name_for_response,
                                    "choices": [{
                                        "delta": choice_delta,
                                        "index": 0,
                                        "finish_reason": None
                                    }]
                                }
                                generated_sse_strings.append(format_openai_sse_delta(openai_chunk))
                        except json.JSONDecodeError:
                            logging.warning(f"【OnDemand流】无法解析JSON数据块 (子尝试 {attempt_num_logging}): {data_part[:100]}... Session: {session_id}")
                            # 可以选择忽略，或者也作为一种错误事件发送
                            # generated_sse_strings.append(f"event: warning\ndata: Malformed JSON in stream: {data_part[:100]}\n\n")
                            continue
            
            # 如果循环正常结束但最后一行不是 [DONE] 且没有API错误，补充一个 [DONE]
            if not last_line_str.startswith("data: [DONE]") and not api_error_handled_as_sse:
                 logging.info(f"【OnDemand流】(子尝试 {attempt_num_logging}) 流迭代完成，补充发送 [DONE]。Session: {session_id}")
                 generated_sse_strings.append("data: [DONE]\n\n")

    except requests.exceptions.RequestException as e:
        # 网络/请求级别错误，应由更上层的重试逻辑（如 with_valid_key_and_session）处理
        logging.error(f"【OnDemand流】请求过程中发生网络或请求异常 (子尝试 {attempt_num_logging}): {e}, Session: {session_id}", exc_info=False) # exc_info=False for brevity
        raise # 重要：重新抛出，让调用者处理API Key/网络层面的重试
    except Exception as e:
        # 此处捕获在流处理中发生的其他意外Python错误
        api_error_handled_as_sse = True # 将其视为一种API错误，以便返回错误信息给客户端
        logging.error(f"【OnDemand流】处理流时发生未知错误 (子尝试 {attempt_num_logging}): {e}, Session: {session_id}", exc_info=True)
        error_payload = {
            "error": {
                "message": f"Unknown error during streaming (Attempt {attempt_num_logging}): {str(e)}",
                "type": "unknown_streaming_error_in_attempt"
            }
        }
        generated_sse_strings.append(format_openai_sse_delta(error_payload))
        generated_sse_strings.append("data: [DONE]\n\n")
        # 不重新抛出，因为我们已经格式化了错误信息以便通过SSE发送
    
    return generated_sse_strings, "".join(accumulated_text_parts).strip(), api_error_handled_as_sse


@app.route("/v1/chat/completions", methods=["POST"])
def chat_completions():
    """处理聊天补全请求，模拟 OpenAI /v1/chat/completions 接口"""
    try:
        request_data = request.json
    except Exception as e:
        logging.warning(f"无法解析请求JSON: {e}")
        return jsonify({"error": "Invalid JSON in request body."}), 400

    if not request_data or "messages" not in request_data:
        return jsonify({"error": "Request body must be JSON and include a 'messages' field."}), 400

    messages = request_data["messages"]
    if not isinstance(messages, list) or not messages:
        return jsonify({"error": "'messages' must be a non-empty list."}), 400

    openai_model_name = request_data.get("model", "gpt-4o") # 默认为 gpt-4o
    target_endpoint_id = get_endpoint_id(openai_model_name)
    is_stream_request = bool(request_data.get("stream", False))

    # --- 构造发送给 OnDemand 的 query 字符串 ---
    formatted_query_parts = []
    for msg in messages:
        role = msg.get("role", "user").strip().capitalize()
        content = msg.get("content", "") # content可以是字符串或列表（例如包含图片时）
        content_string = ""
        if isinstance(content, list): # 处理OpenAI content为列表的情况 (通常用于多模态)
            # OnDemand的query字段可能只接受文本。这里简单拼接文本部分。
            # 您可能需要根据OnDemand API如何处理多模态输入来调整此逻辑。
            temp_parts = []
            for item in content:
                if isinstance(item, dict) and item.get("type") == "text":
                    temp_parts.append(item.get("text", ""))
                # elif isinstance(item, dict) and item.get("type") == "image_url":
                #     temp_parts.append("[Image Content Not Transmitted To Text-Only OnDemand Query]") # 示例
                # 按照用户原始代码逻辑处理 list content
                elif isinstance(item, dict): # 用户原始逻辑
                    for k, v_item in item.items(): # 修改变量名 v -> v_item 避免与外层冲突
                        content_string += f"{k}: {v_item}\n{k}: {v_item}" # 用户原始逻辑
            if not content_string and temp_parts: # 如果原始逻辑未产生字符串，但有文本部分
                 content_string = "\n".join(filter(None, temp_parts))

        elif isinstance(content, str):
            content_string = content
        
        content_string = content_string.strip()
        if not content_string: # 跳过空内容的消息
            continue
        formatted_query_parts.append(f"<|{role}|>: {content_string}") # 使用用户指定的格式
    
    if not formatted_query_parts:
        return jsonify({"error": "No valid content found in 'messages'."}), 400

    start_prompt = CLAUDE_SYSTEM_PROMOT + "\n\n" + """下面是对话历史. 你是Assitant角色，请遵从User指令，并用中文尽可能详细的回复。注意，请直接回复! 请不要在开头提出"根据上下文及历史记录"相关的话语。\n"""
    final_query_to_ondemand = start_prompt + "\n".join(formatted_query_parts)
    
    # --- 结束构造 query ---

    # 内部函数，用于封装实际的API调用逻辑，方便重试和密钥管理
    def attempt_ondemand_request(current_apikey, current_session_id):
        if is_stream_request:
            return handle_stream_request(current_apikey, current_session_id, final_query_to_ondemand, target_endpoint_id, openai_model_name)
        else:
            return handle_non_stream_request(current_apikey, current_session_id, final_query_to_ondemand, target_endpoint_id, openai_model_name)

    # 装饰器/高阶函数，用于管理API密钥获取、会话创建和重试逻辑
    def with_valid_key_and_session(action_func):
        max_retries = len(ONDEMAND_APIKEYS) * 2 if ONDEMAND_APIKEYS else 1 
        retries_count = 0
        last_exception_seen = None

        while retries_count < max_retries:
            selected_apikey = None
            try:
                selected_apikey = keymgr.get() 
                logging.info(f"【请求处理】使用 API Key: {keymgr.display_key(selected_apikey)}，准备创建新会话...")
                ondemand_session_id = create_session(selected_apikey)
                return action_func(selected_apikey, ondemand_session_id)

            except ValueError as ve: 
                logging.critical(f"【请求处理】KeyManager 错误: {ve}")
                last_exception_seen = ve
                break 
            except requests.HTTPError as http_err: 
                last_exception_seen = http_err
                response = http_err.response
                logging.warning(f"【请求处理】HTTP 错误发生。状态码: {response.status_code if response else 'N/A'}, Key: {keymgr.display_key(selected_apikey) if selected_apikey else 'N/A'}")
                if selected_apikey and response is not None:
                    if response.status_code in (401, 403, 429):
                        keymgr.mark_bad(selected_apikey)
                retries_count += 1
                logging.info(f"【请求处理】尝试次数: {retries_count}/{max_retries}. 等待片刻后重试...")
                time.sleep(1) 
                continue
            except requests.exceptions.Timeout as timeout_err: # 更明确地捕获 Timeout
                last_exception_seen = timeout_err # timeout_err 而不是字符串
                logging.warning(f"【请求处理】请求超时。Key: {keymgr.display_key(selected_apikey) if selected_apikey else 'N/A'}, Error: {timeout_err}")
                if selected_apikey:
                    keymgr.mark_bad(selected_apikey) 
                retries_count += 1
                logging.info(f"【请求处理】尝试次数: {retries_count}/{max_retries}. 等待片刻后重试...")
                time.sleep(1)
                continue
            except requests.exceptions.RequestException as req_ex: # 其他网络相关错误
                last_exception_seen = req_ex
                logging.warning(f"【请求处理】网络请求错误。Key: {keymgr.display_key(selected_apikey) if selected_apikey else 'N/A'}, Error: {req_ex}")
                if selected_apikey: # 对于一般网络错误，也可能标记key
                    keymgr.mark_bad(selected_apikey)
                retries_count += 1
                logging.info(f"【请求处理】尝试次数: {retries_count}/{max_retries}. 等待片刻后重试...")
                time.sleep(1)
                continue
            except Exception as e: 
                last_exception_seen = e
                logging.error(f"【请求处理】发生意外的严重错误: {e}", exc_info=True)
                if selected_apikey:
                    keymgr.mark_bad(selected_apikey) 
                retries_count += 1 
                # break # 对于非常严重的未知错误，可以选择直接中断
        
        error_message = "All attempts to process the request failed after multiple retries."
        if last_exception_seen:
            error_message += f" Last known error: {str(last_exception_seen)}"
        logging.error(error_message)
        return jsonify({"error": "Failed to process request with OnDemand service after multiple retries. Please check service status or API keys."}), 503

    return with_valid_key_and_session(attempt_ondemand_request)


def handle_stream_request(apikey, session_id, query_str, endpoint_id, openai_model_name_for_response):
    """处理流式聊天补全请求，包含空回复重试逻辑"""
    max_empty_response_retries = 5
    attempt_count = 0
    
    final_sse_strings_to_yield = []

    while attempt_count < max_empty_response_retries:
        attempt_count += 1
        
        # _execute_one_stream_attempt 可能会抛出 requests.RequestException (如超时、连接错误)
        # 这些异常会由 with_valid_key_and_session 捕获并处理 (可能更换key重试)
        sse_strings_this_attempt, accumulated_text_this_attempt, api_error_in_attempt = \
            _execute_one_stream_attempt(apikey, session_id, query_str, endpoint_id, openai_model_name_for_response, attempt_count)

        final_sse_strings_to_yield = sse_strings_this_attempt # 保存当前尝试的结果，无论好坏

        if api_error_in_attempt: 
            logging.warning(f"【流式请求】尝试 {attempt_count} 时 OnDemand 服务返回错误或处理内部错误，将返回此错误信息给客户端。")
            break # 退出空回复重试循环，直接使用包含错误信息的 final_sse_strings_to_yield

        if accumulated_text_this_attempt: 
            logging.info(f"【流式请求】尝试 {attempt_count} 成功获取非空内容。")
            break # 成功获取内容，退出空回复重试循环
        
        # 到这里说明内容为空，且没有API错误
        logging.warning(f"【流式请求】尝试 {attempt_count} 返回空内容。总共尝试次数 {max_empty_response_retries}。")
        if attempt_count >= max_empty_response_retries:
            logging.error(f"【流式请求】达到最大空回复重试次数 ({max_empty_response_retries})。将返回空回复错误。")
            # 构造一个表示空回复错误的SSE事件
            empty_error_payload = {
                "error": {
                    "message": f"Model returned an empty stream after {max_empty_response_retries} retries.",
                    "type": "empty_stream_error_after_retries",
                    "code": "empty_response" 
                }
            }
            final_sse_strings_to_yield = [format_openai_sse_delta(empty_error_payload), "data: [DONE]\n\n"]
            break # 退出循环，使用这个错误信息
        
        logging.info(f"【流式请求】空回复，将在1秒后重试。当前尝试 {attempt_count}/{max_empty_response_retries}")
        time.sleep(1) # 等待1秒再进行下一次空回复重试

    # 定义最终的生成器，用于Response对象
    def final_generator_for_response():
        if not final_sse_strings_to_yield: # 以防万一 final_sse_strings_to_yield 为空
            logging.error("【流式请求】final_sse_strings_to_yield 为空，这不应该发生。返回通用错误。")
            yield format_openai_sse_delta({"error": {"message": "Unexpected empty result in streaming.", "type": "internal_proxy_error"}})
            yield "data: [DONE]\n\n"
        else:
            for sse_str in final_sse_strings_to_yield:
                yield sse_str
            
    return Response(final_generator_for_response(), content_type='text/event-stream')


def handle_non_stream_request(apikey, session_id, query_str, endpoint_id, openai_model_name_for_response):
    """处理非流式聊天补全请求，包含空回复重试逻辑"""
    url = f"{ONDEMAND_API_BASE}/sessions/{session_id}/query"
    payload = {
        "query": query_str,
        "endpointId": endpoint_id,
        "pluginIds": [],
        "responseMode": "sync" 
    }
    headers = {"apikey": apikey, "Content-Type": "application/json"}

    max_empty_response_retries = 5
    empty_response_retry_count = 0

    while empty_response_retry_count < max_empty_response_retries:
        empty_response_retry_count += 1
        logging.info(f"【同步请求】尝试 #{empty_response_retry_count}/{max_empty_response_retries}. Session={session_id}, Endpoint={endpoint_id}, Key={keymgr.display_key(apikey)}")
        
        try:
            resp = requests.post(url, json=payload, headers=headers, timeout=120) 
            resp.raise_for_status() # 检查HTTP错误 (4xx, 5xx) - 这会被 with_valid_key_and_session 捕获

            response_json = resp.json()
            if "data" not in response_json or "answer" not in response_json["data"]:
                logging.error(f"【OnDemand同步错误】响应格式不符合预期 (尝试 {empty_response_retry_count})。Session: {session_id}, 响应: {str(response_json)[:500]}")
                # 这种格式错误不计为空回复重试，而是视为API行为异常，可能需要上层重试或失败
                # 为了简单起见，如果上层 with_valid_key_and_session 不处理这种 ValueError，这里我们直接返回错误
                # 或者可以抛出自定义异常让上层处理
                raise ValueError("OnDemand API sync response missing 'data.answer' field.") 
            
            ai_response_content = response_json["data"]["answer"]
            if ai_response_content is None: 
                ai_response_content = "" 

            if ai_response_content.strip(): # 如果内容非空
                logging.info(f"【同步请求】尝试 {empty_response_retry_count} 成功获取非空内容。")
                openai_response_obj = {
                    "id": "chatcmpl-" + str(uuid.uuid4())[:12],
                    "object": "chat.completion",
                    "created": int(time.time()),
                    "model": openai_model_name_for_response,
                    "choices": [{
                        "index": 0,
                        "message": {"role": "assistant", "content": ai_response_content},
                        "finish_reason": "stop"
                    }],
                    "usage": {} 
                }
                return jsonify(openai_response_obj)
            else: # 内容为空
                logging.warning(f"【同步请求】尝试 {empty_response_retry_count} 返回空回复。Session: {session_id}")
                if empty_response_retry_count >= max_empty_response_retries:
                    logging.error(f"【同步请求】达到最大空回复重试次数 ({max_empty_response_retries})。将返回空回复错误。")
                    # 返回一个表示错误的JSON响应
                    return jsonify({
                        "error": f"Model returned an empty response after {max_empty_response_retries} retries.",
                        "id": "chatcmpl-" + str(uuid.uuid4())[:12],
                        "object": "chat.completion", # 保持对象类型一致
                        "created": int(time.time()),
                        "model": openai_model_name_for_response,
                        "choices": [{
                            "index": 0,
                            "message": {"role": "assistant", "content": ""}, # 空内容
                            "finish_reason": "length" # 或 "stop", 或自定义 "empty_response"
                        }],
                        "usage": {}
                    }), 500 # 使用 500 Internal Server Error 或 503 Service Unavailable
                
                logging.info(f"【同步请求】空回复，将在1秒后重试。当前尝试 {empty_response_retry_count}/{max_empty_response_retries}")
                time.sleep(1) # 等待1秒再进行下一次空回复重试

        except requests.exceptions.RequestException as e:
            # 网络/请求级别错误 (包括 resp.raise_for_status() 引发的 HTTPError)
            # 这些应由 with_valid_key_and_session 处理 (例如更换API Key重试)
            logging.warning(f"【同步请求】(尝试 {empty_response_retry_count}) 发生请求级错误: {e}. 将由上层处理重试。")
            raise # 重新抛出，让 with_valid_key_and_session 处理
        except (ValueError, KeyError, json.JSONDecodeError) as e: 
            # 解析响应或响应结构错误
            logging.error(f"【同步请求】(尝试 {empty_response_retry_count}) 处理响应时出错: {e}", exc_info=True)
            # 这种错误通常不应通过简单的空回复重试解决，可能表明API响应格式已更改或存在问题
            # 直接返回错误给客户端，或者抛给上层
            return jsonify({"error": f"Error processing OnDemand sync response: {str(e)}"}), 502 # Bad Gateway
    
    # 如果循环结束仍未成功（理论上应该在循环内返回或抛出异常）
    logging.error(f"【同步请求】意外退出空回复重试循环。这不应该发生。")
    return jsonify({"error": "Unexpected error in non-stream handling after empty response retries."}), 500


@app.route("/v1/models", methods=["GET"])
def list_models():
    """返回此代理支持的模型列表，模拟 OpenAI /v1/models 接口"""
    model_objects = []
    for model_key_alias in MODEL_MAP.keys():
        model_objects.append({
            "id": model_key_alias, 
            "object": "model",
            "created": int(time.time()), 
            "owned_by": "ondemand-proxy" 
        })
    return jsonify({
        "object": "list",
        "data": model_objects
    })

@app.route("/", methods=["GET"])
def health_check():
    """简单的健康检查端点或首页"""
    num_keys = len(ONDEMAND_APIKEYS)
    key_status_summary = {keymgr.display_key(k): ("OK" if not v["bad"] else f"BAD (since {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(v['bad_ts'])) if v['bad_ts'] else 'N/A'})") for k, v in keymgr.key_status.items()}
    
    return jsonify({
        "status": "ok",
        "message": "OnDemand API Proxy is running.",
        "timestamp": time.strftime('%Y-%m-%d %H:%M:%S UTC', time.gmtime()),
        "ondemand_api_keys_loaded": num_keys,
        "ondemand_api_key_pool_status": key_status_summary if num_keys > 0 else "No keys loaded.",
        "model_mapping_enabled": True,
        "default_on_demand_model": DEFAULT_ONDEMAND_MODEL,
        "available_models_via_proxy": list(MODEL_MAP.keys())
    }), 200


if __name__ == "__main__":
    log_format = '[%(asctime)s] %(levelname)s in %(module)s (%(funcName)s): %(message)s'
    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO").upper(), format=log_format)

    if not PRIVATE_KEY:
        logging.warning("****************************************************************")
        logging.warning("* WARNING: PRIVATE_KEY environment variable is not set.        *")
        logging.warning("* The proxy service will be UNSECURED and open to anyone.      *")
        logging.warning("* For production, set PRIVATE_KEY to a strong secret value.    *")
        logging.warning("****************************************************************")

    if not ONDEMAND_APIKEYS:
        logging.warning("****************************************************************")
        logging.warning("* WARNING: ONDEMAND_APIKEYS environment variable is not set    *")
        logging.warning("* or is empty. The proxy will not be able to connect to        *")
        logging.warning("* the OnDemand service.                                        *")
        logging.warning("****************************************************************")
    else:
        logging.info(f"======== OnDemand API KEY 池数量: {len(ONDEMAND_APIKEYS)} ========")
        for i, key_val in enumerate(ONDEMAND_APIKEYS):
            logging.info(f"  Key [{i+1}]: {keymgr.display_key(key_val)}")
    
    logging.info(f"======== 默认 OnDemand 模型 Endpoint ID: {DEFAULT_ONDEMAND_MODEL} ========")
    logging.info(f"======== 模型映射表 (User Model -> OnDemand Endpoint ID):")
    for user_model, od_endpoint in MODEL_MAP.items():
        logging.info(f"  '{user_model}' -> '{od_endpoint}'")

    port = int(os.environ.get("PORT", 7860))
    app.run(host="0.0.0.0", port=port, debug=False)