Liang Jiaqing
be17b0adc5
- llmcore: _cc_system() builds payload system blocks; net 0 lines - half injects assets/cc_system_template.txt (CC fingerprint for relay gateways) - legacy fake_cc_system_prompt=True maps to simple
1144 lines
64 KiB
Python
1144 lines
64 KiB
Python
import os, json, re, time, requests, sys, threading, urllib3, base64, importlib, uuid, pathlib, copy
|
||
from datetime import datetime
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
_INFLIGHT = {} # thread ident -> live socket; lets abort() close it even before response headers arrive
|
||
_orig_conn_request = urllib3.connection.HTTPConnection.request
|
||
def _conn_request_hook(self, *a, **k): # after request() the socket is connected+sent; conn.sock may later be None'd by http.client
|
||
r = _orig_conn_request(self, *a, **k); _INFLIGHT[threading.get_ident()] = self.sock; return r
|
||
urllib3.connection.HTTPConnection.request = _conn_request_hook
|
||
_RESP_CACHE_KEY = str(uuid.uuid4()); _RESP_CODEX_KEY = str(uuid.uuid4())
|
||
_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||
if _ROOT not in sys.path: sys.path.append(_ROOT)
|
||
|
||
class MyKeys:
|
||
def __init__(self):
|
||
self.path = self.mtime = self.updated_at = None
|
||
self.data = {}
|
||
self.lock = threading.Lock()
|
||
|
||
def _load(self):
|
||
sys.modules.pop('mykey', None)
|
||
spec = importlib.util.find_spec('mykey')
|
||
if spec is not None:
|
||
path = spec.origin
|
||
mt = os.stat(path).st_mtime_ns
|
||
module = importlib.util.module_from_spec(spec)
|
||
sys.modules['mykey'] = module
|
||
# Compile source directly: pyc timestamps only have second precision.
|
||
try:
|
||
exec(compile(spec.loader.get_source('mykey'), path, 'exec'), module.__dict__)
|
||
except BaseException:
|
||
sys.modules.pop('mykey', None)
|
||
raise
|
||
data = {k: v for k, v in vars(module).items() if not k.startswith('_')}
|
||
else:
|
||
path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'mykey.json')
|
||
mt = os.stat(path).st_mtime_ns
|
||
with open(path, encoding='utf-8') as f: data = json.load(f)
|
||
if isinstance(data, dict) and 'remote_url' in data:
|
||
data = requests.get(data['remote_url'], timeout=10).json()
|
||
if not isinstance(data, dict): raise ValueError('mykey configuration must be a dict')
|
||
if os.stat(path).st_mtime_ns != mt: raise RuntimeError('mykey changed while loading')
|
||
return data, path, mt
|
||
|
||
def reload(self, force=False):
|
||
with self.lock:
|
||
try:
|
||
mt = os.stat(self.path).st_mtime_ns if self.path else None
|
||
if not force and self.updated_at is not None and mt == self.mtime:
|
||
return self.data, self.updated_at, False
|
||
data, path, mt = self._load()
|
||
self.data, self.path, self.mtime = data, path, mt
|
||
self.updated_at = max(time.monotonic_ns(), (self.updated_at or 0) + 1)
|
||
print(f'[Info] Load mykeys from {self.path}')
|
||
return self.data, self.updated_at, True
|
||
except Exception:
|
||
return self.data, self.updated_at, False
|
||
|
||
def snapshot(self):
|
||
data, updated_at, _ = self.reload()
|
||
return data, updated_at
|
||
|
||
|
||
mykeys_store = MyKeys()
|
||
|
||
def reload_mykeys():
|
||
data, _, changed = mykeys_store.reload()
|
||
return data, changed
|
||
|
||
def _load_mykeys():
|
||
with mykeys_store.lock:
|
||
return mykeys_store._load()[0]
|
||
|
||
def __getattr__(name):
|
||
if name == 'mykeys': return reload_mykeys()[0]
|
||
raise AttributeError(f"module 'llmcore' has no attribute {name}")
|
||
|
||
def compress_history_tags(messages, keep_recent=10, max_len=800, force=False, interval=5):
|
||
"""Compress <thinking>/<tool_use>/<tool_result> tags in older messages to save tokens."""
|
||
compress_history_tags._cd = getattr(compress_history_tags, '_cd', 0) + 1
|
||
if force: compress_history_tags._cd = 0
|
||
if compress_history_tags._cd % interval != 0: return messages
|
||
_before = sum(len(json.dumps(m, ensure_ascii=False)) for m in messages)
|
||
_pats = {tag: re.compile(rf'(<{tag}>)([\s\S]*?)(</{tag}>)') for tag in ('thinking', 'think', 'tool_use', 'tool_result')}
|
||
_hist_pat = re.compile(r'<(history|key_info|earlier_context|global_memory)>[\s\S]*?</\1>')
|
||
def _trunc_str(s): return s[:max_len//2] + '\n...[Truncated]...\n' + s[-max_len//2:] if isinstance(s, str) and len(s) > max_len else s
|
||
def _trunc(text):
|
||
text = _hist_pat.sub(lambda m: f'<{m.group(1)}>[...]</{m.group(1)}>', text)
|
||
for pat in _pats.values(): text = pat.sub(lambda m: m.group(1) + _trunc_str(m.group(2)) + m.group(3), text)
|
||
return text
|
||
for i, msg in enumerate(messages):
|
||
if i >= len(messages) - keep_recent: break
|
||
c = msg['content']
|
||
if isinstance(c, str): msg['content'] = _trunc(c)
|
||
elif isinstance(c, list):
|
||
for b in c:
|
||
if not isinstance(b, dict): continue
|
||
t = b.get('type')
|
||
if t == 'text' and isinstance(b.get('text'), str): b['text'] = _trunc(b['text'])
|
||
elif t == 'thinking' and isinstance(b.get('thinking'), str): b['thinking'] = _trunc_str(b['thinking'])
|
||
elif t == 'tool_result':
|
||
tc = b.get('content')
|
||
if isinstance(tc, str): b['content'] = _trunc_str(tc)
|
||
elif isinstance(tc, list):
|
||
for sub in tc:
|
||
if isinstance(sub, dict) and sub.get('type') == 'text': sub['text'] = _trunc_str(sub.get('text'))
|
||
elif t == 'tool_use' and isinstance(b.get('input'), dict):
|
||
for k, v in b['input'].items(): b['input'][k] = _trunc_str(v)
|
||
print(f"[Cut] {_before} -> {sum(len(json.dumps(m, ensure_ascii=False)) for m in messages)}")
|
||
return messages
|
||
|
||
def _sanitize_leading_user_msg(msg):
|
||
"""把 user 消息里的 tool_result 块改写成纯文本，避免孤立引用。
|
||
history 统一使用 Claude content-block 格式：content 是 list of blocks。"""
|
||
msg = dict(msg) # 浅拷贝外层 dict
|
||
content = msg.get('content')
|
||
if not isinstance(content, list): return msg
|
||
texts = []
|
||
for block in content:
|
||
if not isinstance(block, dict): continue
|
||
if block.get('type') == 'tool_result':
|
||
c = block.get('content', '')
|
||
if isinstance(c, list): texts.extend(b.get('text', '') for b in c if isinstance(b, dict))
|
||
else: texts.append(str(c))
|
||
elif block.get('type') == 'text': texts.append(block.get('text', ''))
|
||
msg['content'] = [{"type": "text", "text": '\n'.join(t for t in texts if t)}]
|
||
return msg
|
||
|
||
_oldprint = print
|
||
def safeprint(*argv):
|
||
try: _oldprint(*argv)
|
||
except OSError: pass
|
||
print = safeprint
|
||
|
||
STATS = {}
|
||
|
||
def trim_messages_history(history, sess):
|
||
cap = sess.context_win * 3
|
||
target = int(cap * getattr(sess, 'trim_keep_rate', 0.6))
|
||
kp = sess.trim_keep_prefix
|
||
def cost(ms): return sum(len(json.dumps(m, ensure_ascii=False)) for m in ms)
|
||
compress_history_tags(history, interval=getattr(sess, 'cut_msg_interval', 7))
|
||
STATS.update(ctx=(c := cost(history)), msgs=len(history)); print(f'[Debug] Current context: {c} chars, {len(history)} messages.')
|
||
if c <= cap: return
|
||
compress_history_tags(history, keep_recent=4, force=True)
|
||
if cost(history) <= target: return
|
||
pre, post = history[:kp], history[kp:]; costs = [len(json.dumps(m, ensure_ascii=False)) for m in post]; c = cost(pre) + sum(costs); i = 0
|
||
while len(post) - i > 9 and c > target:
|
||
c -= costs[i]; i += 1
|
||
while i < len(post) and post[i].get('role') != 'user': c -= costs[i]; i += 1
|
||
if i < len(post): old = costs[i]; post[i] = _sanitize_leading_user_msg(post[i]); costs[i] = len(json.dumps(post[i], ensure_ascii=False)); c += costs[i] - old
|
||
post = post[i:]
|
||
if kp and pre:
|
||
m = pre[-1]
|
||
if m.get('role') == 'assistant' and isinstance(m.get('content'), list):
|
||
m['content'] = [b for b in m['content'] if not (isinstance(b, dict) and b.get('type') == 'tool_use')] or [{"type": "text", "text": "..."}]
|
||
_d = lambda: [{"type": "text", "text": "..."}]
|
||
gap = [{"role": "assistant", "content": _d()}] if m.get('role') == 'user' else [{"role": "user", "content": _d()}, {"role": "assistant", "content": _d()}]
|
||
history[:] = pre + gap + post
|
||
else: history[:] = pre + post
|
||
STATS.update(ctx=(c := cost(history)), msgs=len(history)); print(f'[Debug] Trimmed context, current: {c} chars, {len(history)} messages.')
|
||
|
||
def auto_make_url(base, path):
|
||
b, p = base.rstrip('/'), path.strip('/')
|
||
if b.endswith('$'): return b[:-1].rstrip('/')
|
||
if b.endswith(p): return b
|
||
return f"{b}/{p}" if re.search(r'/v\d+(/|$)', b) else f"{b}/v1/{p}"
|
||
|
||
def _parse_claude_json(data):
|
||
if data.get("stop_reason") == "refusal":
|
||
err = "[Error: Claude refusal]"
|
||
yield err
|
||
return [{"type": "text", "text": err}]
|
||
content_blocks = data.get("content", [])
|
||
_record_usage(data.get("usage", {}), "messages")
|
||
for b in content_blocks:
|
||
if b.get("type") == "text": yield b.get("text", "")
|
||
elif b.get("type") == "thinking": yield ""
|
||
return content_blocks
|
||
|
||
def _raise_if_retryable_overload(emsg):
|
||
"""HTTP 200 SSE/body overload → ConnectionError so _stream_with_retry can backoff."""
|
||
if emsg and re.search(r'concurrency|retry later|overloaded|rate.?limit', emsg, re.I):
|
||
raise requests.ConnectionError(emsg)
|
||
|
||
def _parse_claude_sse(resp_lines):
|
||
"""Parse Anthropic SSE stream. Yields text chunks, returns list[content_block]."""
|
||
content_blocks = []; current_block = None; tool_json_buf = ""
|
||
stop_reason = None; got_message_stop = False; warn = None
|
||
for line in resp_lines:
|
||
if not line: continue
|
||
line = line.decode('utf-8') if isinstance(line, bytes) else line
|
||
if not line.startswith("data:"): continue
|
||
data_str = line[5:].lstrip()
|
||
if data_str == "[DONE]": break
|
||
try: evt = json.loads(data_str)
|
||
except Exception as e:
|
||
print(f"[SSE] JSON parse error: {e}, line: {data_str[:200]}")
|
||
continue
|
||
evt_type = evt.get("type", "")
|
||
if evt_type == "message_start":
|
||
usage = evt.get("message", {}).get("usage", {})
|
||
_record_usage(usage, "messages")
|
||
elif evt_type == "content_block_start":
|
||
block = evt.get("content_block", {})
|
||
if block.get("type") == "text": current_block = {"type": "text", "text": ""}
|
||
elif block.get("type") == "thinking": current_block = {"type": "thinking", "thinking": "", "signature": ""}
|
||
elif block.get("type") == "tool_use":
|
||
current_block = {"type": "tool_use", "id": block.get("id", ""), "name": block.get("name", ""), "input": {}}
|
||
tool_json_buf = ""
|
||
elif evt_type == "content_block_delta":
|
||
delta = evt.get("delta", {})
|
||
if delta.get("type") == "text_delta":
|
||
text = delta.get("text", "")
|
||
if current_block and current_block.get("type") == "text": current_block["text"] += text
|
||
if text: yield text
|
||
elif delta.get("type") == "thinking_delta":
|
||
thinking = delta.get("thinking", "")
|
||
if current_block and current_block.get("type") == "thinking": current_block["thinking"] += thinking
|
||
if thinking: yield thinking
|
||
elif delta.get("type") == "signature_delta":
|
||
if current_block and current_block.get("type") == "thinking":
|
||
current_block["signature"] = current_block.get("signature", "") + delta.get("signature", "")
|
||
elif delta.get("type") == "input_json_delta": tool_json_buf += delta.get("partial_json", "")
|
||
elif evt_type == "content_block_stop":
|
||
if current_block:
|
||
if current_block["type"] == "tool_use":
|
||
try: current_block["input"] = json.loads(tool_json_buf) if tool_json_buf else {}
|
||
except: current_block["input"] = {"_raw": tool_json_buf}
|
||
content_blocks.append(current_block)
|
||
current_block = None
|
||
elif evt_type == "message_delta":
|
||
delta = evt.get("delta", {})
|
||
stop_reason = delta.get("stop_reason", stop_reason)
|
||
out_usage = evt.get("usage", {})
|
||
out_tokens = out_usage.get("output_tokens", 0)
|
||
if out_tokens: STATS['out'] = out_tokens; print(f"[Output] tokens={out_tokens} stop_reason={stop_reason}")
|
||
elif evt_type == "message_stop": got_message_stop = True
|
||
elif evt_type == "error":
|
||
err = evt.get("error", {})
|
||
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
|
||
_raise_if_retryable_overload(emsg) # 走 _stream_with_retry，避免落到 ga 应用层
|
||
warn = f"\n\n!!!Error: SSE {emsg}"; break
|
||
if not warn:
|
||
if not got_message_stop and not stop_reason: warn = "\n\n[!!! 流异常中断，未收到完整响应 !!!]"
|
||
elif stop_reason == "max_tokens": warn = "\n\n[!!! Response truncated: max_tokens !!!]"
|
||
elif stop_reason == "refusal": warn = "\n\n[Error: Claude refusal]"
|
||
if current_block:
|
||
if current_block["type"] == "tool_use":
|
||
try: current_block["input"] = json.loads(tool_json_buf) if tool_json_buf else {}
|
||
except: current_block["input"] = {"_raw": tool_json_buf}
|
||
content_blocks.append(current_block); current_block = None
|
||
if warn:
|
||
print(f"[WARN] {warn.strip()}")
|
||
insert_at = next((i for i,b in enumerate(content_blocks) if b.get("type") == "tool_use"), len(content_blocks))
|
||
content_blocks.insert(insert_at, {"type": "text", "text": warn}); yield warn
|
||
return content_blocks
|
||
|
||
def _try_parse_tool_args(raw):
|
||
"""Parse tool args string; split concatenated JSON objects like {..}{..} if needed.
|
||
Returns list of parsed dicts."""
|
||
if not raw: return [{}]
|
||
try: return [json.loads(raw)]
|
||
except: pass
|
||
parts = re.split(r'(?<=\})(?=\{)', raw)
|
||
if len(parts) > 1:
|
||
parsed = []
|
||
for p in parts:
|
||
try: parsed.append(json.loads(p))
|
||
except: return [{"_raw": raw}]
|
||
return parsed
|
||
return [{"_raw": raw}]
|
||
|
||
def _parse_openai_sse(resp_lines, api_mode="chat_completions"):
|
||
"""Parse OpenAI SSE stream (chat_completions or responses API).
|
||
Yields text chunks, returns list[content_block].
|
||
content_block: {type:'text', text:str} | {type:'tool_use', id:str, name:str, input:dict}
|
||
"""
|
||
content_text = ""
|
||
if api_mode == "responses":
|
||
seen_delta = False; fc_buf = {}; current_fc_idx = None; reasoning_text = ""
|
||
for line in resp_lines:
|
||
if not line: continue
|
||
line = line.decode('utf-8', errors='replace') if isinstance(line, bytes) else line
|
||
if not line.startswith("data:"): continue
|
||
data_str = line[5:].lstrip()
|
||
if data_str == "[DONE]": break
|
||
try: evt = json.loads(data_str)
|
||
except: continue
|
||
etype = evt.get("type", "")
|
||
if etype == "response.output_text.delta":
|
||
delta = evt.get("delta", "")
|
||
if delta: seen_delta = True; content_text += delta; yield delta
|
||
elif etype == "response.output_text.done" and not seen_delta:
|
||
text = evt.get("text", "")
|
||
if text: content_text += text; yield text
|
||
elif etype == "response.reasoning_text.delta":
|
||
delta = evt.get("delta", "")
|
||
if delta: reasoning_text += delta
|
||
elif etype == "response.reasoning_text.done":
|
||
text = evt.get("text", "")
|
||
if text: reasoning_text = text
|
||
elif etype == "response.output_item.added":
|
||
item = evt.get("item", {})
|
||
if item.get("type") == "function_call":
|
||
idx = evt.get("output_index", 0)
|
||
fc_buf[idx] = {"id": item.get("call_id", item.get("id", "")), "name": item.get("name", ""), "args": ""}
|
||
current_fc_idx = idx
|
||
elif etype == "response.function_call_arguments.delta":
|
||
idx = evt.get("output_index", current_fc_idx or 0)
|
||
if idx in fc_buf: fc_buf[idx]["args"] += evt.get("delta", "")
|
||
elif etype == "response.function_call_arguments.done":
|
||
idx = evt.get("output_index", current_fc_idx or 0)
|
||
if idx in fc_buf: fc_buf[idx]["args"] = evt.get("arguments", fc_buf[idx]["args"])
|
||
elif etype == "error":
|
||
err = evt.get("error", {})
|
||
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
|
||
_raise_if_retryable_overload(emsg)
|
||
if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}"
|
||
break
|
||
elif etype == "response.completed":
|
||
usage = evt.get("response", {}).get("usage", {})
|
||
_record_usage(usage, api_mode)
|
||
break
|
||
elif etype == "response.incomplete":
|
||
# DeepSeek/OpenAI responses stream may end here (no [DONE]); treat as valid terminal,
|
||
# record usage and surface a truncation marker instead of an empty-response retry storm.
|
||
usage = (evt.get("response") or {}).get("usage", {})
|
||
_record_usage(usage, api_mode)
|
||
reason = ((evt.get("response") or {}).get("incomplete_details") or {}).get("reason", "") or "unknown"
|
||
if not content_text and not fc_buf:
|
||
marker = f"[!!! output truncated: {reason}]"
|
||
content_text += marker; yield marker
|
||
break
|
||
elif etype == "response.failed":
|
||
usage = (evt.get("response") or {}).get("usage", {})
|
||
_record_usage(usage, api_mode)
|
||
err = ((evt.get("response") or {}).get("error") or {})
|
||
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
|
||
_raise_if_retryable_overload(emsg)
|
||
if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}"
|
||
break
|
||
blocks = []
|
||
if reasoning_text: blocks.append({"type": "thinking", "thinking": reasoning_text})
|
||
if content_text: blocks.append({"type": "text", "text": content_text})
|
||
for idx in sorted(fc_buf):
|
||
fc = fc_buf[idx]
|
||
inps = _try_parse_tool_args(fc["args"])
|
||
for i, inp in enumerate(inps):
|
||
bid = fc["id"] or ''
|
||
if len(inps) > 1: bid = f"{bid}_{i}" if bid else f"split_{i}"
|
||
blocks.append({"type": "tool_use", "id": bid, "name": fc["name"], "input": inp})
|
||
return blocks
|
||
else:
|
||
tc_buf = {} # index -> {id, name, args}
|
||
reasoning_text = ""
|
||
for line in resp_lines:
|
||
if not line: continue
|
||
line = line.decode('utf-8', errors='replace') if isinstance(line, bytes) else line
|
||
if not line.startswith("data:"): continue
|
||
data_str = line[5:].lstrip()
|
||
if data_str == "[DONE]": break
|
||
try: evt = json.loads(data_str)
|
||
except: continue
|
||
ch = (evt.get("choices") or [{}])[0]
|
||
delta = ch.get("delta") or {}
|
||
if rc := delta.get("reasoning_content") or delta.get("reasoning", ""):
|
||
reasoning_text += rc; yield rc
|
||
if delta.get("content"):
|
||
text = delta["content"]; content_text += text; yield text
|
||
for tc in (delta.get("tool_calls") or []):
|
||
idx = tc.get("index", 0)
|
||
has_name = bool(tc.get("function", {}).get("name"))
|
||
if idx not in tc_buf:
|
||
if has_name or not tc_buf: tc_buf[idx] = {"id": tc.get("id") or '', "name": "", "args": ""}
|
||
else: idx = max(tc_buf)
|
||
if has_name: tc_buf[idx]["name"] = tc["function"]["name"]
|
||
if tc.get("function", {}).get("arguments"): tc_buf[idx]["args"] += tc["function"]["arguments"]
|
||
if tc.get("id") and not tc_buf[idx]["id"]: tc_buf[idx]["id"] = tc["id"]
|
||
usage = evt.get("usage")
|
||
if usage: _record_usage(usage, api_mode)
|
||
blocks = []
|
||
if reasoning_text: blocks.append({"type": "thinking", "thinking": reasoning_text})
|
||
if content_text: blocks.append({"type": "text", "text": content_text})
|
||
for idx in sorted(tc_buf):
|
||
tc = tc_buf[idx]
|
||
inps = _try_parse_tool_args(tc["args"])
|
||
for i, inp in enumerate(inps):
|
||
bid = tc["id"] or ''
|
||
if len(inps) > 1: bid = f"{bid}_{i}" if bid else f"split_{i}"
|
||
blocks.append({"type": "tool_use", "id": bid, "name": tc["name"], "input": inp})
|
||
return blocks
|
||
|
||
def _record_usage(usage, api_mode):
|
||
if not usage: return
|
||
# 上游常显式给 null；dict.get(k, 0) 遇 null 仍返回 None，后续加法会 TypeError
|
||
def _i(v, default=0):
|
||
try:
|
||
return default if v is None else int(v)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
if api_mode == 'responses':
|
||
cached = _i((usage.get("input_tokens_details") or {}).get("cached_tokens"))
|
||
inp = _i(usage.get("input_tokens")); out = _i(usage.get("output_tokens"))
|
||
print(f"[Cache] input={inp} cached={cached}")
|
||
if out: print(f"[Output] tokens={out}")
|
||
elif api_mode == 'chat_completions':
|
||
cached = _i((usage.get("prompt_tokens_details") or {}).get("cached_tokens"))
|
||
inp = _i(usage.get("prompt_tokens")); out = _i(usage.get("completion_tokens"))
|
||
print(f"[Cache] input={inp} cached={cached}")
|
||
if out: print(f"[Output] tokens={out}")
|
||
elif api_mode == 'messages':
|
||
ci, cr, raw_inp = _i(usage.get("cache_creation_input_tokens")), _i(usage.get("cache_read_input_tokens")), _i(usage.get("input_tokens"))
|
||
inp, cached, out = raw_inp + ci + cr, cr, 0
|
||
print(f"[Cache] input={raw_inp} creation={ci} read={cr}")
|
||
else: return
|
||
STATS.update(inp=inp, cached=cached, out=out)
|
||
|
||
def _parse_openai_json(data, api_mode="chat_completions"):
|
||
blocks = []
|
||
if api_mode == "responses":
|
||
_record_usage(data.get("usage") or {}, api_mode)
|
||
for item in (data.get("output") or []):
|
||
if item.get("type") == "message":
|
||
for p in (item.get("content") or []):
|
||
if p.get("type") in ("output_text", "text") and p.get("text"):
|
||
blocks.append({"type": "text", "text": p["text"]}); yield p["text"]
|
||
elif item.get("type") == "reasoning":
|
||
for p in (item.get("content") or []):
|
||
if p.get("type") in ("reasoning_text", "summary_text") and p.get("text"):
|
||
blocks.append({"type": "thinking", "thinking": p["text"]})
|
||
elif item.get("type") == "function_call":
|
||
try: args = json.loads(item.get("arguments", "")) if item.get("arguments") else {}
|
||
except: args = {"_raw": item.get("arguments", "")}
|
||
blocks.append({"type": "tool_use", "id": item.get("call_id", item.get("id", "")),
|
||
"name": item.get("name", ""), "input": args})
|
||
status = (data.get("status") or "").lower()
|
||
if status == "failed":
|
||
err = data.get("error") or {}
|
||
emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
|
||
_raise_if_retryable_overload(emsg)
|
||
if emsg: blocks.append({"type": "text", "text": f"!!!Error: {emsg}"}); yield f"!!!Error: {emsg}"
|
||
elif status == "incomplete" and not any(b.get("type") == "text" for b in blocks):
|
||
reason = ((data.get("incomplete_details") or {}).get("reason", "")) or "unknown"
|
||
marker = f"[!!! output truncated: {reason}]"
|
||
blocks.append({"type": "text", "text": marker}); yield marker
|
||
else:
|
||
_record_usage(data.get("usage") or {}, api_mode)
|
||
msg = (data.get("choices") or [{}])[0].get("message", {})
|
||
reasoning = msg.get("reasoning_content") or msg.get("reasoning", "")
|
||
if reasoning:
|
||
blocks.append({"type": "thinking", "thinking": reasoning})
|
||
content = msg.get("content", "")
|
||
if content:
|
||
blocks.append({"type": "text", "text": content}); yield content
|
||
for tc in (msg.get("tool_calls") or []):
|
||
fn = tc.get("function", {})
|
||
try: args = json.loads(fn.get("arguments", "")) if fn.get("arguments") else {}
|
||
except: args = {"_raw": fn.get("arguments", "")}
|
||
blocks.append({"type": "tool_use", "id": tc.get("id", ""), "name": fn.get("name", ""), "input": args})
|
||
return blocks
|
||
|
||
def _stamp_oai_cache_markers(messages, model):
|
||
"""Add cache_control to last 2 user messages for Anthropic models via OAI-compatible relay."""
|
||
ml = model.lower()
|
||
if not any(k in ml for k in ('claude', 'anthropic')): return
|
||
user_idxs = [i for i, m in enumerate(messages) if m.get('role') == 'user']
|
||
for idx in user_idxs[-2:]:
|
||
c = messages[idx].get('content')
|
||
if isinstance(c, str):
|
||
messages[idx] = {**messages[idx], 'content': [{'type': 'text', 'text': c, 'cache_control': {'type': 'ephemeral'}}]}
|
||
elif isinstance(c, list) and c:
|
||
c = list(c); c[-1] = dict(c[-1], cache_control={'type': 'ephemeral'})
|
||
messages[idx] = {**messages[idx], 'content': c}
|
||
|
||
def _stream_with_retry(sess, url, headers, payload, parse_fn):
|
||
STATS['session'] = sess.name
|
||
_RETRYABLE = {408, 409, 425, 429, 500, 502, 503, 504, 520, 521, 522, 523, 524, 525, 526, 527, 529}
|
||
cap = float(getattr(sess, 'max_retry_after', 60.0))
|
||
def _delay(resp, attempt):
|
||
try: ra = float((resp.headers or {}).get("retry-after"))
|
||
except: ra = None
|
||
return None if ra and ra > cap else max(0.5, ra or min(30.0, 3.0 * (2 ** attempt)))
|
||
def _stopped(): return getattr(sess, 'should_stop', None) and sess.should_stop()
|
||
def _sleep(d): # interruptible sleep; True if aborted
|
||
end = time.time() + d
|
||
while time.time() < end:
|
||
if _stopped(): return True
|
||
time.sleep(0.2)
|
||
return _stopped()
|
||
for attempt in range(sess.max_retries + 1):
|
||
if _stopped(): return []
|
||
streamed = False
|
||
STATS.update(t_start=time.time(), t_ttft=None)
|
||
if not sess.stream: STATS['t_ttft'] = STATS['t_start']
|
||
try:
|
||
sess._tid = threading.get_ident() # abort() looks up _INFLIGHT[_tid]
|
||
with requests.post(url, headers=headers, json=payload, stream=sess.stream,
|
||
timeout=(sess.connect_timeout, sess.read_timeout), proxies=sess.proxies, verify=sess.verify) as r:
|
||
sess.active_response = r
|
||
if r.status_code >= 400:
|
||
#pathlib.Path(__file__).parent.joinpath('temp','bad_requests.json').write_text(json.dumps({"url":url,"headers":headers,"payload":payload,"t":time.time()},ensure_ascii=False),encoding='utf-8')
|
||
d = _delay(r, attempt) if r.status_code in _RETRYABLE and attempt < sess.max_retries else None
|
||
if d is not None:
|
||
print(f"[LLM Retry] HTTP {r.status_code}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})")
|
||
if _sleep(d): return []
|
||
continue
|
||
try: body = r.text.strip()[:500]
|
||
except: body = ""
|
||
err = f"!!!Error: HTTP {r.status_code}" + (f" (retry-after > {cap:.0f}s)" if d is None and r.status_code in _RETRYABLE and attempt < sess.max_retries else "") + (f": {body}" if body else "")
|
||
yield err; return [{"type": "text", "text": err}]
|
||
if sess.stream: # TTFT = first SSE event (prefill done), not first visible chunk; hidden thinking must count in decode window
|
||
_il = r.iter_lines
|
||
def _probe(*a, **k):
|
||
for line in _il(*a, **k):
|
||
if line and STATS.get('t_ttft') is None: STATS['t_ttft'] = time.time()
|
||
yield line
|
||
r.iter_lines = _probe
|
||
gen = parse_fn(r)
|
||
try:
|
||
while True:
|
||
if getattr(sess, 'should_stop', None) and sess.should_stop():
|
||
STATS['t_end'] = time.time(); return []
|
||
chunk = next(gen)
|
||
if chunk and STATS.get('t_ttft') is None: STATS['t_ttft'] = time.time()
|
||
streamed = True; yield chunk
|
||
except StopIteration as e:
|
||
if not e.value and not streamed: raise requests.ConnectionError("empty response")
|
||
STATS['t_end'] = time.time()
|
||
STATS['tps'] = STATS.get('out', 0) / max(1e-9, STATS['t_end'] - max(STATS['t_ttft'] or 0, STATS['t_start']))
|
||
return e.value or []
|
||
except (requests.Timeout, requests.ConnectionError, requests.exceptions.ChunkedEncodingError) as e:
|
||
err = f"!!!Error: {type(e).__name__}: {e}" if str(e) else f"!!!Error: {type(e).__name__}"
|
||
if getattr(sess, 'should_stop', None) and sess.should_stop(): return []
|
||
if attempt < sess.max_retries:
|
||
d = _delay(None, attempt)
|
||
print(f"[LLM Retry] {type(e).__name__}, retry in {d:.1f}s ({attempt+1}/{sess.max_retries+1})")
|
||
if _sleep(d): return []
|
||
continue
|
||
yield err; return [{"type": "text", "text": err}]
|
||
except Exception as e:
|
||
err = f"\n\n[!!! 流异常中断 {type(e).__name__}: {e} !!!]" if streamed else f"!!!Error: {type(e).__name__}: {e}"
|
||
yield err; return [{"type": "text", "text": err}]
|
||
|
||
def _openai_stream(sess, messages):
|
||
model, api_mode = sess.model, sess.api_mode
|
||
ml = model.lower()
|
||
temperature = sess.temperature
|
||
if 'kimi' in ml or 'moonshot' in ml: temperature = 1
|
||
elif 'minimax' in ml: temperature = max(0.01, min(temperature, 1.0)) # MiniMax requires temp in (0, 1]
|
||
headers = {"Authorization": f"Bearer {sess.api_key}", "Content-Type": "application/json", "Accept": "text/event-stream", 'originator': 'codex_exec'}
|
||
headers["User-Agent"] = sess.user_agent
|
||
if api_mode == "responses":
|
||
url = auto_make_url(sess.api_base, "responses")
|
||
payload = {"model": model, "input": _to_responses_input(messages), "stream": sess.stream,
|
||
"prompt_cache_key": _RESP_CACHE_KEY, "instructions": sess.system or "You are an Omnipotent Executor.",
|
||
"client_metadata": {"x-codex-window-id": f"{_RESP_CACHE_KEY}:0","x-codex-installation-id": _RESP_CODEX_KEY},
|
||
'include': ['reasoning.encrypted_content']}
|
||
if sess.reasoning_effort: payload["reasoning"] = {"effort": sess.reasoning_effort}
|
||
if sess.max_tokens: payload["max_output_tokens"] = sess.max_tokens
|
||
else:
|
||
url = auto_make_url(sess.api_base, "chat/completions")
|
||
if sess.system: messages = [{"role": "system", "content": sess.system}] + messages
|
||
_stamp_oai_cache_markers(messages, model)
|
||
payload = {"model": model, "messages": messages, "stream": sess.stream}
|
||
if sess.stream: payload["stream_options"] = {"include_usage": True}
|
||
if temperature != 1: payload["temperature"] = temperature
|
||
if sess.max_tokens: payload["max_completion_tokens" if ml.startswith(("gpt-5", "o1", "o2", "o3", "o4")) else "max_tokens"] = sess.max_tokens
|
||
if sess.reasoning_effort: payload["reasoning_effort"] = sess.reasoning_effort
|
||
tools = getattr(sess, 'tools', None)
|
||
if tools: payload["tools"] = _prepare_oai_tools(tools, api_mode)
|
||
if sess.service_tier: payload["service_tier"] = sess.service_tier
|
||
parse_fn = (lambda r: _parse_openai_sse(r.iter_lines(), api_mode)) if sess.stream else (lambda r: _parse_openai_json(r.json(), api_mode))
|
||
return (yield from _stream_with_retry(sess, url, headers, payload, parse_fn))
|
||
|
||
def _prepare_oai_tools(tools, api_mode="chat_completions"):
|
||
if api_mode == "responses":
|
||
resp_tools = []
|
||
for t in tools:
|
||
if t.get("type") == "function" and "function" in t:
|
||
rt = {"type": "function"}; rt.update(t["function"])
|
||
resp_tools.append(rt)
|
||
else: resp_tools.append(t)
|
||
return resp_tools
|
||
return tools
|
||
|
||
def _to_responses_input(messages):
|
||
result, pending = [], []
|
||
for msg in messages:
|
||
role = str(msg.get("role", "user")).lower()
|
||
if role == "tool":
|
||
cid = msg.get("tool_call_id") or (pending.pop(0) if pending else f"call_{uuid.uuid4().hex[:8]}")
|
||
result.append({"type": "function_call_output", "call_id": cid, "output": msg.get("content", "")})
|
||
continue
|
||
if role not in ["user", "assistant", "system", "developer"]: role = "user"
|
||
if role == "system": role = "developer" # Responses API uses 'developer' instead of 'system'
|
||
content = msg.get("content", "")
|
||
text_type = "output_text" if role == "assistant" else "input_text"
|
||
parts = []
|
||
if isinstance(content, str):
|
||
if content: parts.append({"type": text_type, "text": content})
|
||
elif isinstance(content, list):
|
||
for part in content:
|
||
if not isinstance(part, dict): continue
|
||
ptype = part.get("type")
|
||
if ptype == "text":
|
||
text = part.get("text", "")
|
||
if text: parts.append({"type": text_type, "text": text})
|
||
elif ptype == "image_url":
|
||
url = (part.get("image_url") or {}).get("url", "")
|
||
if url and role != "assistant": parts.append({"type": "input_image", "image_url": url})
|
||
if len(parts) == 0: parts = [{"type": text_type, "text": str(content) if not isinstance(content, list) else '[empty]'}]
|
||
result.append({"role": role, "content": parts})
|
||
pending = []
|
||
for tc in (msg.get("tool_calls") or []):
|
||
f = tc.get("function", {})
|
||
cid = tc.get("id") or f"call_{uuid.uuid4().hex[:8]}"
|
||
pending.append(cid)
|
||
result.append({"type": "function_call", "call_id": cid, "name": f.get("name", ""), "arguments": f.get("arguments", "")})
|
||
return result
|
||
|
||
|
||
def _msgs_claude2oai(messages):
|
||
result = []
|
||
for msg in messages:
|
||
role = msg.get("role", "user")
|
||
content = msg.get("content", "")
|
||
blocks = content if isinstance(content, list) else [{"type": "text", "text": str(content)}]
|
||
if role == "assistant":
|
||
text_parts, tool_calls, reasoning = [], [], ""
|
||
for b in blocks:
|
||
if not isinstance(b, dict): continue
|
||
if b.get("type") == "thinking" and b.get("thinking"): reasoning = b["thinking"]
|
||
elif b.get("type") == "text" and b.get("text"): text_parts.append({"type": "text", "text": b.get("text", "")})
|
||
elif b.get("type") == "tool_use":
|
||
tool_calls.append({
|
||
"id": b.get("id") or '', "type": "function",
|
||
"function": {"name": b.get("name", ""), "arguments": json.dumps(b.get("input", {}), ensure_ascii=False)}
|
||
})
|
||
m = {"role": "assistant"}
|
||
if reasoning: m["reasoning_content"] = reasoning
|
||
if text_parts: m["content"] = text_parts
|
||
elif not tool_calls: m["content"] = "."
|
||
if tool_calls: m["tool_calls"] = tool_calls
|
||
result.append(m)
|
||
elif role == "user":
|
||
text_parts = []
|
||
for b in blocks:
|
||
if not isinstance(b, dict): continue
|
||
if b.get("type") == "tool_result":
|
||
if text_parts:
|
||
result.append({"role": "user", "content": text_parts})
|
||
text_parts = []
|
||
tr = b.get("content", "")
|
||
if isinstance(tr, list):
|
||
tr = "\n".join(x.get("text", "") for x in tr if isinstance(x, dict) and x.get("type") == "text")
|
||
result.append({"role": "tool", "tool_call_id": b.get("tool_use_id") or '', "content": tr if isinstance(tr, str) else str(tr)})
|
||
elif b.get("type") == "image":
|
||
src = b.get("source") or {}
|
||
if src.get("type") == "base64" and src.get("data"):
|
||
text_parts.append({"type": "image_url", "image_url": {"url": f"data:{src.get('media_type', 'image/png')};base64,{src.get('data', '')}"}})
|
||
elif b.get("type") == "image_url": text_parts.append(b)
|
||
elif b.get("type") == "text" and b.get("text"): text_parts.append({"type": "text", "text": b.get("text", "")})
|
||
if text_parts: result.append({"role": "user", "content": text_parts})
|
||
else: result.append(msg)
|
||
return result
|
||
|
||
|
||
class BaseSession:
|
||
def __init__(self, cfg):
|
||
self.api_key = cfg['apikey']
|
||
self.api_base = cfg['apibase'].rstrip('/')
|
||
self.model = cfg.get('model', '')
|
||
default_context_win = 38000; default_cut_msg_interval = 8
|
||
if 'deepseek' in self.model.lower():
|
||
default_context_win = 80000; default_cut_msg_interval = 25; self.trim_keep_rate = 0.3
|
||
self.context_win = cfg.get('context_win', default_context_win)
|
||
self.maxlen_multiplier = min(max(self.context_win / default_context_win * 0.75, 1.0), 3.0)
|
||
self.cut_msg_interval = int(default_cut_msg_interval * self.maxlen_multiplier)
|
||
self.trim_keep_prefix = max(0, int(cfg.get('trim_keep_prefix', 0) or 0))
|
||
self.history = []; self.lock = threading.Lock(); self.system = ""
|
||
self.name = cfg.get('name', self.model)
|
||
self.extra_sys_prompt = cfg.get('extra_sys_prompt', '')
|
||
if cfg.get('extra_sys_prompt_file'):
|
||
self.extra_sys_prompt = (self.extra_sys_prompt or '') + open(cfg['extra_sys_prompt_file'] if os.path.isabs(cfg['extra_sys_prompt_file']) else os.path.join(_ROOT, cfg['extra_sys_prompt_file']), encoding='utf-8').read()
|
||
proxy = cfg.get('proxy');
|
||
self.proxies = {"http": proxy, "https": proxy} if proxy else None
|
||
self.max_retries = max(0, int(cfg.get('max_retries', 4)))
|
||
self.max_retry_after = float(cfg.get('max_retry_after', 60.0))
|
||
self.verify = cfg.get('verify', True)
|
||
self.stream = cfg.get('stream', True)
|
||
default_ct, default_rt = (5, 40) if self.stream else (10, 240)
|
||
self.connect_timeout = max(1, int(cfg.get('timeout', default_ct)))
|
||
self.read_timeout = max(5, int(cfg.get('read_timeout', default_rt)))
|
||
def _enum(key, valid):
|
||
v = cfg.get(key); v = None if v is None else str(v).strip().lower()
|
||
return v if not v or v in valid else print(f"[WARN] Invalid {key} {v!r}, ignored.")
|
||
self.reasoning_effort = _enum('reasoning_effort', {'none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max'})
|
||
self.service_tier = _enum('service_tier', {'auto', 'default', 'priority', 'flex'})
|
||
self.thinking_type = _enum('thinking_type', {'adaptive', 'enabled', 'disabled'})
|
||
self.thinking_budget_tokens = cfg.get('thinking_budget_tokens')
|
||
self.omit_thinking = cfg.get('omit_thinking', False) # Exclude thinking from session history
|
||
mode = str(cfg.get('api_mode', 'chat_completions')).strip().lower().replace('-', '_')
|
||
self.api_mode = 'responses' if mode in ('responses', 'response') else 'chat_completions'
|
||
self.temperature = cfg.get('temperature', 1)
|
||
self.max_tokens = cfg.get('max_tokens')
|
||
self.default_ua = "claude-cli/2.1.280 (external, cli)"
|
||
self.user_agent = cfg.get("user_agent", self.default_ua)
|
||
def _apply_claude_thinking(self, payload):
|
||
if self.thinking_type:
|
||
thinking = {"type": self.thinking_type}
|
||
if self.thinking_type == 'enabled':
|
||
if self.thinking_budget_tokens is None: print("[WARN] thinking_type='enabled' requires thinking_budget_tokens, ignored.")
|
||
else:
|
||
thinking["budget_tokens"] = self.thinking_budget_tokens; payload["thinking"] = thinking
|
||
else: payload["thinking"] = thinking
|
||
if self.reasoning_effort:
|
||
effort = {'low': 'low', 'medium': 'medium', 'high': 'high', 'xhigh': 'max', 'max': 'max'}.get(self.reasoning_effort)
|
||
if effort: payload["output_config"] = {"effort": effort}
|
||
else: print(f"[WARN] reasoning_effort {self.reasoning_effort!r} is unsupported for Claude output_config.effort, ignored.")
|
||
def ask(self, prompt):
|
||
def _ask_gen():
|
||
with self.lock:
|
||
self.history.append({"role": "user", "content": [{"type": "text", "text": prompt}]})
|
||
trim_messages_history(self.history, self)
|
||
messages = self.make_messages(self.history)
|
||
content_blocks = None; content = ''
|
||
gen = self.raw_ask(messages)
|
||
try:
|
||
while True: chunk = next(gen); content += chunk; yield chunk
|
||
except StopIteration as e: content_blocks = e.value or []
|
||
if len(content_blocks) > 1: print(f"[DEBUG BaseSession.ask] content_blocks: {content_blocks}")
|
||
for block in (content_blocks or []):
|
||
if block.get('type', '') == 'tool_use':
|
||
tu = {'name': block.get('name', ''), 'arguments': block.get('input', {})}
|
||
yield f'<tool_use>{json.dumps(tu, ensure_ascii=False)}</tool_use>'
|
||
if content.strip() and not content.startswith("!!!Error:"): self.history.append({"role": "assistant", "content": [{"type": "text", "text": content}]})
|
||
return _ask_gen()
|
||
|
||
def _keep_claude_block(b): return not isinstance(b, dict) or b.get("type") != "thinking" or b.get("signature")
|
||
def _drop_unsigned_thinking(messages):
|
||
for m in messages:
|
||
c = m.get("content")
|
||
if isinstance(c, list): m["content"] = [b for b in c if _keep_claude_block(b)]
|
||
return messages
|
||
|
||
def _ensure_thinking_blocks(messages, model):
|
||
"""deepseek needs thinking in history!"""
|
||
if 'deepseek' not in model.lower(): return messages
|
||
for m in messages:
|
||
if m.get("role") != "assistant": continue
|
||
c = m.get("content")
|
||
if not isinstance(c, list): continue
|
||
has_thinking = any(isinstance(b, dict) and b.get("type") == "thinking" for b in c)
|
||
if not has_thinking: m["content"] = [{"type": "thinking", "thinking": "...", "signature": "placeholder"}, *c]
|
||
return messages
|
||
|
||
def _fix_messages(messages):
|
||
if not messages: return messages
|
||
W = lambda c: c if isinstance(c, list) else [{"type": "text", "text": str(c)}]
|
||
merged = []
|
||
for m in messages:
|
||
if m.get('role') not in ('user', 'assistant'): continue
|
||
blocks = W(m.get('content', []))
|
||
if merged and m['role'] == merged[-1]['role']:
|
||
merged[-1]['content'] += list(blocks)
|
||
else:
|
||
merged.append({"role": m['role'], "content": list(blocks)})
|
||
while merged and merged[0]['role'] != 'user': merged.pop(0)
|
||
if not merged: return []
|
||
prev_uses = []
|
||
for m in merged:
|
||
c = m['content']
|
||
if m['role'] == 'assistant':
|
||
seen, out = set(), []
|
||
for b in c:
|
||
uid = b.get('id') if isinstance(b, dict) and b.get('type') == 'tool_use' else None
|
||
if uid and uid in seen: continue
|
||
if uid: seen.add(uid)
|
||
out.append(b)
|
||
m['content'] = out
|
||
prev_uses = [b.get('id') for b in out if isinstance(b, dict) and b.get('type') == 'tool_use']
|
||
else:
|
||
got, rest = {}, []
|
||
for b in c:
|
||
tid = b.get('tool_use_id') if isinstance(b, dict) and b.get('type') == 'tool_result' else None
|
||
if tid and tid in prev_uses and tid not in got: got[tid] = b
|
||
elif isinstance(b, dict) and b.get('type') == 'tool_result': rest.append({"type": "text", "text": str(b.get('content', ''))})
|
||
else: rest.append(b)
|
||
m['content'] = [got.get(u) or {"type": "tool_result", "tool_use_id": u, "content": "(error)"} for u in prev_uses] + rest
|
||
prev_uses = []
|
||
for m in merged: m['content'] = [b for b in m['content'] if not (isinstance(b, dict) and b.get('type') == 'text' and not (b.get('text') or '').strip())] or [{"type": "text", "text": "."}]
|
||
return merged
|
||
|
||
def _cc_system(system, compat='off'): # off | simple | half(未来 full)
|
||
b = [{"type": "text", "text": "You are Claude Code, Anthropic's official CLI for Claude.", "cache_control": {"type": "ephemeral"}}] if compat != 'off' or not system else []
|
||
if compat in ('half', 'full'): b.append({"type": "text", "text": open(os.path.join(_ROOT, 'assets', 'cc_system_template.txt'), encoding='utf-8').read()})
|
||
return b + ([{"type": "text", "text": system}] if system else [])
|
||
|
||
class NativeClaudeSession(BaseSession):
|
||
native_ua = "claude-cli/2.1.280 (native, cli)"
|
||
def __init__(self, cfg):
|
||
super().__init__(cfg)
|
||
self.claude_compat = cfg.get("claude_compat") or ("simple" if cfg.get("fake_cc_system_prompt") else "off")
|
||
self.fake_cc_system_prompt = self.claude_compat != "off"
|
||
self._session_id = str(uuid.uuid4())
|
||
self._account_uuid = str(uuid.uuid4())
|
||
self._device_id = uuid.uuid4().hex + uuid.uuid4().hex[:32]
|
||
self.tools = None
|
||
if self.user_agent == self.default_ua: self.user_agent = self.native_ua
|
||
self.api_key_header = str(cfg.get('api_key_header', 'auto')).strip().lower()
|
||
def raw_ask(self, messages):
|
||
if self.max_tokens is None: self.max_tokens = 8192
|
||
model = self.model
|
||
messages = _fix_messages(messages)
|
||
if 'claude' in model.lower(): messages = _drop_unsigned_thinking(messages)
|
||
messages = _ensure_thinking_blocks(messages, self.model)
|
||
beta_parts = ["claude-code-20250219", "interleaved-thinking-2025-05-14", "redact-thinking-2026-02-12", "thinking-token-count-2026-05-13", "context-management-2025-06-27", "prompt-caching-scope-2026-01-05", "mid-conversation-system-2026-04-07", "effort-2025-11-24", "fallback-credit-2026-06-01"]
|
||
if "[1m]" in model.lower():
|
||
beta_parts.insert(1, "context-1m-2025-08-07"); model = model.replace("[1m]", "").replace("[1M]", "")
|
||
headers = {"Content-Type": "application/json", "anthropic-version": "2023-06-01",
|
||
"anthropic-beta": ",".join(beta_parts), "anthropic-dangerous-direct-browser-access": "true",
|
||
"user-agent": self.user_agent, "x-app": "cli"}
|
||
headers.update({"Accept": "application/json", "X-Claude-Code-Session-Id": self._session_id, "X-Stainless-Arch": "x64", "X-Stainless-Lang": "js", "X-Stainless-OS": "Windows", "X-Stainless-Package-Version": "0.94.0", "X-Stainless-Retry-Count": "0", "X-Stainless-Runtime": "node", "X-Stainless-Runtime-Version": "v24.3.0", "X-Stainless-Timeout": "600"})
|
||
if self.api_key_header == 'x-api-key': headers["x-api-key"] = self.api_key
|
||
elif self.api_key_header == 'bearer': headers["authorization"] = f"Bearer {self.api_key}"
|
||
elif self.api_key.startswith("sk-ant-"): headers["x-api-key"] = self.api_key
|
||
else: headers["authorization"] = f"Bearer {self.api_key}"
|
||
payload = {"model": model, "messages": messages, "max_tokens": self.max_tokens, "stream": self.stream}
|
||
#if self.fake_cc_system_prompt: payload["max_tokens"] = 64000
|
||
if self.temperature != 1: payload["temperature"] = self.temperature
|
||
self._apply_claude_thinking(payload)
|
||
#payload["context_management"] = {"edits": [{"type": "clear_thinking_20251015", "keep": "all"}]};
|
||
if self.fake_cc_system_prompt:
|
||
if 'thinking' not in payload: payload["thinking"] = {"type": "adaptive"}
|
||
if 'output_config' not in payload: payload["output_config"] = {"effort": "medium"}
|
||
payload["metadata"] = {"user_id": json.dumps({"device_id": self._device_id, "account_uuid": "", "session_id": self._session_id}, separators=(',', ':'))}
|
||
if self.tools:
|
||
claude_tools = openai_tools_to_claude(self.tools)
|
||
tools = [dict(t) for t in claude_tools]; tools[-1]["cache_control"] = {"type": "ephemeral"}
|
||
payload["tools"] = tools
|
||
else: print("[ERROR] No tools provided for this session.")
|
||
payload['system'] = _cc_system(self.system, self.claude_compat)
|
||
user_idxs = [i for i, m in enumerate(messages) if m['role'] == 'user']
|
||
for idx in user_idxs[-2:]:
|
||
messages[idx] = {**messages[idx], "content": list(messages[idx]["content"])}
|
||
messages[idx]["content"][-1] = dict(messages[idx]["content"][-1], cache_control={"type": "ephemeral"})
|
||
url = auto_make_url(self.api_base, "messages") + '?beta=true'
|
||
parse_fn = (lambda r: _parse_claude_sse(r.iter_lines())) if self.stream else (lambda r: _parse_claude_json(r.json()))
|
||
return (yield from _stream_with_retry(self, url, headers, payload, parse_fn))
|
||
|
||
def ask(self, msg):
|
||
assert type(msg) is dict
|
||
with self.lock:
|
||
self.history.append(msg)
|
||
trim_messages_history(self.history, self)
|
||
messages = [{"role": m["role"], "content": list(m["content"])} for m in self.history]
|
||
content_blocks = None
|
||
gen = self.raw_ask(messages)
|
||
try:
|
||
while True: yield next(gen)
|
||
except StopIteration as e: content_blocks = e.value or []
|
||
if content_blocks and (_injected := _ensure_text_block(content_blocks)): yield _injected
|
||
if content_blocks and not (len(content_blocks) == 1 and content_blocks[0].get("text", "").startswith("!!!Error:")):
|
||
history_blocks = content_blocks
|
||
if self.omit_thinking: history_blocks = [b for b in content_blocks if b.get("type") != "thinking"]
|
||
self.history.append({"role": "assistant", "content": history_blocks})
|
||
text_parts = [b["text"] for b in content_blocks if b.get("type") == "text"]
|
||
content = "\n".join(text_parts).strip()
|
||
tool_calls = [MockToolCall(b["name"], b.get("input", {}), id=b.get("id", "")) for b in content_blocks if b.get("type") == "tool_use"]
|
||
if not tool_calls: tool_calls, content = _parse_text_tool_calls(content)
|
||
thinking_parts = [b["thinking"] for b in content_blocks if b.get("type") == "thinking"]
|
||
thinking = "\n".join(thinking_parts).strip()
|
||
if not thinking:
|
||
think_pattern = r"<think(?:ing)?>(.*?)</think(?:ing)?>"
|
||
think_match = re.search(think_pattern, content, re.DOTALL)
|
||
if think_match:
|
||
thinking = think_match.group(1).strip()
|
||
content = re.sub(think_pattern, "", content, flags=re.DOTALL)
|
||
raw = "[" + ",\n".join(repr(b) for b in content_blocks) + "]"
|
||
return MockResponse(thinking, content, tool_calls, raw)
|
||
|
||
class NativeOAISession(NativeClaudeSession):
|
||
native_ua = "codex_exec/0.139.0 (Windows 10.0.26200; x86_64) unknown (codex_exec; 0.139.0)"
|
||
def raw_ask(self, messages):
|
||
messages = _fix_messages(messages)
|
||
messages = _ensure_thinking_blocks(messages, self.model)
|
||
return (yield from _openai_stream(self, _msgs_claude2oai(messages)))
|
||
|
||
def openai_tools_to_claude(tools):
|
||
"""[{type:'function', function:{name,description,parameters}}] → [{name,description,input_schema}]."""
|
||
result = []
|
||
for t in tools:
|
||
if 'input_schema' in t: result.append(t); continue # 已是claude格式
|
||
fn = t.get('function', t)
|
||
result.append({'name': fn['name'], 'description': fn.get('description', ''),
|
||
'input_schema': fn.get('parameters', {'type': 'object', 'properties': {}})})
|
||
return result
|
||
|
||
class MockFunction:
|
||
def __init__(self, name, arguments): self.name, self.arguments = name, arguments
|
||
|
||
class MockToolCall:
|
||
def __init__(self, name, args, id=''):
|
||
arg_str = json.dumps(args, ensure_ascii=False) if isinstance(args, (dict, list)) else (args or '{}')
|
||
self.function = MockFunction(name, arg_str); self.id = id
|
||
|
||
class MockResponse:
|
||
def __init__(self, thinking, content, tool_calls, raw, stop_reason='end_turn'):
|
||
self.thinking = thinking; self.content = content
|
||
self.tool_calls = tool_calls; self.raw = raw
|
||
self.stop_reason = 'tool_use' if tool_calls else stop_reason
|
||
def __repr__(self):
|
||
return f"<MockResponse thinking={bool(self.thinking)}, content='{self.content}', tools={bool(self.tool_calls)}>"
|
||
|
||
def _parse_text_tool_calls(content):
|
||
"""Fallback: extract tool calls from text when model doesn't use native tool_use blocks."""
|
||
tcs = []
|
||
# try JSON array: [{"type":"tool_use", "name":..., "input":...}]
|
||
_jp = next((p for p in ['[{"type":"tool_use"', '[{"type": "tool_use"'] if p in content), None)
|
||
if _jp and content.endswith('}]'):
|
||
try:
|
||
idx = content.index(_jp); raw = json.loads(content[idx:])
|
||
tcs = [MockToolCall(b["name"], b.get("input", {}), id=b.get("id", "")) for b in raw if b.get("type") == "tool_use"]
|
||
return tcs, content[:idx].strip()
|
||
except: pass
|
||
# try XML tags: <tool_call>{"name":..., "arguments":...}</tool_call>
|
||
_xp = r"<(?:tool_use|tool_call)>((?:(?!<(?:tool_use|tool_call)>).){15,}?)</(?:tool_use|tool_call)>"
|
||
for s in re.findall(_xp, content, re.DOTALL):
|
||
try:
|
||
d = tryparse(s.strip()); name = d.get('name')
|
||
args = d.get('arguments') or d.get('args') or d.get('input') or {}
|
||
if name: tcs.append(MockToolCall(name, args))
|
||
except: pass
|
||
if tcs: content = re.sub(_xp, "", content, flags=re.DOTALL).strip()
|
||
return tcs, content
|
||
|
||
def _ensure_text_block(blocks):
|
||
"""If response has thinking but no text block, inject a synthetic summary from thinking's first line."""
|
||
if any(b.get("type") == "text" for b in blocks): return None
|
||
th = next((b.get("thinking", "") for b in blocks if b.get("type") == "thinking"), "")
|
||
if not th: return None
|
||
line = th.strip().split('\n', 1)[0]
|
||
txt = "<summary>" + (line[:60] + '...' if len(line) > 60 else line) + "</summary>"
|
||
blocks.insert(1, {"type": "text", "text": txt})
|
||
return txt
|
||
|
||
def _write_llm_log(label, content, log_path=None, model=''):
|
||
if log_path is False: return
|
||
if not log_path:
|
||
log_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), f'temp/model_responses/model_responses_{os.getpid()}.txt')
|
||
os.makedirs(os.path.dirname(os.path.abspath(log_path)), exist_ok=True)
|
||
ts = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
if model: model = f' model={model}'
|
||
with open(log_path, 'a', encoding='utf-8', errors='replace') as f:
|
||
f.write(f"=== {label} === {ts}{model}\n{content}\n\n")
|
||
|
||
def tryparse(json_str):
|
||
try: return json.loads(json_str)
|
||
except: pass
|
||
json_str = json_str.strip().strip('`').replace('json\n', '', 1).strip()
|
||
try: return json.loads(json_str)
|
||
except: pass
|
||
try: return json.loads(json_str[:-1])
|
||
except: pass
|
||
if '}' in json_str: json_str = json_str[:json_str.rfind('}') + 1]
|
||
return json.loads(json_str)
|
||
|
||
class MixinSession:
|
||
"""A Session facade backed by multiple routed transport sessions."""
|
||
_TRANSPORT_OVERRIDES = frozenset({
|
||
'stream', 'connect_timeout', 'read_timeout', 'temperature', 'max_tokens',
|
||
'reasoning_effort', 'service_tier', 'thinking_type',
|
||
'thinking_budget_tokens', 'omit_thinking', 'proxies', 'verify',
|
||
})
|
||
_FACADE_STATE = frozenset({'name', 'history', 'system', 'tools', 'lock'})
|
||
|
||
def __init__(self, all_sessions, cfg):
|
||
self._retries = cfg.get('max_retries', 3)
|
||
self._base_delay = cfg.get('base_delay', 3.0)
|
||
self._spring_sec = cfg.get('spring_back', 300)
|
||
selected = [all_sessions[i].backend if isinstance(i, int) else
|
||
next(s.backend for s in all_sessions if type(s) is not dict and s.backend.name == i)
|
||
for i in cfg.get('llm_nos', [])]
|
||
if not selected: raise ValueError('MixinSession: no sessions selected')
|
||
native_groups = {isinstance(s, NativeClaudeSession) for s in selected}
|
||
if len(native_groups) != 1:
|
||
raise ValueError(f"MixinSession: sessions must be in same group (Native or non-Native), got {[type(s).__name__ for s in selected]}")
|
||
|
||
self._sessions = [copy.copy(s) for s in selected]
|
||
for s in self._sessions: s.max_retries = 0
|
||
self._native = native_groups.pop()
|
||
self._ask_impl = selected[0].ask.__func__
|
||
self._cur_idx, self._switched_at = 0, 0.0
|
||
|
||
primary = self._sessions[0]
|
||
self.name = '|'.join(s.name for s in self._sessions)
|
||
self.history = copy.deepcopy(primary.history)
|
||
self.system = primary.system
|
||
self.tools = getattr(primary, 'tools', None)
|
||
self.lock = threading.Lock()
|
||
@property
|
||
def primary(self): return self._sessions[0]
|
||
@property
|
||
def current(self): return self._sessions[self._cur_idx]
|
||
@property
|
||
def current_name(self): return self.current.name
|
||
def __getattr__(self, name): return getattr(self.current, name)
|
||
def __setattr__(self, name, value):
|
||
sessions = self.__dict__.get('_sessions')
|
||
if sessions and name in self._TRANSPORT_OVERRIDES:
|
||
for s in sessions: setattr(s, name, value)
|
||
return
|
||
node_owns = sessions and any(
|
||
name in s.__dict__ or any(name in cls.__dict__ for cls in type(s).__mro__)
|
||
for s in sessions)
|
||
if node_owns and name not in self._FACADE_STATE: raise AttributeError(f"MixinSession.{name} is node-specific and read-only")
|
||
object.__setattr__(self, name, value)
|
||
def ask(self, prompt):
|
||
self._pick() # Select the node before ask() reads its context limits.
|
||
return self._ask_impl(self, prompt)
|
||
def make_messages(self, messages): return messages
|
||
def _pick(self):
|
||
if self._cur_idx and time.time() - self._switched_at > self._spring_sec: self._cur_idx = 0
|
||
return self._cur_idx
|
||
def _prepare(self, idx, messages):
|
||
session = self._sessions[idx]
|
||
session.system = self.system
|
||
session.tools = openai_tools_to_claude(self.tools) if self.tools and type(session) is NativeClaudeSession else self.tools
|
||
return messages if self._native else session.make_messages(messages)
|
||
def raw_ask(self, messages):
|
||
base, n = self._pick(), len(self._sessions)
|
||
test_error = lambda x: isinstance(x, str) and x.lstrip().startswith(('!!!Error:', '[Error:'))
|
||
for attempt in range(self._retries + 1):
|
||
idx = (base + attempt) % n
|
||
session = self._sessions[idx]
|
||
gen = session.raw_ask(self._prepare(idx, messages))
|
||
print(f'[MixinSession] Using session ({session.name})')
|
||
last_chunk, return_val, yielded = None, [], False
|
||
try:
|
||
while True:
|
||
chunk = next(gen); last_chunk = chunk
|
||
if not yielded and test_error(chunk): continue
|
||
yield chunk; yielded = True
|
||
except StopIteration as e: return_val = e.value or []
|
||
is_err = test_error(last_chunk)
|
||
if not is_err:
|
||
if attempt > 0: self._cur_idx = idx; self._switched_at = time.time()
|
||
elif isinstance(last_chunk, str) and '[!!! 流异常中断' in last_chunk and n > 1:
|
||
self._cur_idx = (idx + 1) % n; self._switched_at = time.time()
|
||
print(f'[MixinSession] Partial failure, next call → s{self._cur_idx} ({self.current.name})')
|
||
return return_val
|
||
if attempt >= self._retries:
|
||
yield last_chunk; return return_val
|
||
nxt = (base + attempt + 1) % n
|
||
if nxt == base:
|
||
rnd = (attempt + 1) // n
|
||
delay = min(30, self._base_delay * (1.5 ** rnd))
|
||
print(f'[MixinSession] {last_chunk[:80]}, round {rnd} exhausted, retry in {delay:.1f}s')
|
||
time.sleep(delay)
|
||
else:
|
||
print(f'[MixinSession] {last_chunk[:80]}, retry {attempt+1}/{self._retries} (s{idx}→s{nxt})')
|
||
|
||
THINKING_PROMPT_ZH = """
|
||
### 行动规范（持续有效）
|
||
每次回复（含工具调用轮）都先在回复文字中包含一个<summary></summary> 中输出极简单行（<30字）物理快照：上次结果新信息+本次意图。此内容进入长期工作记忆。
|
||
\n**若用户需求未完成，必须进行工具调用！**
|
||
""".strip()
|
||
THINKING_PROMPT_EN = """
|
||
### Action Protocol (always in effect)
|
||
The reply body should first include a minimal one-line (<30 words) physical snapshot in <summary></summary>: new info from last result + current intent. This goes into long-term working memory.
|
||
\n**If the user's request is not yet complete, tool calls are required!**
|
||
""".strip()
|
||
|
||
class NativeToolClient:
|
||
@staticmethod
|
||
def _thinking_prompt(): return THINKING_PROMPT_EN if os.environ.get('GA_LANG') == 'en' else THINKING_PROMPT_ZH
|
||
def __init__(self, backend):
|
||
self.backend = backend
|
||
self.backend.system = self._thinking_prompt()
|
||
self.name = self.backend.name
|
||
self._pending_tool_ids = []
|
||
self.log_path = None
|
||
def set_system(self, extra_system):
|
||
combined = f"{extra_system}\n\n{self._thinking_prompt()}" if extra_system else self._thinking_prompt()
|
||
if combined != self.backend.system: print(f"[Debug] Updated system prompt, length {len(combined)} chars.")
|
||
self.backend.system = combined
|
||
def chat(self, messages, tools=None):
|
||
if tools: self.backend.tools = tools
|
||
if not self.backend.history: self._pending_tool_ids = []
|
||
combined_content = []; resp = None; tool_results = []
|
||
for msg in messages:
|
||
c = msg.get('content', '')
|
||
if msg['role'] == 'system':
|
||
self.set_system(c); continue
|
||
if isinstance(c, str): combined_content.append({"type": "text", "text": c})
|
||
elif isinstance(c, list): combined_content.extend(c)
|
||
if msg['role'] == 'user' and msg.get('tool_results'): tool_results.extend(msg['tool_results'])
|
||
tr_id_set = set(); tool_result_blocks = []
|
||
for tr in tool_results:
|
||
tool_use_id, content = tr.get("tool_use_id", ""), tr.get("content", "")
|
||
tr_id_set.add(tool_use_id)
|
||
if tool_use_id: tool_result_blocks.append({"type": "tool_result", "tool_use_id": tool_use_id, "content": tr.get("content", "")})
|
||
else: combined_content = [{"type": "text", "text": f'<tool_result>{content}</tool_result>'}] + combined_content
|
||
for tid in self._pending_tool_ids:
|
||
if tid not in tr_id_set: tool_result_blocks.append({"type": "tool_result", "tool_use_id": tid, "content": ""})
|
||
self._pending_tool_ids = []
|
||
# Filter whitespace-only text blocks that cause 400 on strict API proxies
|
||
filtered_content = [c for c in combined_content if c.get("text", "").strip()]
|
||
final_content = tool_result_blocks + filtered_content
|
||
if not final_content: final_content = [{"type": "text", "text": "."}]
|
||
merged = {"role": "user", "content": final_content}
|
||
prompt_raw = '{"role": "user", "content": [\n' + ",\n".join(json.dumps(b, ensure_ascii=False) for b in final_content) + "]}"
|
||
_write_llm_log('Prompt', prompt_raw, self.log_path)
|
||
gen = self.backend.ask(merged)
|
||
try:
|
||
while True:
|
||
chunk = next(gen); yield chunk
|
||
except StopIteration as e: resp = e.value
|
||
if resp: _write_llm_log('Response', resp.raw, self.log_path, model=self.backend.model)
|
||
if resp and hasattr(resp, 'tool_calls') and resp.tool_calls: self._pending_tool_ids = [tc.id for tc in resp.tool_calls]
|
||
return resp
|
||
|
||
def resolve_session(cfg_name, cfg=None):
|
||
if cfg is None: cfg = reload_mykeys()[0].get(cfg_name)
|
||
if not cfg: raise ValueError(f"Config '{cfg_name}' not in mykey")
|
||
cfg = {**cfg, '_mykey_name': cfg_name}
|
||
if 'native' in cfg_name: return (NativeClaudeSession if 'claude' in cfg_name else NativeOAISession)(cfg=cfg)
|
||
print(f"[Legacy] {cfg_name} → 非native文本协议路径(待淘汰, 见legacy_llmcore.py)")
|
||
if 'claude' in cfg_name: return ClaudeSession(cfg=cfg)
|
||
return LLMSession(cfg=cfg) if 'oai' in cfg_name else None
|
||
|
||
def resolve_client(cfg_name, cfg=None):
|
||
s = resolve_session(cfg_name, cfg)
|
||
return (NativeToolClient(s) if isinstance(s, (NativeClaudeSession, NativeOAISession)) else ToolClient(s)) if s else None
|
||
|
||
def fast_ask(prompt, cfg_name):
|
||
sess = resolve_session(cfg_name)
|
||
if not sess: raise ValueError(f"fast_ask: '{cfg_name}' unsupported")
|
||
return "".join(sess.raw_ask([{"role": "user", "content": prompt}]))
|
||
|
||
# legacy三件套(待淘汰)已分离至legacy_llmcore.py, 由其尾部回挂名(禁本行前import)
|
||
try: import legacy_llmcore
|
||
except AttributeError: pass # legacy先import时半初始化, 其尾部自会回挂
|