""" 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 # P0-01：配置读取统一入口（环境变量 HAOCODE_CONFIG_FILE 优先，缺省回落项目内 data/config.json） from core.config_paths import load_config from core import platform_shell # P1-02：提示词平台段（通用正文 + 短 shell/path 段） 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；缺失时用兜底短提示词。 P1-02：通用正文 + 运行时插入的短平台 shell/path 段（占位符替换）； 每次请求重读本函数即重读文件，Windows 段不会出现在 Linux 请求中，反之亦然。 """ text = None try: with open(SYSTEM_PROMPT_FILE, "r", encoding="utf-8") as f: text = f.read().strip() except FileNotFoundError: pass if not text: text = _FALLBACK_SYSTEM_PROMPT return platform_shell.apply_platform_section(text) def _load_config() -> dict: """P0-01：统一走 core.config_paths.load_config（环境变量优先、容错、可见警告）。""" return load_config() 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))