146 lines
8.5 KiB
Python
146 lines
8.5 KiB
Python
# legacy_llmcore.py — llmcore非native三件套(待淘汰), 单向依赖llmcore(禁from-import), 尾部回挂名
|
|
# 删除条件: [Legacy]日志零引用后, 删本文件+llmcore尾3行+resolve中non-native分支
|
|
import os, json, re
|
|
import llmcore
|
|
|
|
class ClaudeSession(llmcore.BaseSession):
|
|
def raw_ask(self, messages):
|
|
messages = llmcore._fix_messages(messages)
|
|
if self.max_tokens is None: self.max_tokens = 8192
|
|
headers = {"x-api-key": self.api_key, "Content-Type": "application/json", "anthropic-version": "2023-06-01", "anthropic-beta": "prompt-caching-2024-07-31"}
|
|
payload = {"model": self.model, "messages": messages, "max_tokens": self.max_tokens, "stream": self.stream}
|
|
if self.temperature != 1: payload["temperature"] = self.temperature
|
|
self._apply_claude_thinking(payload)
|
|
if self.system: payload["system"] = [{"type": "text", "text": self.system, "cache_control": {"type": "persistent"}}]
|
|
url = llmcore.auto_make_url(self.api_base, "messages")
|
|
parse_fn = (lambda r: llmcore._parse_claude_sse(r.iter_lines())) if self.stream else (lambda r: llmcore._parse_claude_json(r.json()))
|
|
return (yield from llmcore._stream_with_retry(self, url, headers, payload, parse_fn))
|
|
def make_messages(self, raw_list):
|
|
msgs = llmcore._drop_unsigned_thinking([{"role": m['role'], "content": list(m['content'])} for m in raw_list])
|
|
user_idxs = [i for i, m in enumerate(msgs) if m['role'] == 'user']
|
|
for idx in user_idxs[-2:]:
|
|
msgs[idx]["content"][-1] = dict(msgs[idx]["content"][-1], cache_control={"type": "ephemeral"})
|
|
return msgs
|
|
|
|
class LLMSession(llmcore.BaseSession):
|
|
def raw_ask(self, messages): return (yield from llmcore._openai_stream(self, messages))
|
|
def make_messages(self, raw_list): return llmcore._msgs_claude2oai(llmcore._fix_messages(raw_list))
|
|
|
|
class ToolClient:
|
|
def __init__(self, backend, auto_save_tokens=True):
|
|
self.backend = backend
|
|
self.auto_save_tokens = auto_save_tokens
|
|
self.last_tools = ''
|
|
self.name = self.backend.name
|
|
self.total_cd_tokens = 0
|
|
self.log_path = None
|
|
|
|
def chat(self, messages, tools=None):
|
|
tools = json.loads(json.dumps(tools, ensure_ascii=False)) if tools else tools
|
|
for t in tools or []:
|
|
f = t.get('function', {})
|
|
if f.get('name') == 'file_write':
|
|
props = f.get('parameters', {}).get('properties', {})
|
|
props.pop('content', None)
|
|
extra = '. Content must be placed in <file_content> tags in reply body, not in args'
|
|
if extra not in f.get('description', ''): f['description'] = f.get('description', '') + extra
|
|
break
|
|
full_prompt = self._build_protocol_prompt(messages, tools)
|
|
print("Full prompt length:", len(full_prompt), 'chars')
|
|
gen = self.backend.ask(full_prompt)
|
|
llmcore._write_llm_log('Prompt', full_prompt, self.log_path)
|
|
raw_text = ''
|
|
for chunk in gen:
|
|
raw_text += chunk; yield chunk
|
|
llmcore._write_llm_log('Response', raw_text, self.log_path, model=self.backend.model)
|
|
return self._parse_mixed_response(raw_text)
|
|
|
|
def _prepare_tool_instruction(self, tools):
|
|
tool_instruction = ""
|
|
if not tools: return tool_instruction
|
|
tools_json = json.dumps(tools, ensure_ascii=False, separators=(',', ':'))
|
|
_en = os.environ.get('GA_LANG') == 'en'
|
|
if _en:
|
|
tool_instruction = f"""
|
|
### Interaction Protocol (must follow strictly, always in effect)
|
|
Follow these steps to think and act:
|
|
1. **Think**: Analyze the current situation and strategy inside `<thinking>` tags.
|
|
2. **Summarize**: Output a minimal one-line (<30 words) physical snapshot in `<summary>`: new info from last tool result + current tool call intent. This goes into long-term working memory. Must contain real information, no filler.
|
|
3. **Act**: If you need to call tools, output one or more **<tool_use> blocks** after your reply, then stop.
|
|
"""
|
|
else:
|
|
tool_instruction = f"""
|
|
### 交互协议 (必须严格遵守，持续有效)
|
|
请按照以下步骤思考并行动：
|
|
1. **思考**: 在 `<thinking>` 标签中先进行思考，分析现状和策略。
|
|
2. **总结**: 在 `<summary>` 中输出*极为简短*的高度概括的单行（<30字）物理快照，包括上次工具调用结果产生的新信息+本次工具调用意图。此内容将进入长期工作记忆，记录关键信息，严禁输出无实际信息增量的描述。
|
|
3. **行动**: 如需调用工具，请在回复正文之后输出一个（或多个）**<tool_use>块**，然后结束。
|
|
"""
|
|
tool_instruction += f'\nFormat: ```<tool_use>{{"name": "tool_name", "arguments": {{...}}}}</tool_use>```\n\n### Tools (mounted, always in effect):\n{tools_json}\n'
|
|
if self.auto_save_tokens and self.last_tools == tools_json:
|
|
tool_instruction = "\n### Tools: still active, **ready to call**. Protocol unchanged.\n" if _en else "\n### 工具库状态：持续有效（code_run/file_read等），**可正常调用**。调用协议沿用。\n"
|
|
else: self.total_cd_tokens = 0
|
|
self.last_tools = tools_json
|
|
return tool_instruction
|
|
|
|
def _build_protocol_prompt(self, messages, tools):
|
|
system_content = next((m['content'] for m in messages if m['role'].lower() == 'system'), "")
|
|
history_msgs = [m for m in messages if m['role'].lower() != 'system']
|
|
tool_instruction = self._prepare_tool_instruction(tools)
|
|
system = ""; user = ""
|
|
if system_content: system += f"{system_content}\n"
|
|
system += f"{tool_instruction}"
|
|
for m in history_msgs:
|
|
role = "USER" if m['role'] == 'user' else "ASSISTANT"
|
|
user += f"=== {role} ===\n"
|
|
for tr in m.get('tool_results', []): user += f'<tool_result>{tr["content"]}</tool_result>\n'
|
|
user += str(m['content']) + "\n"
|
|
self.total_cd_tokens += len(user) // 3
|
|
if self.total_cd_tokens > 9000: self.last_tools = ''
|
|
user += "=== ASSISTANT ===\n"
|
|
return system + user
|
|
|
|
def _parse_mixed_response(self, text):
|
|
remaining_text = text; thinking = ''
|
|
think_match = re.search(r"<think(?:ing)?>(.*?)</think(?:ing)?>", text, re.DOTALL)
|
|
if think_match:
|
|
thinking = think_match.group(1).strip()
|
|
remaining_text = re.sub(r"<think(?:ing)?>(.*?)</think(?:ing)?>", "", remaining_text, flags=re.DOTALL)
|
|
tool_calls, remaining_text = llmcore._parse_text_tool_calls(remaining_text)
|
|
if not tool_calls:
|
|
json_strs = []; errors = []
|
|
if '<tool_use>' in remaining_text:
|
|
weaktoolstr = remaining_text.split('<tool_use>')[-1].strip().strip('><')
|
|
json_str = weaktoolstr if weaktoolstr.endswith('}') else ''
|
|
if json_str == '' and '```' in weaktoolstr and weaktoolstr.split('```')[0].strip().endswith('}'):
|
|
json_str = weaktoolstr.split('```')[0].strip()
|
|
if json_str: json_strs.append(json_str)
|
|
remaining_text = remaining_text.replace('<tool_use>'+weaktoolstr, "")
|
|
elif '"name":' in remaining_text and '"arguments":' in remaining_text:
|
|
json_match = re.search(r'\{.*"name":.*\}', remaining_text, re.DOTALL)
|
|
if json_match:
|
|
json_strs.append(json_match.group(0).strip())
|
|
remaining_text = remaining_text.replace(json_match.group(0), "").strip()
|
|
for json_str in json_strs:
|
|
try:
|
|
data = llmcore.tryparse(json_str)
|
|
func_name = data.get('name') or data.get('function') or data.get('tool')
|
|
args = data.get('arguments') or data.get('args') or data.get('params') or data.get('parameters')
|
|
if args is None: args = data
|
|
if func_name: tool_calls.append(llmcore.MockToolCall(func_name, args))
|
|
except json.JSONDecodeError:
|
|
errors.append(f'Failed to parse tool_use JSON: {json_str[:200]}')
|
|
self.last_tools = ''
|
|
except: pass
|
|
if not tool_calls:
|
|
for e in errors:
|
|
print(f"[Warn] {e}"); tool_calls.append(llmcore.MockToolCall('bad_json', {'msg': e}))
|
|
return llmcore.MockResponse(thinking, remaining_text.strip(), tool_calls, text)
|
|
|
|
# 尾挂名: 回挂llmcore(谁后初始化完谁挂)
|
|
|
|
llmcore.ClaudeSession = ClaudeSession
|
|
llmcore.LLMSession = LLMSession
|
|
llmcore.ToolClient = ToolClient
|
|
|