#!/usr/bin/env python3 """ Zero-dependency OpenAI-compatible proxy for DeepSeek V4 Flash.
Author: g023 License: MIT
All client‑supplied model and generation parameters are ignored. The proxy always uses the model, max output tokens, and other settings defined in the global configuration (see --help and the constants below).
Optimisations:
- System prompt compression (auto-summarized via DeepSeek API; originals stored in ./pre_sys/, summaries cached in ./post_sys/)
- Markdown block deduplication (keeps only the latest occurrence full)
- Conversation summarisation triggers when token budget is exceeded
- Assistant reasoning is cached to avoid redundant re‑generation
- Inter‑message content fingerprinting & deduplication (Feature F-1)
- Removes repeated boilerplate segments (environment_info, userMemory, reminderInstructions, etc.) from user messages across conversation turns.
- Segments are hashed (SHA‑256), duplicates replaced with an empty string (or a minimal placeholder if the message becomes empty).
- Per‑conversation fingerprint storage with LRU eviction.
-
Reads from local file K.dat for API key if DEEPSEEK_API_KEY env var is not set.
-
just a proof of concept pet project. Do not expose this server to the internet. """
import argparse import collections import copy import hashlib import http.server import json import logging import os import re import signal import socketserver import sys import threading import time import urllib.error import urllib.request from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Tuple
DEEPSEEK_BASE = "https://api.deepseek.com" DEFAULT_MODEL = "deepseek-v4-flash" # "deepseek-v4-flash" "deepseek-v4-pro" # model that will always be used MAX_CACHE_SIZE = 500 # LRU cache for assistant reasoning MAX_CONTEXT = 128000 # tokens (context size) SUMMARY_RATIO = 0.8 # trigger summarisation at 80 % of MAX_CONTEXT SUMMARY_MODEL = DEFAULT_MODEL # model used for the summarisation call MAX_OUTPUT_TOKENS = 128000 # max tokens to generate (overrides client) THINKING_MODE = "auto" # "enabled", "disabled", or "auto" (default)
SAVE_PREPOST_MSGS = False # save pre/post message dumps to ./pre_msg/ and ./post_msg/ SAVE_PREPOST_SYSTEM = True # save original/summarized system prompts to ./pre_sys/ and ./post_sys/
SUMMARISE_MAX_RETRIES = 3 SUMMARISE_RETRY_BASE_SLEEP = 2.0 # seconds, doubled each attempt
MAX_FINGERPRINT_HISTORY = 100 # max number of segments stored per conversation
_BOILERPLATE_PATTERNS = { "environment_info": ("<environment_info>", "</environment_info>"), "workspace_info": ("<workspace_info>", "</workspace_info>"), "userMemory": ("<userMemory>", "</userMemory>"), "sessionMemory": ("<sessionMemory>", "</sessionMemory>"), "repoMemory": ("<repoMemory>", "</repoMemory>"), "context": ("<context>", "</context>"), "reminderInstructions": ("<reminderInstructions>", "</reminderInstructions>"), "additional_skills_reminder": ("<additional_skills_reminder>", "</additional_skills_reminder>"), "editorContext": ("<editorContext>", "</editorContext>"), }
_segment_fingerprints: Dict[str, Dict[str, Tuple[int, str]]] = {} _segment_fp_lock = threading.Lock()
DSEEK_KEY = "" if os.environ.get("DEEPSEEK_API_KEY"): DSEEK_KEY = os.environ["DEEPSEEK_API_KEY"] else: try: with open("K.dat", "r") as f: DSEEK_KEY = f.read().strip() except Exception: print("ERROR: DEEPSEEK_API_KEY environment variable not set and K.dat not found.", file=sys.stderr) sys.exit(1)
PRE_MSG_DIR = Path("./pre_msg") POST_MSG_DIR = Path("./post_msg")
def save_json_message(dir_path: Path, prefix: str, messages: list, msg_hash: str): """Save a message list as a tab-indented JSON file.""" try: timestamp = datetime.now().strftime("%Y%m%d%H%M%S_%f") filename = f"{timestamp}_{msg_hash}.json" filepath = dir_path / filename with open(filepath, "w", encoding="utf-8") as f: json.dump(messages, f, ensure_ascii=False, indent="\t") except Exception: logging.exception(f"Failed to save {prefix} message dump")
def _md5_of_messages(messages: list) -> str: """Short (8-char) MD5 hash of the messages list (stable).""" data = json.dumps(messages, sort_keys=True, ensure_ascii=False).encode() return hashlib.md5(data).hexdigest()[:8]
_CJK_RANGES = [ (0x4E00, 0x9FFF), # CJK Unified Ideographs (0x3400, 0x4DBF), # CJK Unified Ideographs Extension A (0x20000, 0x2A6DF), # Extension B (0x2A700, 0x2B73F), # Extension C (0x2B740, 0x2B81F), # Extension D (0x2B820, 0x2CEAF), # Extension E (0x2CEB0, 0x2EBEF), # Extension F (0x30000, 0x3134F), # Extension G (0x31350, 0x323AF), # Extension H ] _CJK_PUNCT = {0x3000, 0x3001, 0x3002, 0xFF0C, 0xFF0E, 0xFF1A, 0xFF1B, 0xFF01, 0xFF1F, 0x300C, 0x300D, 0x300E, 0x300F, 0x3010, 0x3011, 0x300A, 0x300B}
def _is_cjk(cp: int) -> bool: if cp in _CJK_PUNCT: return True for lo, hi in _CJK_RANGES: if lo <= cp <= hi: return True return False
def estimate_tokens(text: str) -> int: """ Token count heuristic: - For CJK‑heavy text (>50% CJK characters): 1 token per character. - Otherwise: 1 token per 3.5 characters (conservative for code/English). """ total = len(text) if total == 0: return 0 cjk_count = sum(1 for ch in text if _is_cjk(ord(ch))) if cjk_count / total > 0.5: return max(1, total) else: return max(1, int(total // 3.5))
class LRUCache: def init(self, maxsize: int): self.maxsize = maxsize self._cache = collections.OrderedDict() self._lock = threading.Lock()
def get(self, key: str):
with self._lock:
if key in self._cache:
self._cache.move_to_end(key)
return self._cache[key]
return None
def set(self, key: str, value: dict):
with self._lock:
if key in self._cache:
self._cache.move_to_end(key)
self._cache[key] = value
if len(self._cache) > self.maxsize:
self._cache.popitem(last=False)
_assistant_cache = LRUCache(MAX_CACHE_SIZE)
def _stable_hash(obj: Any) -> str: """Stable SHA256 hash of a JSON‑serialisable object (dict or list).""" return hashlib.sha256( json.dumps(obj, sort_keys=True, ensure_ascii=False).encode() ).hexdigest()
def _conv_hash(messages: List[dict]) -> str: """Hash of a message list – only fields that affect the conversation identity.""" important_keys = {"role", "content", "tool_calls", "name", "tool_call_id"} cleaned = [{k: v for k, v in m.items() if k in important_keys} for m in messages] return _stable_hash(cleaned)
PRE_SYS_DIR = Path("./pre_sys") POST_SYS_DIR = Path("./post_sys") _sys_lock = threading.Lock()
def _ensure_sys_dirs(): PRE_SYS_DIR.mkdir(parents=True, exist_ok=True) POST_SYS_DIR.mkdir(parents=True, exist_ok=True)
def _sys_original_path(sys_hash: str) -> Path: return PRE_SYS_DIR / f"{sys_hash}.txt"
def _sys_summary_path(sys_hash: str) -> Path: return POST_SYS_DIR / f"{sys_hash}.txt"
def load_summarized_prompt(sys_hash: str) -> Optional[str]: """Return the cached summarized prompt if it exists, else None.""" path = _sys_summary_path(sys_hash) if path.exists(): try: return path.read_text(encoding="utf-8") except Exception: logging.warning(f"Failed to read summarized prompt {sys_hash}") return None
def save_original_prompt(sys_hash: str, content: str): """Atomically save the original system prompt to disk (thread‑safe).""" path = _sys_original_path(sys_hash) if path.exists(): return # already saved tmp_path = path.with_suffix(".tmp") with _sys_lock: try: tmp_path.write_text(content, encoding="utf-8") tmp_path.rename(path) except Exception as e: logging.warning(f"Failed to save original prompt {sys_hash}: {e}")
def save_summarized_prompt(sys_hash: str, content: str): """Atomically write a summarized prompt to disk (thread‑safe).""" path = _sys_summary_path(sys_hash) tmp_path = path.with_suffix(".tmp") with _sys_lock: try: tmp_path.write_text(content, encoding="utf-8") tmp_path.rename(path) except Exception as e: logging.warning(f"Failed to write summarized prompt {sys_hash}: {e}")
def summarize_system_prompt(original: str, api_key: str) -> str: """Call DeepSeek to produce a concise summary of a system prompt.""" summary_prompt = ( "You are a prompt compression assistant. Summarize the following system prompt " "as concisely as possible while preserving ALL critical instructions, constraints, " "formatting rules, and behavioral guidelines. Remove redundancy, examples, and " "verbose explanations. Output ONLY the compressed prompt — no commentary.\n\n" f"{original}" ) payload = { "model": SUMMARY_MODEL, "messages": [{"role": "user", "content": summary_prompt}], "max_tokens": 2000, "temperature": 0.0, "thinking": {"type": "disabled"}, } headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } last_exc = None for attempt in range(1, SUMMARISE_MAX_RETRIES + 1): try: req = urllib.request.Request( f"{DEEPSEEK_BASE}/chat/completions", data=json.dumps(payload).encode(), headers=headers, method="POST", ) with urllib.request.urlopen(req, timeout=120) as resp: body = json.loads(resp.read().decode()) return body["choices"][0]["message"]["content"] except Exception as e: last_exc = e if attempt < SUMMARISE_MAX_RETRIES: sleep_time = SUMMARISE_RETRY_BASE_SLEEP * (2 ** (attempt - 1)) logging.warning("System prompt summarization attempt %d failed, retrying in %.1fs: %s", attempt, sleep_time, e) time.sleep(sleep_time) else: logging.error("System prompt summarization failed after %d attempts", SUMMARISE_MAX_RETRIES) raise RuntimeError(f"System prompt summarization failed: {last_exc}")
_FENCE_PATTERN = re.compile(r"(```|~~~)(\w*)\n(.*?)\1", re.DOTALL)
def deduplicate_markdown_blocks(messages: List[dict]) -> int: """ For each fenced code block, keep the last occurrence full; replace earlier occurrences with a placeholder. Modifies messages in-place. Returns the count of blocks replaced. """ block_info: Dict[str, int] = {} # hash -> latest global ID all_matches: List[Dict[str, Any]] = [] # per-match info replaced_count = 0
global_counter = 0
for msg_idx, msg in enumerate(messages):
content = msg.get("content", "")
if not isinstance(content, str):
continue
for match in _FENCE_PATTERN.finditer(content):
inner_text = match.group(3)
h = hashlib.sha256(inner_text.encode()).hexdigest()
all_matches.append({
"global_id": global_counter,
"msg_idx": msg_idx,
"start": match.start(3),
"end": match.end(3),
"hash": h,
"full_text": inner_text,
})
block_info[h] = global_counter
global_counter += 1
msg_matches: Dict[int, List[dict]] = {}
for m in all_matches:
msg_matches.setdefault(m["msg_idx"], []).append(m)
for msg_idx, matches in msg_matches.items():
msg = messages[msg_idx]
original = msg["content"]
matches_sorted = sorted(matches, key=lambda x: x["start"], reverse=True)
new_parts = []
prev_end = len(original)
for match in matches_sorted:
start, end = match["start"], match["end"]
is_last = block_info.get(match["hash"]) == match["global_id"]
if not is_last:
replacement = ".. (code omitted, see later version) .."
replaced_count += 1
else:
replacement = match["full_text"]
new_parts.append(original[end:prev_end])
new_parts.append(replacement)
prev_end = start
new_parts.append(original[:prev_end])
msg["content"] = "".join(reversed(new_parts))
return replaced_count
def _get_conversation_base_id(messages: List[dict]) -> str: """ Derive a stable conversation identifier from the first system message and the first user message (excluding boilerplate tags). This ID persists across requests of the same conversation, even as new messages are added. """ sys_content = "" for msg in messages: if msg.get("role") == "system" and isinstance(msg.get("content"), str): sys_content = msg["content"] break
user_content = ""
for msg in messages:
if msg.get("role") == "user" and isinstance(msg.get("content"), str):
content = msg["content"]
for open_tag, close_tag in _BOILERPLATE_PATTERNS.values():
pattern = re.escape(open_tag) + r".*?" + re.escape(close_tag)
content = re.sub(pattern, "", content, flags=re.DOTALL)
user_content = content.strip()
break
combined = f"{sys_content}\n{user_content}".strip()
if not combined:
return _conv_hash(messages)
return hashlib.sha256(combined.encode()).hexdigest()
def deduplicate_user_message_segments( messages: List[dict], conv_id: str ) -> Tuple[List[dict], int]: """ For each user message, identify boilerplate segments and omit those that are identical to previously-seen segments in this conversation. Modifies messages in-place. Returns (messages, segments_removed). If a user message becomes empty after removals, it is replaced with a minimal placeholder "(no new content)". """ global _segment_fingerprints
with _segment_fp_lock:
if conv_id not in _segment_fingerprints:
_segment_fingerprints[conv_id] = {}
history = _segment_fingerprints[conv_id]
segments_removed = 0
for msg in messages:
if msg.get("role") != "user":
continue
content = msg.get("content", "")
if not isinstance(content, str):
continue
new_content = content
for open_tag, close_tag in _BOILERPLATE_PATTERNS.values():
idx = 0
while True:
start = new_content.find(open_tag, idx)
if start == -1:
break
end = new_content.find(close_tag, start + len(open_tag))
if end == -1:
break
end += len(close_tag)
segment = new_content[start:end]
seg_hash = hashlib.sha256(segment.encode()).hexdigest()
if seg_hash in history:
new_content = new_content[:start] + new_content[end:]
segments_removed += 1
idx = start
else:
history[seg_hash] = (len(history), segment)
if len(history) > MAX_FINGERPRINT_HISTORY:
oldest = min(history.keys(), key=lambda k: history[k][0])
del history[oldest]
idx = end
new_content = new_content.strip()
if new_content == "":
new_content = "(no new content)"
msg["content"] = new_content
return messages, segments_removed
def _total_tokens(messages: List[dict]) -> int: """Estimate total token count for a list of messages.""" total = 0 for m in messages: content = m.get("content", "") if isinstance(content, str): total += estimate_tokens(content) for tc in m.get("tool_calls", []): total += estimate_tokens(json.dumps(tc.get("function", {}).get("arguments", ""))) return total
def _summarise_messages_with_retry(summarise_payload: dict, api_key: str) -> str: """ Call DeepSeek summarisation with retries (exponential backoff). Raises RuntimeError if all attempts fail. """ headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } last_exc = None for attempt in range(1, SUMMARISE_MAX_RETRIES + 1): try: req = urllib.request.Request( f"{DEEPSEEK_BASE}/chat/completions", data=json.dumps(summarise_payload).encode(), headers=headers, method="POST", ) with urllib.request.urlopen(req, timeout=120) as resp: body = json.loads(resp.read().decode()) return body["choices"][0]["message"]["content"] except Exception as e: last_exc = e if attempt < SUMMARISE_MAX_RETRIES: sleep_time = SUMMARISE_RETRY_BASE_SLEEP * (2 ** (attempt - 1)) logging.warning("Summarisation attempt %d failed, retrying in %.1fs: %s", attempt, sleep_time, e) time.sleep(sleep_time) else: logging.error("Summarisation failed after %d attempts", SUMMARISE_MAX_RETRIES) raise RuntimeError(f"Summarisation API call failed: {last_exc}")
def _merge_system_messages(messages: List[dict]) -> Tuple[Optional[str], List[dict]]: """Extract and merge all system messages into a single string. Returns the merged content (or None if none) and the remaining non‑system messages.""" systems = [m["content"] for m in messages if m["role"] == "system" and isinstance(m.get("content"), str)] others = [m for m in messages if m["role"] != "system"] merged = "\n".join(systems) if systems else None return merged, others
def maybe_summarize(messages: List[dict], api_key: str, max_context: int = MAX_CONTEXT, ratio: float = SUMMARY_RATIO) -> Tuple[List[dict], bool]: """ If total estimated tokens > max_context * ratio, summarise oldest messages (except system) and replace them with a condensed summary message. Handles multiple system messages by merging them. Returns a new message list (does not modify original) and a boolean indicating whether summarisation was performed. Guarantees at most ONE system message. """ threshold = int(max_context * ratio) total = _total_tokens(messages) if total <= threshold: merged_sys, non_sys = _merge_system_messages(messages) if merged_sys is not None: return [{"role": "system", "content": merged_sys}] + non_sys, False return messages, False
merged_sys, non_system = _merge_system_messages(messages)
if not non_system:
prefix = [{"role": "system", "content": merged_sys}] if merged_sys else []
return prefix, False
deficit = total - threshold
accumulated = 0
idx = 0
for i, msg in enumerate(non_system):
accumulated += estimate_tokens(msg.get("content", ""))
for tc in msg.get("tool_calls", []):
accumulated += estimate_tokens(json.dumps(tc.get("function", {}).get("arguments", "")))
if accumulated >= deficit and i >= len(non_system) // 2:
idx = i + 1 # summarise up to and including this message
break
else:
idx = max(1, len(non_system) // 2)
while idx < len(non_system):
msg = non_system[idx]
if msg.get("role") == "tool":
idx += 1
continue
if msg.get("role") == "assistant" and msg.get("tool_calls"):
if idx + 1 < len(non_system) and non_system[idx + 1].get("role") == "tool":
idx += 1
while idx < len(non_system) and non_system[idx].get("role") == "tool":
idx += 1
continue
break
to_summarise = non_system[:idx]
to_keep = non_system[idx:]
if not to_summarise:
prefix = [{"role": "system", "content": merged_sys}] if merged_sys else []
return prefix + non_system, False
summary_prompt = (
"Summarise the following conversation excerpt. "
"Retain all critical facts, decisions, and code fragments. "
"Be concise but complete.\n\n"
)
summarise_text = ""
for m in to_summarise:
role = m["role"]
content = m.get("content", "")
if isinstance(content, str):
summarise_text += f"[{role}]: {content}\n"
payload = {
"model": SUMMARY_MODEL,
"messages": [
{"role": "user", "content": summary_prompt + summarise_text}
],
"max_tokens": 1000,
"temperature": 0.0,
"thinking": {"type": "disabled"},
}
try:
summary = _summarise_messages_with_retry(payload, api_key)
except RuntimeError as e:
logging.warning(f"Summarisation failed, falling back to truncation: {e}")
summary = "[Earlier conversation truncated due to length]"
new_messages: List[dict] = []
if merged_sys:
new_sys_content = merged_sys + "\n\n[Earlier conversation summary]\n" + summary
new_messages.append({"role": "system", "content": new_sys_content})
else:
new_messages.append({"role": "system", "content": f"[Earlier conversation summary]\n{summary}"})
new_messages.extend(to_keep)
return new_messages, True
def cache_assistant_message(original_msgs: List[dict], assistant_msg: dict): """Cache assistant message (with reasoning) so it can be reused on subsequent turns.""" if not assistant_msg.get("tool_calls") and not assistant_msg.get("reasoning_content"): return prefix = [m.copy() for m in original_msgs] clean_asst = {k: v for k, v in assistant_msg.items() if k != "reasoning_content"} prefix.append(clean_asst) _assistant_cache.set(_conv_hash(prefix), assistant_msg)
def inject_reasoning(messages: List[dict]): """Look up cached reasoning_content for tool‑call assistant messages and inject it.""" for i, msg in enumerate(messages): if msg.get("role") != "assistant": continue if not msg.get("tool_calls"): continue if "reasoning_content" in msg: continue prefix = messages[:i+1] cached = _assistant_cache.get(_conv_hash(prefix)) if cached and "reasoning_content" in cached: msg["reasoning_content"] = cached["reasoning_content"]
def should_disable_thinking(messages: List[dict]) -> bool: """Return True if thinking should be disabled for the current conversation (i.e. there is a tool‑call assistant message WITHOUT reasoning, meaning the model doesn’t need to produce new reasoning).""" return any( m.get("role") == "assistant" and m.get("tool_calls") and "reasoning_content" not in m for m in messages )
def _make_deepseek_request(payload: dict, stream: bool) -> urllib.request.Request: headers = { "Authorization": f"Bearer {DSEEK_KEY}", "Content-Type": "application/json", "Accept": "text/event-stream" if stream else "application/json", } return urllib.request.Request( f"{DEEPSEEK_BASE}/chat/completions", data=json.dumps(payload).encode(), headers=headers, method="POST", )
def deepseek_nonstream(payload: dict) -> dict: """Perform a non‑streaming request. Raises RuntimeError on HTTP errors.""" req = _make_deepseek_request(payload, stream=False) try: with urllib.request.urlopen(req, timeout=600) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: error_body = e.read().decode() if e.fp else "" raise RuntimeError(f"DeepSeek HTTP {e.code}: {error_body}") from e
class StreamBuffer: def init(self): self.reasoning = "" self.content = "" self.tool_calls: Dict[int, dict] = {} self.finish_reason: Optional[str] = None self.usage: Optional[dict] = None
def process_chunk(self, chunk: dict) -> None:
for choice in chunk.get("choices", []):
delta = choice.get("delta", {})
if "reasoning_content" in delta:
rc = delta["reasoning_content"]
self.reasoning = "" if rc is None else self.reasoning + rc
if "content" in delta:
ct = delta["content"]
self.content = "" if ct is None else self.content + ct
for tc in delta.get("tool_calls", []):
idx = tc.get("index")
if idx is None:
continue
if idx not in self.tool_calls:
self.tool_calls[idx] = {
"id": tc.get("id", ""),
"type": tc.get("type", "function"),
"function": {
"name": "",
"arguments": "",
},
}
cur = self.tool_calls[idx]
tid = tc.get("id")
if tid is not None:
cur["id"] = tid
ttype = tc.get("type")
if ttype is not None:
cur["type"] = ttype
func_raw = tc.get("function")
func = func_raw if isinstance(func_raw, dict) else {}
name = func.get("name")
if name is not None:
cur["function"]["name"] = name
args = func.get("arguments")
if args is not None:
cur["function"]["arguments"] += args
if "message" in choice:
msg = choice["message"]
rc_msg = msg.get("reasoning_content")
ct_msg = msg.get("content")
if rc_msg is not None:
self.reasoning = rc_msg
if ct_msg is not None:
self.content = ct_msg
finish = choice.get("finish_reason")
if finish is not None:
self.finish_reason = finish
if "usage" in chunk:
self.usage = chunk["usage"]
def build_assistant_message(self) -> dict:
msg: Dict[str, Any] = {"role": "assistant"}
if self.reasoning:
msg["reasoning_content"] = self.reasoning
if self.content:
msg["content"] = self.content
if self.tool_calls:
msg["tool_calls"] = [self.tool_calls[k] for k in sorted(self.tool_calls)]
return msg
class ProxyHandler(http.server.BaseHTTPRequestHandler): def log_message(self, format, *args): logging.info("%s - %s", self.client_address[0], format % args)
def do_POST(self):
if self.path != "/v1/chat/completions":
self.send_error(404)
return
content_length = int(self.headers.get("Content-Length", 0))
if not content_length:
self.send_error(400, "Empty body")
return
body = self.rfile.read(content_length)
try:
client_req = json.loads(body)
except json.JSONDecodeError:
self.send_error(400, "Invalid JSON")
return
messages = client_req.get("messages", [])
if not isinstance(messages, list) or len(messages) == 0:
self.send_error(400, "Empty or invalid 'messages' array")
return
stream = client_req.get("stream", False)
original_messages = copy.deepcopy(messages)
original_msg_count = len(original_messages)
original_tokens = _total_tokens(original_messages)
msg_hash = _md5_of_messages(original_messages)
if SAVE_PREPOST_MSGS:
_save_json_message(PRE_MSG_DIR, "pre", original_messages, msg_hash)
replaced_blocks = deduplicate_markdown_blocks(messages)
conv_id = _get_conversation_base_id(original_messages)
messages, deduped_segments = deduplicate_user_message_segments(messages, conv_id)
messages, summarized = maybe_summarize(
messages,
api_key=DSEEK_KEY,
max_context=MAX_CONTEXT,
ratio=SUMMARY_RATIO,
)
compressed_prompts_found = 0
for msg in messages:
if msg["role"] == "system" and isinstance(msg.get("content"), str):
sys_content = msg["content"]
h = _stable_hash({"content": sys_content})
if SAVE_PREPOST_SYSTEM:
save_original_prompt(h, sys_content)
summarized_sys = load_summarized_prompt(h)
if summarized_sys:
logging.debug("Using summarized system prompt for hash %s", h[:12])
msg["content"] = summarized_sys
compressed_prompts_found += 1
else:
try:
logging.info("Generating summarized system prompt for hash %s", h[:12])
summarized_sys = summarize_system_prompt(sys_content, DSEEK_KEY)
if SAVE_PREPOST_SYSTEM:
save_summarized_prompt(h, summarized_sys)
msg["content"] = summarized_sys
compressed_prompts_found += 1
except RuntimeError as e:
logging.warning("Failed to summarize system prompt, using original: %s", e)
inject_reasoning(messages)
payload = dict(client_req)
payload["messages"] = messages
payload["model"] = DEFAULT_MODEL
payload["max_tokens"] = MAX_OUTPUT_TOKENS
if THINKING_MODE == "enabled":
payload["thinking"] = {"type": "enabled"}
elif THINKING_MODE == "disabled":
payload["thinking"] = {"type": "disabled"}
else:
payload["thinking"] = {
"type": "disabled" if should_disable_thinking(messages) else "enabled"
}
payload["stream"] = stream
final_msg_count = len(messages)
final_tokens = _total_tokens(messages)
compression_pct = (1 - final_tokens / original_tokens) * 100 if original_tokens > 0 else 0.0
if SAVE_PREPOST_MSGS:
_save_json_message(POST_MSG_DIR, "post", messages, msg_hash)
logging.info(
f"REQ {msg_hash} | msgs: {original_msg_count} → {final_msg_count} "
f"| tokens: {original_tokens} → {final_tokens} ({compression_pct:+.1f}%) "
f"| blocks dedup'd: {replaced_blocks} "
f"| dedup'd segments: {deduped_segments} "
f"| summarized: {'yes' if summarized else 'no'} "
f"| compressed prompts: {compressed_prompts_found} "
f"| stream: {stream} "
f"| thinking: {payload['thinking']['type']}"
)
try:
if stream:
self._handle_stream(payload, original_messages)
else:
self._handle_nonstream(payload, original_messages)
except RuntimeError as e:
logging.error("Upstream error: %s", e)
self.send_error(502, f"Upstream error: {e}")
except Exception:
logging.exception("Unexpected proxy error")
try:
self.send_error(500, "Internal proxy error")
except Exception:
pass
def _handle_nonstream(self, payload, original_msgs):
resp = deepseek_nonstream(payload)
choices = resp.get("choices")
if not choices or len(choices) == 0:
logging.error("DeepSeek returned empty choices array")
self.send_error(502, "Empty response from upstream")
return
choice = choices[0]
assistant_msg = choice.get("message", {}).copy()
cache_assistant_message(original_msgs, assistant_msg)
if "reasoning_content" in assistant_msg:
del assistant_msg["reasoning_content"]
choice["message"] = assistant_msg
body = json.dumps(resp).encode()
try:
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
except (BrokenPipeError, ConnectionResetError, OSError) as e:
logging.warning("Client disconnected while sending non‑stream response: %s", e)
def _handle_stream(self, payload, original_msgs):
req = _make_deepseek_request(payload, stream=True)
try:
with urllib.request.urlopen(req, timeout=600) as upstream:
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.end_headers()
buffer = StreamBuffer()
done = False
try:
for line in upstream:
line_str = line.decode() if isinstance(line, bytes) else line
if line_str.startswith("data:"):
data_part = line_str[5:].strip()
if data_part == "[DONE]":
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
done = True
break
try:
chunk = json.loads(data_part)
except json.JSONDecodeError:
self.wfile.write(line_str.encode() + b"\n")
self.wfile.flush()
continue
buffer.process_chunk(chunk)
for choice in chunk.get("choices", []):
if "delta" in choice:
choice["delta"].pop("reasoning_content", None)
if "message" in choice:
choice["message"].pop("reasoning_content", None)
self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
self.wfile.flush()
else:
self.wfile.write(line_str.encode() + b"\n")
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError, OSError) as e:
logging.warning("Client disconnected during stream: %s", e)
done = True
except Exception:
logging.exception("Unexpected error while streaming response")
done = True
if not done:
try:
self.wfile.write(b"data: [DONE]\n\n")
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError, OSError):
pass
assistant_msg = buffer.build_assistant_message()
if assistant_msg.get("tool_calls") or assistant_msg.get("reasoning_content"):
cache_assistant_message(original_msgs, assistant_msg)
except (urllib.error.HTTPError, urllib.error.URLError) as e:
logging.error("Upstream connection error: %s", e)
error_detail = ""
if isinstance(e, urllib.error.HTTPError):
try:
error_detail = e.read().decode()
except Exception:
pass
self.send_error(502, f"Upstream error: {error_detail}")
def main(): global MAX_CONTEXT, SUMMARY_RATIO, MAX_OUTPUT_TOKENS, THINKING_MODE
parser = argparse.ArgumentParser(description="DeepSeek V4 Flash OpenAI Proxy (globals forced)")
parser.add_argument("--port", type=int, default=8080, help="Listening port")
parser.add_argument("--host", default="0.0.0.0", help="Bind address")
parser.add_argument("--max-context", type=int, default=128000, help="Max context tokens")
parser.add_argument("--summarize-ratio", type=float, default=0.8,
help="Trigger summarisation when tokens exceed ratio * max-context")
parser.add_argument("--disable-compression", action="store_true",
help="Do not auto-summarize system prompts (use originals as-is)")
parser.add_argument("--max-output-tokens", type=int, default=128000,
help="Force this many max tokens for generation (overrides client)")
parser.add_argument("--thinking", choices=["enabled", "disabled", "auto"], default="auto",
help="Force thinking mode (default: auto)")
args = parser.parse_args()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
if not args.disable_compression and SAVE_PREPOST_SYSTEM:
_ensure_sys_dirs()
MAX_CONTEXT = args.max_context
SUMMARY_RATIO = args.summarize_ratio
MAX_OUTPUT_TOKENS = args.max_output_tokens
THINKING_MODE = args.thinking
if SAVE_PREPOST_MSGS:
PRE_MSG_DIR.mkdir(parents=True, exist_ok=True)
POST_MSG_DIR.mkdir(parents=True, exist_ok=True)
server = socketserver.ThreadingTCPServer((args.host, args.port), ProxyHandler)
server.daemon_threads = True
shutdown_lock = threading.Lock()
shutting_down = False
def _shutdown(signum, frame):
nonlocal shutting_down
with shutdown_lock:
if shutting_down:
return
shutting_down = True
logging.info("Received signal %s, shutting down.", signum)
threading.Thread(target=server.shutdown, daemon=True).start()
signal.signal(signal.SIGTERM, _shutdown)
signal.signal(signal.SIGINT, _shutdown)
logging.info(f"Proxy listening on {args.host}:{args.port}")
logging.info(f"Forced model: {DEFAULT_MODEL}, max_tokens: {MAX_OUTPUT_TOKENS}, thinking: {THINKING_MODE}")
logging.info(f"Pre/post message dumps: {PRE_MSG_DIR} / {POST_MSG_DIR}")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
logging.info("Server stopped.")
if name == "main": main()