sorrow404null
a7412824e0
Import the pre-repair source tree as the history baseline. Runtime data (data/), virtualenvs, bytecode caches and logs are gitignored so local secrets and user state stay out of the repo.
447 lines
20 KiB
Python
447 lines
20 KiB
Python
"""
|
||
core/llm_engine.py
|
||
==================
|
||
Qt 胶水层 —— 把 core.agent（pi 1:1 核心）接到 PyQt6 信号系统。
|
||
|
||
🌟 对照 pi 架构：
|
||
AgentWorker(QThread)
|
||
内部持有 Agent + AgentRunner（对照 pi 的 Agent + AgentSession 会话层）
|
||
run() = pi 的 _runAgentPrompt：prompt → 循环 post-agent-run 恢复 → settle
|
||
事件桥：AgentEvent → pyqtSignal（跨线程 queued 投递到 UI 线程）
|
||
|
||
ChatWorker(QThread)
|
||
chat 模式：普通聊天单次流式（无工具循环/无重试/无压缩）。
|
||
|
||
TitleWorker(QThread)
|
||
轻量单次流式补全（标题生成），保留原 LLMWorker 的信号面。
|
||
|
||
信号面（与旧 LLMWorker 兼容 + 新增工具事件）：
|
||
chunk_received(str) # text_delta
|
||
reasoning_received(str) # thinking_delta
|
||
error_occurred(str) # 终局错误（已重试/压缩恢复仍失败）
|
||
tool_execution_started(str, str, str) # (call_id, tool_name, args_json)
|
||
tool_execution_updated(str, str) # (call_id, 执行中增量输出)
|
||
tool_execution_finished(str, str, bool, str) # (call_id, tool_name, ok, 结果摘要)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from openai import OpenAI
|
||
from PyQt6.QtCore import QThread, pyqtSignal
|
||
|
||
from core.agent import (Agent, AgentConfig, AgentEvent, AgentMessage, AgentRunner,
|
||
ModelConfig, RetryConfig, calculate_context_tokens,
|
||
default_tools, from_openai_messages,
|
||
openai_stream)
|
||
from core.agent.stream_fn import _pick_reasoning, _pick_usage
|
||
|
||
CONFIG_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "config.json")
|
||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||
# 🌟 Agent 系统提示词文件（项目根目录，core 前面那个 .md）
|
||
SYSTEM_PROMPT_FILE = os.path.join(PROJECT_ROOT, "SYSTEM_PROMPT.md")
|
||
_FALLBACK_SYSTEM_PROMPT = (
|
||
"你是 haocode 的本地智能体，可以读取/写入文件并执行 bash 命令"
|
||
"（工作目录为项目根目录）。请用简体中文简洁地回答。"
|
||
)
|
||
|
||
|
||
def load_system_prompt() -> str:
|
||
"""读取 SYSTEM_PROMPT.md；缺失时用兜底短提示词。"""
|
||
try:
|
||
with open(SYSTEM_PROMPT_FILE, "r", encoding="utf-8") as f:
|
||
text = f.read().strip()
|
||
if text:
|
||
return text
|
||
except FileNotFoundError:
|
||
pass
|
||
return _FALLBACK_SYSTEM_PROMPT
|
||
|
||
|
||
def _load_config() -> dict:
|
||
if os.path.exists(CONFIG_PATH):
|
||
try:
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
print(f"[llm_engine] 读取配置失败: {e}")
|
||
return {}
|
||
|
||
|
||
def _provider_info(config: dict, provider_name: str) -> dict:
|
||
return config.get("providers", {}).get(provider_name, {}) or {}
|
||
|
||
|
||
class AgentWorker(QThread):
|
||
"""
|
||
一次「发送」= 一个 AgentWorker（对照 pi：一个 AgentSession 实例）。
|
||
多轮工具循环在一次 run() 内完成；UI 只管收信号 + 最终入库。
|
||
"""
|
||
|
||
# ---- 与旧 LLMWorker 兼容的信号 ----
|
||
chunk_received = pyqtSignal(str)
|
||
reasoning_received = pyqtSignal(str)
|
||
error_occurred = pyqtSignal(str)
|
||
# ---- 新增：工具执行事件（pi tool_execution_* 事件）----
|
||
tool_execution_started = pyqtSignal(str, str, str) # (call_id, name, args)
|
||
tool_execution_updated = pyqtSignal(str, str) # (call_id, 增量输出)
|
||
tool_execution_timed = pyqtSignal(str, int, int) # (call_id, 已耗秒, 超时秒) 🆕 bash 读秒
|
||
tool_execution_finished = pyqtSignal(str, str, bool, str) # (call_id, name, ok, 结果)
|
||
# 上下文压缩发生（UI 可提示「已自动压缩上下文」）
|
||
context_compacted = pyqtSignal(dict)
|
||
# 🆕 压缩开始（摘要 LLM 阻塞调用前）→ 前端显示「执行中」动态气泡
|
||
# payload: {summary, before, after, duration_ms, path}
|
||
compaction_started = pyqtSignal(str) # path: pre_prompt/overflow_compact/length_compact
|
||
# 🆕 P1: 收到本轮精确 usage（UI 上下文标签做 usage 锚定，含 system+tools）
|
||
usage_updated = pyqtSignal(dict)
|
||
# 🆕 M3: 重试调度 / 重试结果（对照 pi onRetryScheduled/onRetryFinished）
|
||
retry_scheduled = pyqtSignal(int, int, float, str) # (attempt, max_attempts, delay_ms, reason)
|
||
retry_finished = pyqtSignal(bool)
|
||
|
||
def __init__(self, provider_name: str, model_name: str,
|
||
openai_messages: List[Dict[str, Any]],
|
||
tool_context: Optional[Dict[str, Any]] = None,
|
||
enable_tools: bool = True,
|
||
temperature: Optional[float] = None):
|
||
"""
|
||
openai_messages: build_api_context 的产物（OpenAI 格式）。
|
||
最后一条 user 消息 = 本次提问；其余 = 历史上下文。
|
||
"""
|
||
super().__init__()
|
||
self.provider_name = provider_name
|
||
self.model_name = model_name
|
||
self.openai_messages = openai_messages
|
||
self.tool_context = tool_context or {"cwd": PROJECT_ROOT}
|
||
self.enable_tools = enable_tools
|
||
self._temperature_override = temperature
|
||
|
||
self.config = _load_config()
|
||
self._aborted = False
|
||
self._agent: Optional[Agent] = None
|
||
self._runner: Optional[AgentRunner] = None
|
||
self._compactions_before = 0
|
||
|
||
# ------------------------------------------------------------------
|
||
# 中断（对照 pi abort：当前流立刻停止，已产生的内容保留）
|
||
# ------------------------------------------------------------------
|
||
def abort(self):
|
||
self._aborted = True
|
||
if self._agent is not None:
|
||
self._agent.abort()
|
||
|
||
# 兼容旧调用名
|
||
def cancel(self):
|
||
self.abort()
|
||
|
||
# ------------------------------------------------------------------
|
||
# 模型配置（对照 pi models.json 条目 → ModelConfig）
|
||
# ------------------------------------------------------------------
|
||
def _build_model_config(self) -> ModelConfig:
|
||
p = _provider_info(self.config, self.provider_name)
|
||
context_window = int(p.get("model_contexts", {}).get(self.model_name, 128000))
|
||
max_tokens = int(p.get("model_max_tokens", {}).get(self.model_name, 8192))
|
||
temperature = (self._temperature_override
|
||
if self._temperature_override is not None
|
||
else float(self.config.get("temperature", 0.7)))
|
||
return ModelConfig(
|
||
provider=self.provider_name,
|
||
name=self.model_name,
|
||
context_window=context_window,
|
||
max_tokens=max_tokens,
|
||
temperature=temperature,
|
||
api_key=p.get("api_key", ""),
|
||
base_url=p.get("base_url", "https://api.openai.com/v1"),
|
||
# 🆕 thinking reasoning_content 回传默认开启（ModelConfig.pass_reasoning=True）
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 压缩用的非流式 LLM 调用（对照 pi completeSimpleWithRetries）
|
||
# 新签名 1:1 对照 pi：(prompt_text, system_prompt, max_tokens) -> str
|
||
# ------------------------------------------------------------------
|
||
def _make_summarize_fn(self, model: ModelConfig):
|
||
def summarize(prompt_text: str, system_prompt: str, max_tokens: int) -> str:
|
||
client = OpenAI(api_key=model.api_key, base_url=model.base_url,
|
||
timeout=120.0, max_retries=0)
|
||
resp = client.chat.completions.create(
|
||
model=model.name,
|
||
messages=[
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": prompt_text},
|
||
],
|
||
stream=False,
|
||
max_tokens=max_tokens,
|
||
temperature=0.3,
|
||
)
|
||
return (resp.choices[0].message.content or "").strip()
|
||
|
||
return summarize
|
||
|
||
# ------------------------------------------------------------------
|
||
# 事件桥：AgentEvent → Qt 信号（在工作线程内 emit，queued 投递）
|
||
# ------------------------------------------------------------------
|
||
def _bridge(self, e: AgentEvent):
|
||
if e.type == "message_end" and e.message is not None:
|
||
# 🆕 P1: 本轮刚结束 → 它的 usage 是「上一张照片」（精确覆盖
|
||
# system+tools+历史+本轮输出），供 UI 显示锚定
|
||
usage = e.message.usage or {}
|
||
if calculate_context_tokens(usage) > 0:
|
||
self.usage_updated.emit(dict(usage))
|
||
if e.type == "message_update" and e.assistant_message_event is not None:
|
||
ev = e.assistant_message_event
|
||
# 🌟 字段是 ev.text（不是 ev.data）；_emit 会吞掉订阅者异常，
|
||
# 写错字段 = 静默无流式输出，必须靠 test_agent_worker_stream_fn_wiring 守住
|
||
if ev.type == "text_delta" and ev.text:
|
||
self.chunk_received.emit(ev.text)
|
||
elif ev.type == "thinking_delta" and ev.text:
|
||
self.reasoning_received.emit(ev.text)
|
||
return
|
||
if e.type == "tool_execution_start" and e.tool_call is not None:
|
||
try:
|
||
args = json.dumps(e.tool_call.arguments, ensure_ascii=False) \
|
||
if e.tool_call.arguments else (e.tool_call.raw_arguments or "")
|
||
except Exception:
|
||
args = str(e.tool_call.arguments)
|
||
self.tool_execution_started.emit(e.tool_call.id, e.tool_call.name,
|
||
str(args)[:2000])
|
||
elif e.type == "tool_execution_update" and e.arg:
|
||
self.tool_execution_updated.emit(e.tool_call.id, str(e.arg))
|
||
elif e.type == "tool_execution_timer" and isinstance(
|
||
getattr(e, "arg", None), tuple) and len(e.arg) == 2:
|
||
# 🆕 bash 运行中每秒滴一次 → 前端气泡读秒
|
||
self.tool_execution_timed.emit(e.tool_call.id, int(e.arg[0]),
|
||
int(e.arg[1]))
|
||
elif e.type == "tool_execution_end" and e.tool_call is not None:
|
||
text = ""
|
||
if e.result is not None:
|
||
c = e.result.content
|
||
text = c if isinstance(c, str) else \
|
||
"".join(x.get("text", "") for x in c if isinstance(x, dict))
|
||
# 🌟 修复: 原来 ok = is_error（反了）——成功的工具被显示成失败
|
||
ok = not (e.result.is_error if e.result is not None else True)
|
||
self.tool_execution_finished.emit(
|
||
e.tool_call.id, e.tool_call.name, bool(ok), text[:20000]) # 长结果供 UI 展开
|
||
|
||
# ------------------------------------------------------------------
|
||
# 主入口 —— 对照 pi _runAgentPrompt
|
||
# ------------------------------------------------------------------
|
||
def run(self):
|
||
provider = _provider_info(self.config, self.provider_name)
|
||
if not provider:
|
||
self.error_occurred.emit(f"未找到 provider 配置: {self.provider_name}")
|
||
return
|
||
|
||
history = from_openai_messages(self.openai_messages)
|
||
if not history or history[-1].role != "user":
|
||
self.error_occurred.emit("上下文必须以 user 消息结尾")
|
||
return
|
||
last_user = history.pop() # 本次提问（其余为历史）
|
||
|
||
model = self._build_model_config()
|
||
cfg = AgentConfig(
|
||
model=model,
|
||
# 🌟 系统提示词：SYSTEM_PROMPT.md（每次请求头部注入，不入历史）
|
||
system_prompt=load_system_prompt() if self.enable_tools else "",
|
||
tools=default_tools() if self.enable_tools else [],
|
||
tool_context=self.tool_context,
|
||
retry=RetryConfig(max_attempts=3, base_delay_ms=2000, factor=2.0),
|
||
)
|
||
agent = Agent(cfg)
|
||
# 🌟 关键：注入流函数（对照 pi 的 agentLoopConfig.streamFn 注入点）。
|
||
# 漏掉这一步 Agent 会抛 AgentError("未配置 stream_fn")。
|
||
agent.set_stream_fn(openai_stream)
|
||
agent.state.messages = list(history)
|
||
agent.subscribe(self._bridge)
|
||
|
||
summarize = self._make_summarize_fn(model)
|
||
runner = AgentRunner(agent, summarize_fn=summarize,
|
||
# 🆕 M3: 重试回调 → Qt 信号（跨线程 queued 投递）
|
||
on_retry_scheduled=lambda a, m, d, r:
|
||
self.retry_scheduled.emit(int(a), int(m), float(d), str(r)),
|
||
on_retry_finished=lambda ok:
|
||
self.retry_finished.emit(bool(ok)),
|
||
# 🆕 压缩开始回调 → 前端动态气泡
|
||
on_compaction_started=lambda p:
|
||
self.compaction_started.emit(str(p)),
|
||
# 🆕 压缩完成回调 → 气泡即时定格（不等整轮结束）
|
||
on_compaction_finished=lambda p:
|
||
self.context_compacted.emit(dict(p)))
|
||
# 🆕 接线轮中主动压缩检查（haocode 增强，偏离 pi 1:1）：
|
||
# 内层循环每次 LLM 请求前主动检查，单条巨型工具输出不再依赖
|
||
# 「一次失败往返 + 响应式兜底」
|
||
agent.config.compact_fn = runner.compact_if_needed
|
||
self._agent = agent
|
||
self._runner = runner
|
||
self._compactions_before = 0
|
||
|
||
try:
|
||
result = runner.run(last_user)
|
||
except Exception as e:
|
||
if not self._aborted:
|
||
self.error_occurred.emit(f"\n[Agent 异常]: {e}")
|
||
return
|
||
|
||
if self._aborted:
|
||
return # 取消路径：UI 已自行清理，不回退时间线
|
||
|
||
# 注：压缩事件（context_compacted）已由 on_compaction_finished 在压缩结束
|
||
# 瞬间实时发出（气泡即时定格），这里不再重复 flush runner.compaction_events
|
||
|
||
if result.error is not None and result.stop_reason == "error":
|
||
# 恢复逻辑（重试 + 压缩）全部用尽后的终局错误
|
||
self.error_occurred.emit(
|
||
f"\n[API 请求异常]: {result.error.message or '未知错误'}")
|
||
|
||
# ------------------------------------------------------------------
|
||
# 供 UI 读取最终状态（入库用）
|
||
# ------------------------------------------------------------------
|
||
def final_messages(self) -> List[AgentMessage]:
|
||
return list(self._agent.state.messages) if self._agent else []
|
||
|
||
|
||
class ChatWorker(QThread):
|
||
"""
|
||
chat 模式 —— 普通聊天：单次流式补全（对照旧 LLMWorker 的行为）。
|
||
不进入 agent 循环：无工具调用、无重试、无压缩。
|
||
worker 模式请用 AgentWorker。
|
||
"""
|
||
|
||
chunk_received = pyqtSignal(str)
|
||
reasoning_received = pyqtSignal(str)
|
||
error_occurred = pyqtSignal(str)
|
||
# 🆕 P1: 收到精确 usage（UI 上下文标签 usage 锚定）
|
||
usage_updated = pyqtSignal(dict)
|
||
|
||
def __init__(self, provider_name: str, model_name: str,
|
||
openai_messages: List[Dict[str, Any]],
|
||
temperature: Optional[float] = None):
|
||
super().__init__()
|
||
self.provider_name = provider_name
|
||
self.model_name = model_name
|
||
self.openai_messages = openai_messages
|
||
self._temperature_override = temperature
|
||
self.config = _load_config()
|
||
self._cancelled = False
|
||
|
||
def cancel(self):
|
||
self._cancelled = True
|
||
|
||
def abort(self):
|
||
self._cancelled = True
|
||
|
||
def run(self):
|
||
p = _provider_info(self.config, self.provider_name)
|
||
if not p:
|
||
self.error_occurred.emit(f"未找到 provider 配置: {self.provider_name}")
|
||
return
|
||
model = self._build_model_config(p)
|
||
max_tokens = int(p.get("model_max_tokens", {}).get(self.model_name, 8192))
|
||
try:
|
||
client = OpenAI(api_key=p.get("api_key", ""),
|
||
base_url=p.get("base_url", "https://api.openai.com/v1"),
|
||
timeout=model.timeout_seconds, max_retries=0)
|
||
response = client.chat.completions.create(
|
||
model=self.model_name,
|
||
messages=self.openai_messages,
|
||
stream=True,
|
||
temperature=model.temperature,
|
||
max_tokens=max_tokens,
|
||
# 🆕 P1: 拿精确 usage 供上下文标签锚定（对照 pi stream_options）
|
||
stream_options={"include_usage": True},
|
||
)
|
||
for chunk in response:
|
||
if self._cancelled:
|
||
try:
|
||
response.close()
|
||
except Exception:
|
||
pass
|
||
return
|
||
# 🆕 P1/M2: usage 读取（含 choice 兜底）
|
||
_u = _pick_usage(chunk)
|
||
if _u is not None and (
|
||
getattr(_u, "prompt_tokens", 0) or getattr(_u, "completion_tokens", 0)):
|
||
self.usage_updated.emit({
|
||
"input": getattr(_u, "prompt_tokens", 0) or 0,
|
||
"output": getattr(_u, "completion_tokens", 0) or 0,
|
||
"cacheRead": 0, "cacheWrite": 0,
|
||
})
|
||
if not chunk.choices:
|
||
continue
|
||
delta = chunk.choices[0].delta
|
||
# 🆕 M1: 思考字段优先级对照 pi（reasoning_content/reasoning/reasoning_text）
|
||
_r = _pick_reasoning(delta)
|
||
if _r:
|
||
self.reasoning_received.emit(_r)
|
||
if delta.content:
|
||
self.chunk_received.emit(delta.content)
|
||
except Exception as e:
|
||
if not self._cancelled:
|
||
self.error_occurred.emit(str(e))
|
||
|
||
def _build_model_config(self, p: dict) -> ModelConfig:
|
||
context_window = int(p.get("model_contexts", {}).get(self.model_name, 128000))
|
||
temperature = (self._temperature_override
|
||
if self._temperature_override is not None
|
||
else float(self.config.get("temperature", 0.7)))
|
||
return ModelConfig(
|
||
provider=self.provider_name,
|
||
name=self.model_name,
|
||
context_window=context_window,
|
||
max_tokens=int(p.get("model_max_tokens", {}).get(self.model_name, 8192)),
|
||
temperature=temperature,
|
||
api_key=p.get("api_key", ""),
|
||
base_url=p.get("base_url", "https://api.openai.com/v1"),
|
||
# 🆕 thinking reasoning_content 回传默认开启
|
||
)
|
||
|
||
|
||
class TitleWorker(QThread):
|
||
"""标题生成：单次流式补全（保留原信号面）"""
|
||
|
||
chunk_received = pyqtSignal(str)
|
||
error_occurred = pyqtSignal(str)
|
||
|
||
def __init__(self, provider_name: str, model_name: str,
|
||
openai_messages: List[Dict[str, Any]]):
|
||
super().__init__()
|
||
self.provider_name = provider_name
|
||
self.model_name = model_name
|
||
self.openai_messages = openai_messages
|
||
self.config = _load_config()
|
||
self._cancelled = False
|
||
|
||
def cancel(self):
|
||
self._cancelled = True
|
||
|
||
def run(self):
|
||
p = _provider_info(self.config, self.provider_name)
|
||
if not p:
|
||
return
|
||
try:
|
||
client = OpenAI(api_key=p.get("api_key", ""),
|
||
base_url=p.get("base_url", "https://api.openai.com/v1"),
|
||
timeout=60.0, max_retries=0)
|
||
temperature = float(self.config.get("temperature", 0.7))
|
||
response = client.chat.completions.create(
|
||
model=self.model_name,
|
||
messages=self.openai_messages,
|
||
stream=True,
|
||
temperature=temperature,
|
||
max_tokens=60,
|
||
)
|
||
for chunk in response:
|
||
if self._cancelled:
|
||
try:
|
||
response.close()
|
||
except Exception:
|
||
pass
|
||
return
|
||
if chunk.choices:
|
||
delta = chunk.choices[0].delta
|
||
if delta.content:
|
||
self.chunk_received.emit(delta.content)
|
||
except Exception as e:
|
||
if not self._cancelled:
|
||
self.error_occurred.emit(str(e))
|