|
|
|
|
|
|
|
|
|
|
| import os
|
| import re
|
| import json
|
| import uuid
|
| import httpx
|
| import secrets
|
| import string
|
| import traceback
|
| import time
|
| import random
|
| import threading
|
| import logging
|
| from typing import List, Dict, Any, Optional, Literal, Union
|
| from collections import OrderedDict
|
|
|
| from fastapi import FastAPI, Request, Header, HTTPException, Depends
|
| from fastapi.responses import JSONResponse, StreamingResponse
|
| from pydantic import BaseModel, ValidationError
|
|
|
| from config_loader import config_loader
|
|
|
| logger = logging.getLogger(__name__)
|
|
|
| def generate_random_trigger_signal() -> str:
|
| """Generate a random, self-closing trigger signal like <Function_AB1c_Start/>."""
|
| chars = string.ascii_letters + string.digits
|
| random_str = ''.join(secrets.choice(chars) for _ in range(4))
|
| return f"<Function_{random_str}_Start/>"
|
|
|
| try:
|
| app_config = config_loader.load_config()
|
|
|
| log_level_str = app_config.features.log_level
|
| if log_level_str == "DISABLED":
|
| log_level = logging.CRITICAL + 1
|
| else:
|
| log_level = getattr(logging, log_level_str, logging.INFO)
|
|
|
| logging.basicConfig(
|
| level=log_level,
|
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
| datefmt='%Y-%m-%d %H:%M:%S'
|
| )
|
|
|
| logger.info(f"✅ Configuration loaded successfully: {config_loader.config_path}")
|
| logger.info(f"📊 Configured {len(app_config.upstream_services)} upstream services")
|
| logger.info(f"🔑 Configured {len(app_config.client_authentication.allowed_keys)} client keys")
|
|
|
| MODEL_TO_SERVICE_MAPPING, ALIAS_MAPPING = config_loader.get_model_to_service_mapping()
|
| DEFAULT_SERVICE = config_loader.get_default_service()
|
| ALLOWED_CLIENT_KEYS = config_loader.get_allowed_client_keys()
|
| GLOBAL_TRIGGER_SIGNAL = generate_random_trigger_signal()
|
|
|
| logger.info(f"🎯 Configured {len(MODEL_TO_SERVICE_MAPPING)} model mappings")
|
| if ALIAS_MAPPING:
|
| logger.info(f"🔄 Configured {len(ALIAS_MAPPING)} model aliases: {list(ALIAS_MAPPING.keys())}")
|
| logger.info(f"🔄 Default service: {DEFAULT_SERVICE['name']}")
|
|
|
| except Exception as e:
|
| logger.error(f"❌ Configuration loading failed: {type(e).__name__}")
|
| logger.error(f"❌ Error details: {str(e)}")
|
| logger.error("💡 Please ensure config.yaml file exists and is properly formatted")
|
| exit(1)
|
| class ToolCallMappingManager:
|
| """
|
| Tool call mapping manager with TTL (Time To Live) and size limit
|
|
|
| Features:
|
| 1. Automatic expiration cleanup - entries are automatically deleted after specified time
|
| 2. Size limit - prevents unlimited memory growth
|
| 3. LRU eviction - removes least recently used entries when size limit is reached
|
| 4. Thread safe - supports concurrent access
|
| 5. Periodic cleanup - background thread regularly cleans up expired entries
|
| """
|
|
|
| def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600, cleanup_interval: int = 300):
|
| """
|
| Initialize mapping manager
|
|
|
| Args:
|
| max_size: Maximum number of stored entries
|
| ttl_seconds: Entry time to live (seconds)
|
| cleanup_interval: Cleanup interval (seconds)
|
| """
|
| self.max_size = max_size
|
| self.ttl_seconds = ttl_seconds
|
| self.cleanup_interval = cleanup_interval
|
|
|
| self._data: OrderedDict[str, Dict[str, Any]] = OrderedDict()
|
| self._timestamps: Dict[str, float] = {}
|
| self._lock = threading.RLock()
|
|
|
| self._cleanup_thread = threading.Thread(target=self._periodic_cleanup, daemon=True)
|
| self._cleanup_thread.start()
|
|
|
| logger.debug(f"🔧 [INIT] Tool call mapping manager started - Max entries: {max_size}, TTL: {ttl_seconds}s, Cleanup interval: {cleanup_interval}s")
|
|
|
| def store(self, tool_call_id: str, name: str, args: dict, description: str = "") -> None:
|
| """Store tool call mapping"""
|
| with self._lock:
|
| current_time = time.time()
|
|
|
| if tool_call_id in self._data:
|
| del self._data[tool_call_id]
|
| del self._timestamps[tool_call_id]
|
|
|
| while len(self._data) >= self.max_size:
|
| oldest_key = next(iter(self._data))
|
| del self._data[oldest_key]
|
| del self._timestamps[oldest_key]
|
| logger.debug(f"🔧 [CLEANUP] Removed oldest entry due to size limit: {oldest_key}")
|
|
|
| self._data[tool_call_id] = {
|
| "name": name,
|
| "args": args,
|
| "description": description,
|
| "created_at": current_time
|
| }
|
| self._timestamps[tool_call_id] = current_time
|
|
|
| logger.debug(f"🔧 Stored tool call mapping: {tool_call_id} -> {name}")
|
| logger.debug(f"🔧 Current mapping table size: {len(self._data)}")
|
|
|
| def get(self, tool_call_id: str) -> Optional[Dict[str, Any]]:
|
| """Get tool call mapping (updates LRU order)"""
|
| with self._lock:
|
| current_time = time.time()
|
|
|
| if tool_call_id not in self._data:
|
| logger.debug(f"🔧 Tool call mapping not found: {tool_call_id}")
|
| logger.debug(f"🔧 All IDs in current mapping table: {list(self._data.keys())}")
|
| return None
|
|
|
| if current_time - self._timestamps[tool_call_id] > self.ttl_seconds:
|
| logger.debug(f"🔧 Tool call mapping expired: {tool_call_id}")
|
| del self._data[tool_call_id]
|
| del self._timestamps[tool_call_id]
|
| return None
|
|
|
| result = self._data[tool_call_id]
|
| self._data.move_to_end(tool_call_id)
|
|
|
| logger.debug(f"🔧 Found tool call mapping: {tool_call_id} -> {result['name']}")
|
| return result
|
|
|
| def cleanup_expired(self) -> int:
|
| """Clean up expired entries, return the number of cleaned entries"""
|
| with self._lock:
|
| current_time = time.time()
|
| expired_keys = []
|
|
|
| for key, timestamp in self._timestamps.items():
|
| if current_time - timestamp > self.ttl_seconds:
|
| expired_keys.append(key)
|
|
|
| for key in expired_keys:
|
| del self._data[key]
|
| del self._timestamps[key]
|
|
|
| if expired_keys:
|
| logger.debug(f"🔧 [CLEANUP] Cleaned up {len(expired_keys)} expired entries")
|
|
|
| return len(expired_keys)
|
|
|
| def get_stats(self) -> Dict[str, Any]:
|
| """Get statistics"""
|
| with self._lock:
|
| current_time = time.time()
|
| expired_count = sum(1 for ts in self._timestamps.values()
|
| if current_time - ts > self.ttl_seconds)
|
|
|
| return {
|
| "total_entries": len(self._data),
|
| "expired_entries": expired_count,
|
| "active_entries": len(self._data) - expired_count,
|
| "max_size": self.max_size,
|
| "ttl_seconds": self.ttl_seconds,
|
| "memory_usage_ratio": len(self._data) / self.max_size
|
| }
|
|
|
| def _periodic_cleanup(self) -> None:
|
| """Background periodic cleanup thread"""
|
| while True:
|
| try:
|
| time.sleep(self.cleanup_interval)
|
| cleaned = self.cleanup_expired()
|
|
|
| stats = self.get_stats()
|
| if stats["total_entries"] > 0:
|
| logger.debug(f"🔧 [STATS] Mapping table status: Total={stats['total_entries']}, "
|
| f"Active={stats['active_entries']}, Memory usage={stats['memory_usage_ratio']:.1%}")
|
|
|
| except Exception as e:
|
| logger.error(f"❌ Background cleanup thread exception: {e}")
|
|
|
| TOOL_CALL_MAPPING_MANAGER = ToolCallMappingManager(
|
| max_size=1000,
|
| ttl_seconds=3600,
|
| cleanup_interval=300
|
| )
|
|
|
| def store_tool_call_mapping(tool_call_id: str, name: str, args: dict, description: str = ""):
|
| """Store mapping between tool call ID and call content"""
|
| TOOL_CALL_MAPPING_MANAGER.store(tool_call_id, name, args, description)
|
|
|
| def get_tool_call_mapping(tool_call_id: str) -> Optional[Dict[str, Any]]:
|
| """Get call content corresponding to tool call ID"""
|
| return TOOL_CALL_MAPPING_MANAGER.get(tool_call_id)
|
|
|
| def format_tool_result_for_ai(tool_call_id: str, result_content: str) -> str:
|
| """Format tool call results for AI understanding with English prompts and XML structure"""
|
| logger.debug(f"🔧 Formatting tool call result: tool_call_id={tool_call_id}")
|
| tool_info = get_tool_call_mapping(tool_call_id)
|
| if not tool_info:
|
| logger.debug(f"🔧 Tool call mapping not found, using default format")
|
| return f"Tool execution result:\n<tool_result>\n{result_content}\n</tool_result>"
|
|
|
| formatted_text = f"""Tool execution result:
|
| - Tool name: {tool_info['name']}
|
| - Execution result:
|
| <tool_result>
|
| {result_content}
|
| </tool_result>"""
|
|
|
| logger.debug(f"🔧 Formatting completed, tool name: {tool_info['name']}")
|
| return formatted_text
|
|
|
| def format_assistant_tool_calls_for_ai(tool_calls: List[Dict[str, Any]], trigger_signal: str) -> str:
|
| """Format assistant tool calls into AI-readable string format."""
|
| logger.debug(f"🔧 Formatting assistant tool calls. Count: {len(tool_calls)}")
|
|
|
| xml_calls_parts = []
|
| for tool_call in tool_calls:
|
| function_info = tool_call.get("function", {})
|
| name = function_info.get("name", "")
|
| arguments_json = function_info.get("arguments", "{}")
|
|
|
| try:
|
|
|
| args_dict = json.loads(arguments_json)
|
| except (json.JSONDecodeError, TypeError):
|
|
|
| args_dict = {"raw_arguments": arguments_json}
|
|
|
| args_parts = []
|
| for key, value in args_dict.items():
|
|
|
| json_value = json.dumps(value, ensure_ascii=False)
|
| args_parts.append(f"<{key}>{json_value}</{key}>")
|
|
|
| args_content = "\n".join(args_parts)
|
|
|
| xml_call = f"<function_call>\n<tool>{name}</tool>\n<args>\n{args_content}\n</args>\n</function_call>"
|
| xml_calls_parts.append(xml_call)
|
|
|
| all_calls = "\n".join(xml_calls_parts)
|
| final_str = f"{trigger_signal}\n<function_calls>\n{all_calls}\n</function_calls>"
|
|
|
| logger.debug("🔧 Assistant tool calls formatted successfully.")
|
| return final_str
|
|
|
| def get_function_call_prompt_template(trigger_signal: str) -> str:
|
| """
|
| Generate prompt template based on dynamic trigger signal
|
| """
|
| custom_template = app_config.features.prompt_template
|
| if custom_template:
|
| logger.info("🔧 Using custom prompt template from configuration")
|
| return custom_template.format(
|
| trigger_signal=trigger_signal,
|
| tools_list="{tools_list}"
|
| )
|
|
|
| return f"""
|
| You have access to the following available tools to help solve problems:
|
|
|
| {{tools_list}}
|
|
|
| **IMPORTANT CONTEXT NOTES:**
|
| 1. You can call MULTIPLE tools in a single response if needed.
|
| 2. The conversation context may already contain tool execution results from previous function calls. Review the conversation history carefully to avoid unnecessary duplicate tool calls.
|
| 3. When tool execution results are present in the context, they will be formatted with XML tags like <tool_result>...</tool_result> for easy identification.
|
| 4. This is the ONLY format you can use for tool calls, and any deviation will result in failure.
|
|
|
| When you need to use tools, you **MUST** strictly follow this format. Do NOT include any extra text, explanations, or dialogue on the first and second lines of the tool call syntax:
|
|
|
| 1. When starting tool calls, begin on a new line with exactly:
|
| {trigger_signal}
|
| No leading or trailing spaces, output exactly as shown above. The trigger signal MUST be on its own line and appear only once.
|
|
|
| 2. Starting from the second line, **immediately** follow with the complete <function_calls> XML block.
|
|
|
| 3. For multiple tool calls, include multiple <function_call> blocks within the same <function_calls> wrapper.
|
|
|
| 4. Do not add any text or explanation after the closing </function_calls> tag.
|
|
|
| STRICT ARGUMENT KEY RULES:
|
| - You MUST use parameter keys EXACTLY as defined (case- and punctuation-sensitive). Do NOT rename, add, or remove characters.
|
| - If a key starts with a hyphen (e.g., -i, -C), you MUST keep the hyphen in the tag name. Example: <-i>true</-i>, <-C>2</-C>.
|
| - Never convert "-i" to "i" or "-C" to "C". Do not pluralize, translate, or alias parameter keys.
|
| - The <tool> tag must contain the exact name of a tool from the list. Any other tool name is invalid.
|
| - The <args> must contain all required arguments for that tool.
|
|
|
| CORRECT Example (multiple tool calls, including hyphenated keys):
|
| ...response content (optional)...
|
| {trigger_signal}
|
| <function_calls>
|
| <function_call>
|
| <tool>Grep</tool>
|
| <args>
|
| <-i>true</-i>
|
| <-C>2</-C>
|
| <path>.</path>
|
| </args>
|
| </function_call>
|
| <function_call>
|
| <tool>search</tool>
|
| <args>
|
| <keywords>["Python Document", "how to use python"]</keywords>
|
| </args>
|
| </function_call>
|
| </function_calls>
|
|
|
| INCORRECT Example (extra text + wrong key names — DO NOT DO THIS):
|
| ...response content (optional)...
|
| {trigger_signal}
|
| I will call the tools for you.
|
| <function_calls>
|
| <function_call>
|
| <tool>Grep</tool>
|
| <args>
|
| <i>true</i>
|
| <C>2</C>
|
| <path>.</path>
|
| </args>
|
| </function_call>
|
| </function_calls>
|
|
|
| Now please be ready to strictly follow the above specifications.
|
| """
|
|
|
| class ToolFunction(BaseModel):
|
| name: str
|
| description: Optional[str] = None
|
| parameters: Dict[str, Any]
|
|
|
| class Tool(BaseModel):
|
| type: Literal["function"]
|
| function: ToolFunction
|
|
|
| class Message(BaseModel):
|
| role: str
|
| content: Optional[str] = None
|
| tool_calls: Optional[List[Dict[str, Any]]] = None
|
| tool_call_id: Optional[str] = None
|
| name: Optional[str] = None
|
|
|
| class Config:
|
| extra = "allow"
|
|
|
| class ToolChoice(BaseModel):
|
| type: Literal["function"]
|
| function: Dict[str, str]
|
|
|
| class ChatCompletionRequest(BaseModel):
|
| model: str
|
| messages: List[Dict[str, Any]]
|
| tools: Optional[List[Tool]] = None
|
| tool_choice: Optional[Union[str, ToolChoice]] = None
|
| stream: Optional[bool] = False
|
| stream_options: Optional[Dict[str, Any]] = None
|
| temperature: Optional[float] = None
|
| max_tokens: Optional[int] = None
|
| top_p: Optional[float] = None
|
| frequency_penalty: Optional[float] = None
|
| presence_penalty: Optional[float] = None
|
| n: Optional[int] = None
|
| stop: Optional[Union[str, List[str]]] = None
|
|
|
| class Config:
|
| extra = "allow"
|
|
|
|
|
| def generate_function_prompt(tools: List[Tool], trigger_signal: str) -> tuple[str, str]:
|
| """
|
| Generate injected system prompt based on tools definition in client request.
|
| Returns: (prompt_content, trigger_signal)
|
| """
|
| tools_list_str = []
|
| for i, tool in enumerate(tools):
|
| func = tool.function
|
| name = func.name
|
| description = func.description or ""
|
|
|
|
|
| schema: Dict[str, Any] = func.parameters or {}
|
| props: Dict[str, Any] = schema.get("properties", {}) or {}
|
| required_list: List[str] = schema.get("required", []) or []
|
|
|
|
|
| params_summary = ", ".join([
|
| f"{p_name} ({(p_info or {}).get('type', 'any')})" for p_name, p_info in props.items()
|
| ]) or "None"
|
|
|
|
|
| detail_lines: List[str] = []
|
| for p_name, p_info in props.items():
|
| p_info = p_info or {}
|
| p_type = p_info.get("type", "any")
|
| is_required = "Yes" if p_name in required_list else "No"
|
| p_desc = p_info.get("description")
|
| enum_vals = p_info.get("enum")
|
| default_val = p_info.get("default")
|
| examples_val = p_info.get("examples") or p_info.get("example")
|
|
|
|
|
| constraints: Dict[str, Any] = {}
|
| for key in [
|
| "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum",
|
| "minLength", "maxLength", "pattern", "format",
|
| "minItems", "maxItems", "uniqueItems"
|
| ]:
|
| if key in p_info:
|
| constraints[key] = p_info.get(key)
|
|
|
|
|
| if p_type == "array":
|
| items = p_info.get("items") or {}
|
| if isinstance(items, dict):
|
| itype = items.get("type")
|
| if itype:
|
| constraints["items.type"] = itype
|
|
|
|
|
| detail_lines.append(f"- {p_name}:")
|
| detail_lines.append(f" - type: {p_type}")
|
| detail_lines.append(f" - required: {is_required}")
|
| if p_desc:
|
| detail_lines.append(f" - description: {p_desc}")
|
| if enum_vals is not None:
|
| try:
|
| detail_lines.append(f" - enum: {json.dumps(enum_vals, ensure_ascii=False)}")
|
| except Exception:
|
| detail_lines.append(f" - enum: {enum_vals}")
|
| if default_val is not None:
|
| try:
|
| detail_lines.append(f" - default: {json.dumps(default_val, ensure_ascii=False)}")
|
| except Exception:
|
| detail_lines.append(f" - default: {default_val}")
|
| if examples_val is not None:
|
| try:
|
| detail_lines.append(f" - examples: {json.dumps(examples_val, ensure_ascii=False)}")
|
| except Exception:
|
| detail_lines.append(f" - examples: {examples_val}")
|
| if constraints:
|
| try:
|
| detail_lines.append(f" - constraints: {json.dumps(constraints, ensure_ascii=False)}")
|
| except Exception:
|
| detail_lines.append(f" - constraints: {constraints}")
|
|
|
| detail_block = "\n".join(detail_lines) if detail_lines else "(no parameter details)"
|
|
|
| desc_block = f"```\n{description}\n```" if description else "None"
|
|
|
| tools_list_str.append(
|
| f"{i + 1}. <tool name=\"{name}\">\n"
|
| f" Description:\n{desc_block}\n"
|
| f" Parameters summary: {params_summary}\n"
|
| f" Required parameters: {', '.join(required_list) if required_list else 'None'}\n"
|
| f" Parameter details:\n{detail_block}"
|
| )
|
|
|
| prompt_template = get_function_call_prompt_template(trigger_signal)
|
| prompt_content = prompt_template.replace("{tools_list}", "\n\n".join(tools_list_str))
|
|
|
| return prompt_content, trigger_signal
|
|
|
| def remove_think_blocks(text: str) -> str:
|
| """
|
| Temporarily remove all <think>...</think> blocks for XML parsing
|
| Supports nested think tags
|
| Note: This function is only used for temporary parsing and does not affect the original content returned to the user
|
| """
|
| while '<think>' in text and '</think>' in text:
|
| start_pos = text.find('<think>')
|
| if start_pos == -1:
|
| break
|
|
|
| pos = start_pos + 7
|
| depth = 1
|
|
|
| while pos < len(text) and depth > 0:
|
| if text[pos:pos+7] == '<think>':
|
| depth += 1
|
| pos += 7
|
| elif text[pos:pos+8] == '</think>':
|
| depth -= 1
|
| pos += 8
|
| else:
|
| pos += 1
|
|
|
| if depth == 0:
|
| text = text[:start_pos] + text[pos:]
|
| else:
|
| break
|
|
|
| return text
|
|
|
| class StreamingFunctionCallDetector:
|
| """Enhanced streaming function call detector, supports dynamic trigger signals, avoids misjudgment within <think> tags
|
|
|
| Core features:
|
| 1. Avoid triggering tool call detection within <think> blocks
|
| 2. Normally output <think> block content to the user
|
| 3. Supports nested think tags
|
| """
|
|
|
| def __init__(self, trigger_signal: str):
|
| self.trigger_signal = trigger_signal
|
| self.reset()
|
|
|
| def reset(self):
|
| self.content_buffer = ""
|
| self.state = "detecting"
|
| self.in_think_block = False
|
| self.think_depth = 0
|
| self.signal = self.trigger_signal
|
| self.signal_len = len(self.signal)
|
|
|
| def process_chunk(self, delta_content: str) -> tuple[bool, str]:
|
| """
|
| Process streaming content chunk
|
| Returns: (is_tool_call_detected, content_to_yield)
|
| """
|
| if not delta_content:
|
| return False, ""
|
|
|
| self.content_buffer += delta_content
|
| content_to_yield = ""
|
|
|
| if self.state == "tool_parsing":
|
| return False, ""
|
|
|
| if delta_content:
|
| logger.debug(f"🔧 Processing chunk: {repr(delta_content[:50])}{'...' if len(delta_content) > 50 else ''}, buffer length: {len(self.content_buffer)}, think state: {self.in_think_block}")
|
|
|
| i = 0
|
| while i < len(self.content_buffer):
|
| skip_chars = self._update_think_state(i)
|
| if skip_chars > 0:
|
| for j in range(skip_chars):
|
| if i + j < len(self.content_buffer):
|
| content_to_yield += self.content_buffer[i + j]
|
| i += skip_chars
|
| continue
|
|
|
| if not self.in_think_block and self._can_detect_signal_at(i):
|
| if self.content_buffer[i:i+self.signal_len] == self.signal:
|
| logger.debug(f"🔧 Improved detector: detected trigger signal in non-think block! Signal: {self.signal[:20]}...")
|
| logger.debug(f"🔧 Trigger signal position: {i}, think state: {self.in_think_block}, think depth: {self.think_depth}")
|
| self.state = "tool_parsing"
|
| self.content_buffer = self.content_buffer[i:]
|
| return True, content_to_yield
|
|
|
| remaining_len = len(self.content_buffer) - i
|
| if remaining_len < self.signal_len or remaining_len < 8:
|
| break
|
|
|
| content_to_yield += self.content_buffer[i]
|
| i += 1
|
|
|
| self.content_buffer = self.content_buffer[i:]
|
| return False, content_to_yield
|
|
|
| def _update_think_state(self, pos: int):
|
| """Update think tag state, supports nesting"""
|
| remaining = self.content_buffer[pos:]
|
|
|
| if remaining.startswith('<think>'):
|
| self.think_depth += 1
|
| self.in_think_block = True
|
| logger.debug(f"🔧 Entering think block, depth: {self.think_depth}")
|
| return 7
|
|
|
| elif remaining.startswith('</think>'):
|
| self.think_depth = max(0, self.think_depth - 1)
|
| self.in_think_block = self.think_depth > 0
|
| logger.debug(f"🔧 Exiting think block, depth: {self.think_depth}")
|
| return 8
|
|
|
| return 0
|
|
|
| def _can_detect_signal_at(self, pos: int) -> bool:
|
| """Check if signal can be detected at the specified position"""
|
| return (pos + self.signal_len <= len(self.content_buffer) and
|
| not self.in_think_block)
|
|
|
| def finalize(self) -> Optional[List[Dict[str, Any]]]:
|
| """Final processing when stream ends"""
|
| if self.state == "tool_parsing":
|
| return parse_function_calls_xml(self.content_buffer, self.trigger_signal)
|
| return None
|
|
|
| def parse_function_calls_xml(xml_string: str, trigger_signal: str) -> Optional[List[Dict[str, Any]]]:
|
| """
|
| Enhanced XML parsing function, supports dynamic trigger signals
|
| 1. Retain <think>...</think> blocks (they should be returned normally to the user)
|
| 2. Temporarily remove think blocks only when parsing function_calls to prevent think content from interfering with XML parsing
|
| 3. Find the last occurrence of the trigger signal
|
| 4. Start parsing function_calls from the last trigger signal
|
| """
|
| logger.debug(f"🔧 Improved parser starting processing, input length: {len(xml_string) if xml_string else 0}")
|
| logger.debug(f"🔧 Using trigger signal: {trigger_signal[:20]}...")
|
|
|
| if not xml_string or trigger_signal not in xml_string:
|
| logger.debug(f"🔧 Input is empty or doesn't contain trigger signal")
|
| return None
|
|
|
| cleaned_content = remove_think_blocks(xml_string)
|
| logger.debug(f"🔧 Content length after temporarily removing think blocks: {len(cleaned_content)}")
|
|
|
| signal_positions = []
|
| start_pos = 0
|
| while True:
|
| pos = cleaned_content.find(trigger_signal, start_pos)
|
| if pos == -1:
|
| break
|
| signal_positions.append(pos)
|
| start_pos = pos + 1
|
|
|
| if not signal_positions:
|
| logger.debug(f"🔧 No trigger signal found in cleaned content")
|
| return None
|
|
|
| logger.debug(f"🔧 Found {len(signal_positions)} trigger signal positions: {signal_positions}")
|
|
|
| last_signal_pos = signal_positions[-1]
|
| content_after_signal = cleaned_content[last_signal_pos:]
|
| logger.debug(f"🔧 Content starting from last trigger signal: {repr(content_after_signal[:100])}")
|
|
|
| calls_content_match = re.search(r"<function_calls>([\s\S]*?)</function_calls>", content_after_signal)
|
| if not calls_content_match:
|
| logger.debug(f"🔧 No function_calls tag found")
|
| return None
|
|
|
| calls_content = calls_content_match.group(1)
|
| logger.debug(f"🔧 function_calls content: {repr(calls_content)}")
|
|
|
| results = []
|
| call_blocks = re.findall(r"<function_call>([\s\S]*?)</function_call>", calls_content)
|
| logger.debug(f"🔧 Found {len(call_blocks)} function_call blocks")
|
|
|
| for i, block in enumerate(call_blocks):
|
| logger.debug(f"🔧 Processing function_call #{i+1}: {repr(block)}")
|
|
|
| tool_match = re.search(r"<tool>(.*?)</tool>", block)
|
| if not tool_match:
|
| logger.debug(f"🔧 No tool tag found in block #{i+1}")
|
| continue
|
|
|
| name = tool_match.group(1).strip()
|
| args = {}
|
|
|
| args_block_match = re.search(r"<args>([\s\S]*?)</args>", block)
|
| if args_block_match:
|
| args_content = args_block_match.group(1)
|
|
|
| arg_matches = re.findall(r"<([^\s>/]+)>([\s\S]*?)</\1>", args_content)
|
|
|
| def _coerce_value(v: str):
|
| try:
|
| return json.loads(v)
|
| except Exception:
|
| pass
|
| return v
|
|
|
| for k, v in arg_matches:
|
| args[k] = _coerce_value(v)
|
|
|
| result = {"name": name, "args": args}
|
| results.append(result)
|
| logger.debug(f"🔧 Added tool call: {result}")
|
|
|
| logger.debug(f"🔧 Final parsing result: {results}")
|
| return results if results else None
|
|
|
| def find_upstream(model_name: str) -> tuple[Dict[str, Any], str]:
|
| """Find upstream configuration by model name, handling aliases and passthrough mode."""
|
|
|
|
|
| if app_config.features.model_passthrough:
|
| logger.info("🔄 Model passthrough mode is active. Forwarding to 'openai' service.")
|
| openai_service = None
|
| for service in app_config.upstream_services:
|
| if service.name == "openai":
|
| openai_service = service.model_dump()
|
| break
|
|
|
| if openai_service:
|
| if not openai_service.get("api_key"):
|
| raise HTTPException(status_code=500, detail="Configuration error: API key not found for the 'openai' service in model passthrough mode.")
|
|
|
| return openai_service, model_name
|
| else:
|
| raise HTTPException(status_code=500, detail="Configuration error: 'model_passthrough' is enabled, but no upstream service named 'openai' was found.")
|
|
|
|
|
| chosen_model_entry = model_name
|
|
|
| if model_name in ALIAS_MAPPING:
|
| chosen_model_entry = random.choice(ALIAS_MAPPING[model_name])
|
| logger.info(f"🔄 Model alias '{model_name}' detected. Randomly selected '{chosen_model_entry}' for this request.")
|
|
|
| service = MODEL_TO_SERVICE_MAPPING.get(chosen_model_entry)
|
|
|
| if service:
|
| if not service.get("api_key"):
|
| raise HTTPException(status_code=500, detail=f"Model configuration error: API key not found for service '{service.get('name')}'.")
|
| else:
|
| logger.warning(f"⚠️ Model '{model_name}' not found in configuration, using default service")
|
| service = DEFAULT_SERVICE
|
| if not service.get("api_key"):
|
| raise HTTPException(status_code=500, detail="Service configuration error: Default API key not found.")
|
|
|
| actual_model_name = chosen_model_entry
|
| if ':' in chosen_model_entry:
|
| parts = chosen_model_entry.split(':', 1)
|
| if len(parts) == 2:
|
| _, actual_model_name = parts
|
|
|
| return service, actual_model_name
|
|
|
| app = FastAPI()
|
| http_client = httpx.AsyncClient()
|
|
|
| @app.middleware("http")
|
| async def debug_middleware(request: Request, call_next):
|
| """Middleware for debugging validation errors, does not log conversation content."""
|
| response = await call_next(request)
|
|
|
| if response.status_code == 422:
|
| logger.debug(f"🔍 Validation error detected for {request.method} {request.url.path}")
|
| logger.debug(f"🔍 Response status code: 422 (Pydantic validation failure)")
|
|
|
| return response
|
|
|
| @app.exception_handler(ValidationError)
|
| async def validation_exception_handler(request: Request, exc: ValidationError):
|
| """Handle Pydantic validation errors with detailed error information"""
|
| logger.error(f"❌ Pydantic validation error: {exc}")
|
| logger.error(f"❌ Request URL: {request.url}")
|
| logger.error(f"❌ Error details: {exc.errors()}")
|
|
|
| for error in exc.errors():
|
| logger.error(f"❌ Validation error location: {error.get('loc')}")
|
| logger.error(f"❌ Validation error message: {error.get('msg')}")
|
| logger.error(f"❌ Validation error type: {error.get('type')}")
|
|
|
| return JSONResponse(
|
| status_code=422,
|
| content={
|
| "error": {
|
| "message": "Invalid request format",
|
| "type": "invalid_request_error",
|
| "code": "invalid_request"
|
| }
|
| }
|
| )
|
|
|
| @app.exception_handler(Exception)
|
| async def general_exception_handler(request: Request, exc: Exception):
|
| """Handle all uncaught exceptions"""
|
| logger.error(f"❌ Unhandled exception: {exc}")
|
| logger.error(f"❌ Request URL: {request.url}")
|
| logger.error(f"❌ Exception type: {type(exc).__name__}")
|
| logger.error(f"❌ Error stack: {traceback.format_exc()}")
|
|
|
| return JSONResponse(
|
| status_code=500,
|
| content={
|
| "error": {
|
| "message": "Internal server error",
|
| "type": "server_error",
|
| "code": "internal_error"
|
| }
|
| }
|
| )
|
|
|
| async def verify_api_key(authorization: str = Header(...)):
|
| """Dependency: verify client API key"""
|
| client_key = authorization.replace("Bearer ", "")
|
| if app_config.features.key_passthrough:
|
|
|
| return client_key
|
| if client_key not in ALLOWED_CLIENT_KEYS:
|
| raise HTTPException(status_code=401, detail="Unauthorized")
|
| return client_key
|
|
|
| def preprocess_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
| """Preprocess messages, convert tool-type messages to AI-understandable format, return dictionary list to avoid Pydantic validation issues"""
|
| processed_messages = []
|
|
|
| for message in messages:
|
| if isinstance(message, dict):
|
| if message.get("role") == "tool":
|
| tool_call_id = message.get("tool_call_id")
|
| content = message.get("content")
|
|
|
| if tool_call_id and content:
|
| formatted_content = format_tool_result_for_ai(tool_call_id, content)
|
| processed_message = {
|
| "role": "user",
|
| "content": formatted_content
|
| }
|
| processed_messages.append(processed_message)
|
| logger.debug(f"🔧 Converted tool message to user message: tool_call_id={tool_call_id}")
|
| else:
|
| logger.debug(f"🔧 Skipped invalid tool message: tool_call_id={tool_call_id}, content={bool(content)}")
|
| elif message.get("role") == "assistant" and "tool_calls" in message and message["tool_calls"]:
|
| tool_calls = message.get("tool_calls", [])
|
| formatted_tool_calls_str = format_assistant_tool_calls_for_ai(tool_calls, GLOBAL_TRIGGER_SIGNAL)
|
|
|
|
|
| original_content = message.get("content") or ""
|
| final_content = f"{original_content}\n{formatted_tool_calls_str}".strip()
|
|
|
| processed_message = {
|
| "role": "assistant",
|
| "content": final_content
|
| }
|
|
|
| for key, value in message.items():
|
| if key not in ["role", "content", "tool_calls"]:
|
| processed_message[key] = value
|
|
|
| processed_messages.append(processed_message)
|
| logger.debug(f"🔧 Converted assistant tool_calls to content.")
|
|
|
| elif message.get("role") == "developer":
|
| if app_config.features.convert_developer_to_system:
|
| processed_message = message.copy()
|
| processed_message["role"] = "system"
|
| processed_messages.append(processed_message)
|
| logger.debug(f"🔧 Converted developer message to system message for better upstream compatibility")
|
| else:
|
| processed_messages.append(message)
|
| logger.debug(f"🔧 Keeping developer role unchanged (based on configuration)")
|
| else:
|
| processed_messages.append(message)
|
| else:
|
| processed_messages.append(message)
|
|
|
| return processed_messages
|
|
|
| @app.post("/v1/chat/completions")
|
| async def chat_completions(
|
| request: Request,
|
| body: ChatCompletionRequest,
|
| _api_key: str = Depends(verify_api_key)
|
| ):
|
| """Main chat completion endpoint, proxy and inject function calling capabilities."""
|
| try:
|
| logger.debug(f"🔧 Received request, model: {body.model}")
|
| logger.debug(f"🔧 Number of messages: {len(body.messages)}")
|
| logger.debug(f"🔧 Number of tools: {len(body.tools) if body.tools else 0}")
|
| logger.debug(f"🔧 Streaming: {body.stream}")
|
|
|
| upstream, actual_model = find_upstream(body.model)
|
| upstream_url = f"{upstream['base_url']}/chat/completions"
|
|
|
| logger.debug(f"🔧 Starting message preprocessing, original message count: {len(body.messages)}")
|
| processed_messages = preprocess_messages(body.messages)
|
| logger.debug(f"🔧 Preprocessing completed, processed message count: {len(processed_messages)}")
|
|
|
| if not validate_message_structure(processed_messages):
|
| logger.error(f"❌ Message structure validation failed, but continuing processing")
|
|
|
| request_body_dict = body.model_dump(exclude_unset=True)
|
| request_body_dict["model"] = actual_model
|
| request_body_dict["messages"] = processed_messages
|
| is_fc_enabled = app_config.features.enable_function_calling
|
| has_tools_in_request = bool(body.tools)
|
| has_function_call = is_fc_enabled and has_tools_in_request
|
|
|
| logger.debug(f"🔧 Request body constructed, message count: {len(processed_messages)}")
|
|
|
| except Exception as e:
|
| logger.error(f"❌ Request preprocessing failed: {str(e)}")
|
| logger.error(f"❌ Error type: {type(e).__name__}")
|
| if hasattr(app_config, 'debug') and app_config.debug:
|
| logger.error(f"❌ Error stack: {traceback.format_exc()}")
|
|
|
| return JSONResponse(
|
| status_code=422,
|
| content={
|
| "error": {
|
| "message": "Invalid request format",
|
| "type": "invalid_request_error",
|
| "code": "invalid_request"
|
| }
|
| }
|
| )
|
|
|
| if has_function_call:
|
| logger.debug(f"🔧 Using global trigger signal for this request: {GLOBAL_TRIGGER_SIGNAL}")
|
|
|
| function_prompt, _ = generate_function_prompt(body.tools, GLOBAL_TRIGGER_SIGNAL)
|
|
|
| tool_choice_prompt = safe_process_tool_choice(body.tool_choice)
|
| if tool_choice_prompt:
|
| function_prompt += tool_choice_prompt
|
|
|
| system_message = {"role": "system", "content": function_prompt}
|
| request_body_dict["messages"].insert(0, system_message)
|
|
|
| if "tools" in request_body_dict:
|
| del request_body_dict["tools"]
|
| if "tool_choice" in request_body_dict:
|
| del request_body_dict["tool_choice"]
|
|
|
| elif has_tools_in_request and not is_fc_enabled:
|
| logger.info(f"🔧 Function calling is disabled by configuration, ignoring 'tools' and 'tool_choice' in request.")
|
| if "tools" in request_body_dict:
|
| del request_body_dict["tools"]
|
| if "tool_choice" in request_body_dict:
|
| del request_body_dict["tool_choice"]
|
|
|
| headers = {
|
| "Content-Type": "application/json",
|
| "Authorization": f"Bearer {_api_key}" if app_config.features.key_passthrough else f"Bearer {upstream['api_key']}",
|
| "Accept": "application/json" if not body.stream else "text/event-stream"
|
| }
|
|
|
| logger.info(f"📝 Forwarding request to upstream: {upstream['name']}")
|
| logger.info(f"📝 Model: {request_body_dict.get('model', 'unknown')}, Messages: {len(request_body_dict.get('messages', []))}")
|
|
|
| if not body.stream:
|
| try:
|
| logger.debug(f"🔧 Sending upstream request to: {upstream_url}")
|
| logger.debug(f"🔧 has_function_call: {has_function_call}")
|
| logger.debug(f"🔧 Request body contains tools: {bool(body.tools)}")
|
|
|
| upstream_response = await http_client.post(
|
| upstream_url, json=request_body_dict, headers=headers, timeout=app_config.server.timeout
|
| )
|
| upstream_response.raise_for_status()
|
|
|
| response_json = upstream_response.json()
|
| logger.debug(f"🔧 Upstream response status code: {upstream_response.status_code}")
|
|
|
| if has_function_call:
|
| content = response_json["choices"][0]["message"]["content"]
|
| logger.debug(f"🔧 Complete response content: {repr(content)}")
|
|
|
| parsed_tools = parse_function_calls_xml(content, GLOBAL_TRIGGER_SIGNAL)
|
| logger.debug(f"🔧 XML parsing result: {parsed_tools}")
|
|
|
| if parsed_tools:
|
| logger.debug(f"🔧 Successfully parsed {len(parsed_tools)} tool calls")
|
| tool_calls = []
|
| for tool in parsed_tools:
|
| tool_call_id = f"call_{uuid.uuid4().hex}"
|
| store_tool_call_mapping(
|
| tool_call_id,
|
| tool["name"],
|
| tool["args"],
|
| f"Calling tool {tool['name']}"
|
| )
|
| tool_calls.append({
|
| "id": tool_call_id,
|
| "type": "function",
|
| "function": {
|
| "name": tool["name"],
|
| "arguments": json.dumps(tool["args"])
|
| }
|
| })
|
| logger.debug(f"🔧 Converted tool_calls: {tool_calls}")
|
|
|
| response_json["choices"][0]["message"] = {
|
| "role": "assistant",
|
| "content": None,
|
| "tool_calls": tool_calls,
|
| }
|
| response_json["choices"][0]["finish_reason"] = "tool_calls"
|
| logger.debug(f"🔧 Function call conversion completed")
|
| else:
|
| logger.debug(f"🔧 No tool calls detected, returning original content (including think blocks)")
|
| else:
|
| logger.debug(f"🔧 No function calls detected or conversion conditions not met")
|
|
|
| return JSONResponse(content=response_json)
|
|
|
| except httpx.HTTPStatusError as e:
|
| logger.error(f"❌ Upstream service response error: status_code={e.response.status_code}")
|
| logger.error(f"❌ Upstream error details: {e.response.text}")
|
|
|
| if e.response.status_code == 400:
|
| error_response = {
|
| "error": {
|
| "message": "Invalid request parameters",
|
| "type": "invalid_request_error",
|
| "code": "bad_request"
|
| }
|
| }
|
| elif e.response.status_code == 401:
|
| error_response = {
|
| "error": {
|
| "message": "Authentication failed",
|
| "type": "authentication_error",
|
| "code": "unauthorized"
|
| }
|
| }
|
| elif e.response.status_code == 403:
|
| error_response = {
|
| "error": {
|
| "message": "Access forbidden",
|
| "type": "permission_error",
|
| "code": "forbidden"
|
| }
|
| }
|
| elif e.response.status_code == 429:
|
| error_response = {
|
| "error": {
|
| "message": "Rate limit exceeded",
|
| "type": "rate_limit_error",
|
| "code": "rate_limit_exceeded"
|
| }
|
| }
|
| elif e.response.status_code >= 500:
|
| error_response = {
|
| "error": {
|
| "message": "Upstream service temporarily unavailable",
|
| "type": "service_error",
|
| "code": "upstream_error"
|
| }
|
| }
|
| else:
|
| error_response = {
|
| "error": {
|
| "message": "Request processing failed",
|
| "type": "api_error",
|
| "code": "unknown_error"
|
| }
|
| }
|
|
|
| return JSONResponse(content=error_response, status_code=e.response.status_code)
|
|
|
| else:
|
| return StreamingResponse(
|
| stream_proxy_with_fc_transform(upstream_url, request_body_dict, headers, body.model, has_function_call, GLOBAL_TRIGGER_SIGNAL),
|
| media_type="text/event-stream"
|
| )
|
|
|
| async def stream_proxy_with_fc_transform(url: str, body: dict, headers: dict, model: str, has_fc: bool, trigger_signal: str):
|
| """
|
| Enhanced streaming proxy, supports dynamic trigger signals, avoids misjudgment within think tags
|
| """
|
| logger.info(f"📝 Starting streaming response from: {url}")
|
| logger.info(f"📝 Function calling enabled: {has_fc}")
|
|
|
| if not has_fc or not trigger_signal:
|
| try:
|
| async with http_client.stream("POST", url, json=body, headers=headers, timeout=app_config.server.timeout) as response:
|
| async for chunk in response.aiter_bytes():
|
| yield chunk
|
| except httpx.RemoteProtocolError:
|
| logger.debug("🔧 Upstream closed connection prematurely, ending stream response")
|
| return
|
| return
|
|
|
| detector = StreamingFunctionCallDetector(trigger_signal)
|
|
|
| def _prepare_tool_calls(parsed_tools: List[Dict[str, Any]]):
|
| tool_calls = []
|
| for i, tool in enumerate(parsed_tools):
|
| tool_call_id = f"call_{uuid.uuid4().hex}"
|
| store_tool_call_mapping(
|
| tool_call_id,
|
| tool["name"],
|
| tool["args"],
|
| f"Calling tool {tool['name']}"
|
| )
|
| tool_calls.append({
|
| "index": i, "id": tool_call_id, "type": "function",
|
| "function": { "name": tool["name"], "arguments": json.dumps(tool["args"]) }
|
| })
|
| return tool_calls
|
|
|
| def _build_tool_call_sse_chunks(parsed_tools: List[Dict[str, Any]], model_id: str) -> List[str]:
|
| tool_calls = _prepare_tool_calls(parsed_tools)
|
| chunks: List[str] = []
|
|
|
| initial_chunk = {
|
| "id": f"chatcmpl-{uuid.uuid4().hex}", "object": "chat.completion.chunk",
|
| "created": int(os.path.getmtime(__file__)), "model": model_id,
|
| "choices": [{"index": 0, "delta": {"role": "assistant", "content": None, "tool_calls": tool_calls}, "finish_reason": None}],
|
| }
|
| chunks.append(f"data: {json.dumps(initial_chunk)}\n\n")
|
|
|
|
|
| final_chunk = {
|
| "id": f"chatcmpl-{uuid.uuid4().hex}", "object": "chat.completion.chunk",
|
| "created": int(os.path.getmtime(__file__)), "model": model_id,
|
| "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}],
|
| }
|
| chunks.append(f"data: {json.dumps(final_chunk)}\n\n")
|
| chunks.append("data: [DONE]\n\n")
|
| return chunks
|
|
|
| try:
|
| async with http_client.stream("POST", url, json=body, headers=headers, timeout=app_config.server.timeout) as response:
|
| if response.status_code != 200:
|
| error_content = await response.aread()
|
| logger.error(f"❌ Upstream service stream response error: status_code={response.status_code}")
|
| logger.error(f"❌ Upstream error details: {error_content.decode('utf-8', errors='ignore')}")
|
|
|
| if response.status_code == 401:
|
| error_message = "Authentication failed"
|
| elif response.status_code == 403:
|
| error_message = "Access forbidden"
|
| elif response.status_code == 429:
|
| error_message = "Rate limit exceeded"
|
| elif response.status_code >= 500:
|
| error_message = "Upstream service temporarily unavailable"
|
| else:
|
| error_message = "Request processing failed"
|
|
|
| error_chunk = {"error": {"message": error_message, "type": "upstream_error"}}
|
| yield f"data: {json.dumps(error_chunk)}\n\n"
|
| yield "data: [DONE]\n\n"
|
| return
|
|
|
| async for line in response.aiter_lines():
|
| if detector.state == "tool_parsing":
|
| if line.startswith("data:"):
|
| line_data = line[len("data: "):].strip()
|
| if line_data and line_data != "[DONE]":
|
| try:
|
| chunk_json = json.loads(line_data)
|
| delta_content = chunk_json.get("choices", [{}])[0].get("delta", {}).get("content", "") or ""
|
| detector.content_buffer += delta_content
|
|
|
| if "</function_calls>" in detector.content_buffer:
|
| logger.debug("🔧 Detected </function_calls> in stream, finalizing early...")
|
| parsed_tools = detector.finalize()
|
| if parsed_tools:
|
| logger.debug(f"🔧 Early finalize: parsed {len(parsed_tools)} tool calls")
|
| for sse in _build_tool_call_sse_chunks(parsed_tools, model):
|
| yield sse
|
| return
|
| else:
|
| logger.error("❌ Early finalize failed to parse tool calls")
|
| error_content = "Error: Detected tool use signal but failed to parse function call format"
|
| error_chunk = { "id": "error-chunk", "choices": [{"delta": {"content": error_content}}]}
|
| yield f"data: {json.dumps(error_chunk)}\n\n"
|
| yield "data: [DONE]\n\n"
|
| return
|
| except (json.JSONDecodeError, IndexError):
|
| pass
|
| continue
|
|
|
| if line.startswith("data:"):
|
| line_data = line[len("data: "):].strip()
|
| if not line_data or line_data == "[DONE]":
|
| continue
|
|
|
| try:
|
| chunk_json = json.loads(line_data)
|
| delta_content = chunk_json.get("choices", [{}])[0].get("delta", {}).get("content", "") or ""
|
|
|
| if delta_content:
|
| is_detected, content_to_yield = detector.process_chunk(delta_content)
|
|
|
| if content_to_yield:
|
| yield_chunk = {
|
| "id": f"chatcmpl-passthrough-{uuid.uuid4().hex}",
|
| "object": "chat.completion.chunk",
|
| "created": int(os.path.getmtime(__file__)),
|
| "model": model,
|
| "choices": [{"index": 0, "delta": {"content": content_to_yield}}]
|
| }
|
| yield f"data: {json.dumps(yield_chunk)}\n\n"
|
|
|
| if is_detected:
|
|
|
| continue
|
|
|
| except (json.JSONDecodeError, IndexError):
|
| yield line + "\n\n"
|
|
|
| except httpx.RequestError as e:
|
| logger.error(f"❌ Failed to connect to upstream service: {e}")
|
| logger.error(f"❌ Error type: {type(e).__name__}")
|
|
|
| error_message = "Failed to connect to upstream service"
|
| error_chunk = {"error": {"message": error_message, "type": "connection_error"}}
|
| yield f"data: {json.dumps(error_chunk)}\n\n"
|
| yield "data: [DONE]\n\n"
|
| return
|
|
|
| if detector.state == "tool_parsing":
|
| logger.debug(f"🔧 Stream ended, starting to parse tool call XML...")
|
| parsed_tools = detector.finalize()
|
| if parsed_tools:
|
| logger.debug(f"🔧 Streaming processing: Successfully parsed {len(parsed_tools)} tool calls")
|
| for sse in _build_tool_call_sse_chunks(parsed_tools, model):
|
| yield sse
|
| return
|
| else:
|
| logger.error(f"❌ Detected tool call signal but XML parsing failed, buffer content: {detector.content_buffer}")
|
| error_content = "Error: Detected tool use signal but failed to parse function call format"
|
| error_chunk = { "id": "error-chunk", "choices": [{"delta": {"content": error_content}}]}
|
| yield f"data: {json.dumps(error_chunk)}\n\n"
|
|
|
| elif detector.state == "detecting" and detector.content_buffer:
|
|
|
| final_yield_chunk = {
|
| "id": f"chatcmpl-finalflush-{uuid.uuid4().hex}", "object": "chat.completion.chunk",
|
| "created": int(os.path.getmtime(__file__)), "model": model,
|
| "choices": [{"index": 0, "delta": {"content": detector.content_buffer}}]
|
| }
|
| yield f"data: {json.dumps(final_yield_chunk)}\n\n"
|
|
|
| yield "data: [DONE]\n\n"
|
|
|
|
|
| @app.get("/")
|
| def read_root():
|
| return {
|
| "status": "OpenAI Function Call Middleware is running",
|
| "config": {
|
| "upstream_services_count": len(app_config.upstream_services),
|
| "client_keys_count": len(app_config.client_authentication.allowed_keys),
|
| "models_count": len(MODEL_TO_SERVICE_MAPPING),
|
| "features": {
|
| "function_calling": app_config.features.enable_function_calling,
|
| "log_level": app_config.features.log_level,
|
| "convert_developer_to_system": app_config.features.convert_developer_to_system,
|
| "random_trigger": True
|
| }
|
| }
|
| }
|
|
|
| @app.get("/v1/models")
|
| async def list_models(_api_key: str = Depends(verify_api_key)):
|
| """List all available models"""
|
| visible_models = set()
|
| for model_name in MODEL_TO_SERVICE_MAPPING.keys():
|
| if ':' in model_name:
|
| parts = model_name.split(':', 1)
|
| if len(parts) == 2:
|
| alias, _ = parts
|
| visible_models.add(alias)
|
| else:
|
| visible_models.add(model_name)
|
| else:
|
| visible_models.add(model_name)
|
|
|
| models = []
|
| for model_id in sorted(visible_models):
|
| models.append({
|
| "id": model_id,
|
| "object": "model",
|
| "created": 1677610602,
|
| "owned_by": "openai",
|
| "permission": [],
|
| "root": model_id,
|
| "parent": None
|
| })
|
|
|
| return {
|
| "object": "list",
|
| "data": models
|
| }
|
|
|
|
|
| def validate_message_structure(messages: List[Dict[str, Any]]) -> bool:
|
| """Validate if message structure meets requirements"""
|
| try:
|
| valid_roles = ["system", "user", "assistant", "tool"]
|
| if not app_config.features.convert_developer_to_system:
|
| valid_roles.append("developer")
|
|
|
| for i, msg in enumerate(messages):
|
| if "role" not in msg:
|
| logger.error(f"❌ Message {i} missing role field")
|
| return False
|
|
|
| if msg["role"] not in valid_roles:
|
| logger.error(f"❌ Invalid role value for message {i}: {msg['role']}")
|
| return False
|
|
|
| if msg["role"] == "tool":
|
| if "tool_call_id" not in msg:
|
| logger.error(f"❌ Tool message {i} missing tool_call_id field")
|
| return False
|
|
|
| content = msg.get("content")
|
| content_info = ""
|
| if content:
|
| if isinstance(content, str):
|
| content_info = f", content=text({len(content)} chars)"
|
| elif isinstance(content, list):
|
| text_parts = [item for item in content if isinstance(item, dict) and item.get('type') == 'text']
|
| image_parts = [item for item in content if isinstance(item, dict) and item.get('type') == 'image_url']
|
| content_info = f", content=multimodal(text={len(text_parts)}, images={len(image_parts)})"
|
| else:
|
| content_info = f", content={type(content).__name__}"
|
| else:
|
| content_info = ", content=empty"
|
|
|
| logger.debug(f"✅ Message {i} validation passed: role={msg['role']}{content_info}")
|
|
|
| logger.debug(f"✅ All messages validated successfully, total {len(messages)} messages")
|
| return True
|
| except Exception as e:
|
| logger.error(f"❌ Message validation exception: {e}")
|
| return False
|
|
|
| def safe_process_tool_choice(tool_choice) -> str:
|
| """Safely process tool_choice field to avoid type errors"""
|
| try:
|
| if tool_choice is None:
|
| return ""
|
|
|
| if isinstance(tool_choice, str):
|
| if tool_choice == "none":
|
| return "\n\n**IMPORTANT:** You are prohibited from using any tools in this round. Please respond like a normal chat assistant and answer the user's question directly."
|
| else:
|
| logger.debug(f"🔧 Unknown tool_choice string value: {tool_choice}")
|
| return ""
|
|
|
| elif hasattr(tool_choice, 'function') and hasattr(tool_choice.function, 'name'):
|
| required_tool_name = tool_choice.function.name
|
| return f"\n\n**IMPORTANT:** In this round, you must use ONLY the tool named `{required_tool_name}`. Generate the necessary parameters and output in the specified XML format."
|
|
|
| else:
|
| logger.debug(f"🔧 Unsupported tool_choice type: {type(tool_choice)}")
|
| return ""
|
|
|
| except Exception as e:
|
| logger.error(f"❌ Error processing tool_choice: {e}")
|
| return ""
|
|
|
| if __name__ == "__main__":
|
| import uvicorn
|
| logger.info(f"🚀 Starting server on {app_config.server.host}:{app_config.server.port}")
|
| logger.info(f"⏱️ Request timeout: {app_config.server.timeout} seconds")
|
|
|
| uvicorn.run(
|
| app,
|
| host=app_config.server.host,
|
| port=app_config.server.port,
|
| log_level=app_config.features.log_level.lower() if app_config.features.log_level != "DISABLED" else "critical"
|
| ) |
