# -*- coding: utf-8 -*- """OpenAI 插件卷路径发现、容器内文件读写、plugin_daemon 重启。""" from __future__ import annotations import re import tempfile from pathlib import Path import _config as C def find_plugin_file(d, suffix: str) -> str: """suffix 例如 models/llm/llm.py 或 models/llm/chat.py""" out, err = d.sh( "docker exec %s sh -c 'ls -d /app/storage/cwd/langgenius/openai-*@*/%s 2>/dev/null | head -1'" % (C.PLUGIN_CONTAINER, suffix) ) path = (out or err).strip() if not path or "No such file" in path: raise RuntimeError("找不到插件文件 %s: %s" % (suffix, out or err)) return path def container_read(d, container: str, path: str) -> str: out, err = d.sh("docker exec %s cat %s" % (container, path)) if not out and err: raise RuntimeError(err) # normalize CRLF -> LF so patch string comparisons work uniformly return out.replace("\r\n", "\n") def container_write(d, container: str, path: str, content: str) -> None: local = Path(tempfile.gettempdir()) / ("patch_%s.py" % container.replace("-", "_")) local.write_text(content, encoding="utf-8") remote = "/tmp/%s" % local.name d.sftp.put(str(local), remote) out, err = d.sh("docker cp %s %s:%s" % (remote, container, path)) if err and "Error" in err: raise RuntimeError(err) def grep_in_container(d, container: str, path: str, pattern: str) -> str: safe = pattern.replace("'", "'\\''") out, err = d.sh( "docker exec %s grep -Fn '%s' %s 2>/dev/null || true" % (container, safe, path) ) return out or err def restart_plugin_daemon(d) -> None: out, err = d.sh( "cd %s 2>/dev/null || cd /dify/dify/docker; docker compose restart plugin_daemon" % C.DIFY_DOCKER_DIR, t=120, ) print(out or err) LLM_PATCH_MARKER = "Fallback for custom / non-OpenAI base models" CHAT_PATCH_MARKER = "_ensure_system_first" LF_TRACE_MARKER = "Langfuse v4 requires observation body.id; auto-assign if missing" LF_AGENT_TOOL_SPANS_MARKER = "# >>> AGENT TOOL SPANS PATCH" LF_AGENT_TOOL_SPANS_MARKER_V2 = "# >>> AGENT TOOL SPANS PATCH v2" LF_AGENT_TOOL_SPANS_MARKER_V2 = "# >>> AGENT TOOL SPANS PATCH v2" NEW_GET_CUSTOMIZABLE_MODEL_SCHEMA = ''' def get_customizable_model_schema( self, model: str, credentials: dict ) -> AIModelEntity: schema = {item.model: item for item in self.predefined_models()}.get( _base_model(model) ) if schema is None: # Fallback for custom / non-OpenAI base models served via an # OpenAI-compatible endpoint (e.g. Qwen, DeepSeek, local vLLM): # clone a chat-capable predefined schema so the model is listable # and can be set as the default model. template = next( (m for m in self.predefined_models() if m.model_properties.get("mode") == "chat"), None, ) if template is None: template = self.predefined_models()[0] props = dict(template.model_properties) mp = (credentials or {}).get("model_properties") or {} VLLM_MAX_CTX = 64000 try: props["context_size"] = int(mp["context_size"]) if "context_size" in mp else VLLM_MAX_CTX except (ValueError, TypeError): props["context_size"] = VLLM_MAX_CTX schema = template.model_copy(deep=True, update={"model_properties": props}) return schema.model_copy( deep=True, update={ "model": model, "label": I18nObject(zh_hans=model, en_us=model), "fetch_from": FetchFrom.CUSTOMIZABLE_MODEL, }, ) ''' ENSURE_SYSTEM_FIRST_FN = ''' def _ensure_system_first(messages: list[PromptMessage]) -> list[PromptMessage]: """Qwen3/vLLM chat templates accept ONLY one system message, at the start.""" system_parts: list[str] = [] rest: list[PromptMessage] = [] for m in messages: if isinstance(m, (SystemPromptMessage, DeveloperPromptMessage)): text = _text_content(m) if text.strip(): system_parts.append(text) else: rest.append(m) if system_parts: rest.insert(0, SystemPromptMessage(content="\\n\\n".join(system_parts))) return rest ''' def patch_llm_schema(content: str) -> str: if LLM_PATCH_MARKER in content: return content m = re.search( r"\n def get_customizable_model_schema\([\s\S]*?\n def get_num_tokens\(", content, ) if not m: raise RuntimeError("无法在 llm.py 中定位 get_customizable_model_schema") replacement = NEW_GET_CUSTOMIZABLE_MODEL_SCHEMA + "\n def get_num_tokens(" return content[: m.start()] + replacement + content[m.end() :] def patch_chat_system(content: str) -> str: if CHAT_PATCH_MARKER in content: return content lines = content.splitlines(keepends=True) insert_at = 0 for i, line in enumerate(lines): if line.strip() == "from __future__ import annotations": insert_at = i + 1 break if insert_at == 0: raise RuntimeError("chat.py 中未找到 from __future__ import annotations") # 跳过 future 后连续的空行 while insert_at < len(lines) and lines[insert_at].strip() == "": insert_at += 1 fn_block = ENSURE_SYSTEM_FIRST_FN.strip("\n") + "\n\n" content = "".join(lines[:insert_at]) + fn_block + "".join(lines[insert_at:]) old_call = "messages=cast(Any, [message(item) for item in prompt_messages])" new_call = "messages=cast(Any, [message(item) for item in _ensure_system_first(prompt_messages)])" if old_call not in content: raise RuntimeError("chat.py 中未找到 messages= prompt_messages 调用点") return content.replace(old_call, new_call, 1) def apply_langfuse_trace_patch(content: str) -> str: if LF_TRACE_MARKER in content: return content old_span = """ data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {} try: body = CreateSpanBody( id=data.get("id"), """ new_span = """ data = filter_none_values(langfuse_span_data.model_dump()) if langfuse_span_data else {} try: # Langfuse v4 requires observation body.id; auto-assign if missing if not data.get("id"): data["id"] = str(uuid.uuid4()) body = CreateSpanBody( id=data.get("id"), """ old_gen = """ usage_data = data.pop("usage", None) usage = None if usage_data: usage = Usage( input=usage_data.get("input", 0) or 0, output=usage_data.get("output", 0) or 0, total=usage_data.get("total", 0) or 0, unit=usage_data.get("unit"), input_cost=usage_data.get("inputCost"), output_cost=usage_data.get("outputCost"), total_cost=usage_data.get("totalCost"), ) body = CreateGenerationBody( id=data.get("id"), """ new_gen = """ usage_data = data.pop("usage", None) usage = None if usage_data: usage = Usage( input=usage_data.get("input", 0) or 0, output=usage_data.get("output", 0) or 0, total=usage_data.get("total", 0) or 0, unit=usage_data.get("unit"), input_cost=usage_data.get("inputCost"), output_cost=usage_data.get("outputCost"), total_cost=usage_data.get("totalCost"), ) # Langfuse v4 requires observation body.id; auto-assign if missing if not data.get("id"): data["id"] = str(uuid.uuid4()) body = CreateGenerationBody( id=data.get("id"), """ if old_span not in content: raise RuntimeError("langfuse_trace.py span 锚点未找到") if old_gen not in content: raise RuntimeError("langfuse_trace.py generation 锚点未找到") content = content.replace(old_span, new_span, 1).replace(old_gen, new_gen, 1) return content def apply_langfuse_agent_tool_spans_patch(content: str) -> str: """在 langfuse_trace.py 的 message_trace() 内补写 mode=agent 的工具调用 spans。 版本 v2：改用与 ops_trace_manager.tool_trace() 相同的字段名 （agent_thought.tools / .tool_meta / .observation / .created_at）， 以及 base_agent_runner 保存的 tool_invoke_meta 等字段。 """ if LF_AGENT_TOOL_SPANS_MARKER_V2 in content: # already has v2 block, but might be missing: # 1) stable generation id assignment: langfuse_generation_data.id = message_id # 2) parent linking on tool span: parent_observation_id=langfuse_generation_data.id has_parent = "parent_observation_id=langfuse_generation_data.id" in content has_gen_id = "langfuse_generation_data.id = message_id" in content if has_parent and has_gen_id: return content # targeted fix #1: inject generation id assignment before add_generation old_start = ( " self.add_generation(langfuse_generation_data)\n\n " + LF_AGENT_TOOL_SPANS_MARKER_V2 + "\n" ) new_start = ( " try:\n" " langfuse_generation_data.id = message_id\n" " except Exception:\n" " pass\n\n" " self.add_generation(langfuse_generation_data)\n\n " + LF_AGENT_TOOL_SPANS_MARKER_V2 + "\n" ) if old_start in content: content = content.replace(old_start, new_start, 1) # targeted fix #2: add parent_observation_id to tool spans old_trace_line = " trace_id=trace_id,\n start_time=start_t,\n" new_trace_line = ( " trace_id=trace_id,\n" " parent_observation_id=langfuse_generation_data.id,\n" " start_time=start_t,\n" ) if old_trace_line in content: content = content.replace(old_trace_line, new_trace_line, 1) return content # 判断是否有旧版补丁，若有则把旧块整体换成 v2 if LF_AGENT_TOOL_SPANS_MARKER in content: # 旧版补丁结束标志（紧接 moderation_trace 函数定义） old_block_start = "\n " + LF_AGENT_TOOL_SPANS_MARKER + "\n" # 找到旧块结束点：紧跟旧补丁最后一行的 " def moderation_trace" old_block_end_marker = "agent tool spans patch failed" if old_block_start not in content or old_block_end_marker not in content: raise RuntimeError("找到旧版 PATCH marker 但无法定位完整旧块边界") idx_start = content.index(old_block_start) # 从 idx_start 之后找到 " def moderation_trace" 的位置 idx_mod = content.index("\n def moderation_trace", idx_start) # 切掉整段旧补丁（从 \n # >>> ... 到 \n def moderation_trace 之前） content = content[:idx_start] + content[idx_mod:] # 现在 content 在 add_generation 之后直接有 " def moderation_trace" old_anchor = ( " self.add_generation(langfuse_generation_data)\n\n" " def moderation_trace" ) if old_anchor not in content: raise RuntimeError("清除旧块后锚点仍未找到，请手动检查容器文件") else: old_anchor = ( " self.add_generation(langfuse_generation_data)\n\n" " def moderation_trace" ) if old_anchor not in content: raise RuntimeError("langfuse_trace.py message_trace 锚点未找到") # 完整的补丁插入块（以换行形式展开，避免 f-string 混乱） patch_code = ( " " + LF_AGENT_TOOL_SPANS_MARKER_V2 + "\n" " # ensure generation id stable so tool spans can link as children\n" " try:\n" " langfuse_generation_data.id = message_id\n" " except Exception:\n" " pass\n" "\n" " self.add_generation(langfuse_generation_data)\n" "\n" " # mode=agent: derive tool spans from message_data.agent_thoughts\n" " # Uses same fields as ops_trace_manager.tool_trace():\n" " # agent_thought.tools / .tool_meta / .observation / .created_at\n" " try:\n" " agent_thoughts_list = getattr(message_data, 'agent_thoughts', None) or []\n" " for at in agent_thoughts_list:\n" " # tools is a comma-separated or list of tool names\n" " raw_tools = getattr(at, 'tools', None)\n" " if not raw_tools:\n" " continue\n" " if isinstance(raw_tools, str):\n" " tool_name_list = [t.strip() for t in raw_tools.split(',') if t.strip()]\n" " elif isinstance(raw_tools, list):\n" " tool_name_list = [str(t) for t in raw_tools if t]\n" " else:\n" " tool_name_list = []\n" "\n" " raw_obs = getattr(at, 'observation', None) or ''\n" " obs_map = {}\n" " if isinstance(raw_obs, dict):\n" " obs_map = raw_obs\n" " elif isinstance(raw_obs, str) and raw_obs:\n" " # observation is a single string; map to first tool\n" " if tool_name_list:\n" " obs_map = {tool_name_list[0]: raw_obs}\n" "\n" " raw_tool_input = getattr(at, 'tool_input', None) or ''\n" " input_map = {}\n" " if isinstance(raw_tool_input, dict):\n" " input_map = raw_tool_input\n" " elif isinstance(raw_tool_input, str) and raw_tool_input:\n" " try:\n" " parsed = json.loads(raw_tool_input)\n" " if isinstance(parsed, dict):\n" " input_map = parsed\n" " else:\n" " input_map = {tool_name_list[0]: raw_tool_input} if tool_name_list else {}\n" " except Exception:\n" " input_map = {tool_name_list[0]: raw_tool_input} if tool_name_list else {}\n" "\n" " tool_meta_raw = getattr(at, 'tool_meta', None) or {}\n" " at_start = getattr(at, 'created_at', None) or trace_info.start_time\n" "\n" " for tn in tool_name_list:\n" " tool_output = obs_map.get(tn, obs_map.get(None) if len(tool_name_list) == 1 else None)\n" " if tool_output is None and len(obs_map) == 1:\n" " tool_output = next(iter(obs_map.values()), None)\n" " tool_inp = input_map.get(tn, input_map.get(None))\n" " if tool_inp is None and len(input_map) == 1:\n" " tool_inp = next(iter(input_map.values()), None)\n" "\n" " tmeta = tool_meta_raw.get(tn, {}) if isinstance(tool_meta_raw, dict) else {}\n" " time_cost_s = tmeta.get('time_cost', 0) or 0\n" " try:\n" " time_cost_f = float(time_cost_s)\n" " except (TypeError, ValueError):\n" " time_cost_f = 0.0\n" "\n" " start_t = at_start\n" " end_t = start_t + timedelta(seconds=time_cost_f) if start_t else None\n" " err = tmeta.get('error')\n" "\n" " tool_span = LangfuseSpan(\n" " name=tn,\n" " input=tool_inp,\n" " output=tool_output,\n" " trace_id=trace_id,\n" " parent_observation_id=langfuse_generation_data.id,\n" " start_time=start_t,\n" " end_time=end_t,\n" " metadata={**metadata, 'tool_name': tn, 'time_cost': time_cost_f},\n" " level=(LevelEnum.DEFAULT if not err else LevelEnum.ERROR),\n" " status_message=err,\n" " )\n" " self.add_span(langfuse_span_data=tool_span)\n" " except Exception:\n" " logger.exception('agent tool spans patch v2 failed')\n" "\n" " def moderation_trace" ) return content.replace(old_anchor, patch_code, 1)