From 1b4c07fec6fca9a77546db02de674dead8f6f114 Mon Sep 17 00:00:00 2001 From: dark Date: Fri, 5 Jun 2026 13:18:52 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E8=BF=87=E6=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pam_deploy_graph/llm/prompts.py | 6 +- pam_deploy_graph/llm/text_filter.py | 165 ++++++++++++++++++++++++++-- tests/test_llm_text_filter.py | 54 +++++++++ 3 files changed, 216 insertions(+), 9 deletions(-) diff --git a/pam_deploy_graph/llm/prompts.py b/pam_deploy_graph/llm/prompts.py index 9ba03f1..8a65ecf 100644 --- a/pam_deploy_graph/llm/prompts.py +++ b/pam_deploy_graph/llm/prompts.py @@ -109,7 +109,8 @@ CHAT_PROMPT = """你是 PAM 部署 Agent 的交互助手。 - 如果用户想执行完整部署，提示使用 `analyze <需求>` 先分析，确认后再输入 `run`。 - 如果用户想单独执行 action，提示使用 `action propose <需求>` 或 `action run ...`，执行前仍需要人工确认。 - 不要输出密钥、token、Authorization、CLIENT_SECRET 或 api_key。 -- 不要输出 ``、``、推理过程、内部思考或隐藏分析内容。 +- 不要输出 ``、``、`Thinking Process`、`Reasoning Process`、`Chain of Thought`、推理过程、内部思考或隐藏分析内容。 +- 只输出可以直接展示给用户的最终回答。 """ LOG_ANALYSIS_PROMPT = """分析 PAM Agent 或部署脚本日志。 @@ -119,7 +120,8 @@ LOG_ANALYSIS_PROMPT = """分析 PAM Agent 或部署脚本日志。 - 不要输出密钥、token、Authorization、CLIENT_SECRET 或 api_key。 - 输入通常是日志尾部摘要，不代表完整文件。 - 不要因为日志来自 stderr 就直接判定失败，要结合 ERROR、Exception、fail、状态码和上下文判断。 -- 不要输出 ``、``、推理过程、内部思考或隐藏分析内容。 +- 不要输出 ``、``、`Thinking Process`、`Reasoning Process`、`Chain of Thought`、推理过程、内部思考或隐藏分析内容。 +- 只输出可以直接展示给用户的最终分析结果。 """ SINGLE_ACTION_PROMPT = """把用户自然语言解析成一次 PAM action 调用建议。 diff --git a/pam_deploy_graph/llm/text_filter.py b/pam_deploy_graph/llm/text_filter.py index 31bcc78..8dcb60f 100644 --- a/pam_deploy_graph/llm/text_filter.py +++ b/pam_deploy_graph/llm/text_filter.py @@ -3,20 +3,44 @@ from __future__ import annotations from collections.abc import Iterable, Iterator +import re OPEN_THINK_TAG = "" CLOSE_THINK_TAG = "" +REASONING_START_RE = re.compile( + r"^\s*(?:[#>\-*]+\s*)*(?:\*\*)?" + r"(?:thinking process|thought process|reasoning process|chain of thought|internal reasoning|inner monologue|" + r"思考过程|推理过程|内部思考)" + r"(?:\*\*)?\s*(?:[:：]|\s*$)", + flags=re.IGNORECASE, +) +FINAL_ANSWER_RE = re.compile( + r"^\s*(?:[#>\-*]+\s*)*(?:\*\*)?" + r"(?:final answer|final response|answer|response|最终答案|最终回答|正式回答|回答|回复|结论)" + r"(?:\*\*)?\s*[:：]\s*", + flags=re.IGNORECASE | re.MULTILINE, +) +REASONING_LINE_RE = re.compile( + r"(thinking process|thought process|reasoning process|chain of thought|internal reasoning|inner monologue|" + r"analyze the request|determine the response|drafting the response|refining the response|" + r"user question|input json|role:|constraints:|requirements:|" + r"do not output|do not automatically|hidden analysis|forbidden tags|" + r"i need to|i should|i must|i will|i can|must ensure|should briefly|ensure no|keep it concise|" + r"思考过程|推理过程|内部思考|分析请求|确定回答|起草回答|优化回答|隐藏分析)", + flags=re.IGNORECASE, +) +MAX_REASONING_PREFIX_HOLD = 80 def strip_thinking_text(text: str) -> str: - """移除 LLM 普通文本输出里的思考标签和内容。""" + """移除 LLM 普通文本输出里的思考标签、显式思考段和内容。""" filter_ = ThinkingTextStreamFilter() visible = filter_.feed(text) + filter_.finish() return visible.strip() def filter_thinking_chunks(chunks: Iterable[str]) -> Iterator[str]: - """按流式分片移除 `...`，避免跨分片泄露思考内容。""" + """按流式分片移除思考内容，避免跨分片泄露。""" filter_ = ThinkingTextStreamFilter() for chunk in chunks: visible = filter_.feed(str(chunk)) @@ -28,12 +52,13 @@ def filter_thinking_chunks(chunks: Iterable[str]) -> Iterator[str]: class ThinkingTextStreamFilter: - """支持跨 chunk 识别 think 标签的流式过滤器。""" + """支持跨 chunk 识别 think 标签和显式思考段的流式过滤器。""" def __init__(self) -> None: """初始化可见/隐藏状态和待判定缓冲区。""" self._pending = "" self._inside_think = False + self._reasoning_filter = ExplicitReasoningStreamFilter() def feed(self, chunk: str) -> str: """输入一个文本分片，返回当前可安全展示的可见文本。""" @@ -73,21 +98,147 @@ class ThinkingTextStreamFilter: output.append(self._pending) self._pending = "" break - return "".join(output) + return self._reasoning_filter.feed("".join(output)) def finish(self) -> str: """结束流式过滤，丢弃未闭合 think 内容和未完成标签。""" if self._inside_think: self._pending = "" self._inside_think = False - return "" + return self._reasoning_filter.finish() lowered = self._pending.lower() if lowered in _tag_prefixes(): self._pending = "" - return "" + return self._reasoning_filter.finish() tail = self._pending self._pending = "" - return tail + return self._reasoning_filter.feed(tail) + self._reasoning_filter.finish() + + +class ExplicitReasoningStreamFilter: + """过滤以 `Thinking Process:` 等形式输出的显式思考段。""" + + def __init__(self) -> None: + """初始化思考段识别状态。""" + self._buffer = "" + self._mode = "undecided" + + def feed(self, chunk: str) -> str: + """输入已去掉 think 标签的文本，返回可展示内容。""" + if not chunk: + return "" + if self._mode == "pass": + return chunk + self._buffer += chunk + if self._mode == "suppress": + final_text = _extract_after_final_answer_marker(self._buffer) + if final_text is not None: + self._buffer = "" + self._mode = "pass" + return final_text + return "" + if _starts_with_reasoning_marker(self._buffer): + self._mode = "suppress" + final_text = _extract_after_final_answer_marker(self._buffer) + if final_text is not None: + self._buffer = "" + self._mode = "pass" + return final_text + return "" + if _could_be_reasoning_marker_prefix(self._buffer): + return "" + self._mode = "pass" + visible = self._buffer + self._buffer = "" + return visible + + def finish(self) -> str: + """结束过滤，输出普通缓冲或清理被压住的显式思考段。""" + if not self._buffer: + return "" + if self._mode == "suppress" or _starts_with_reasoning_marker(self._buffer): + visible = _strip_leading_reasoning_section(self._buffer) + else: + visible = self._buffer + self._buffer = "" + self._mode = "pass" + return visible + + +def _starts_with_reasoning_marker(text: str) -> bool: + """判断文本首个非空内容是否是显式思考段标记。""" + return REASONING_START_RE.match(text) is not None + + +def _could_be_reasoning_marker_prefix(text: str) -> bool: + """流式初始阶段判断当前缓冲是否可能是思考段标记的一部分。""" + candidate = _normalize_marker_prefix(text) + if not candidate: + return True + markers = ( + "thinking process", + "thought process", + "reasoning process", + "chain of thought", + "internal reasoning", + "inner monologue", + "思考过程", + "推理过程", + "内部思考", + ) + return len(candidate) < MAX_REASONING_PREFIX_HOLD and any(marker.startswith(candidate) for marker in markers) + + +def _normalize_marker_prefix(text: str) -> str: + """把流式开头清理成便于判断的 marker 前缀。""" + stripped = text.lstrip() + stripped = re.sub(r"^(?:[#>\-*]+\s*)+", "", stripped) + stripped = stripped.strip("*").strip() + return stripped.lower() + + +def _extract_after_final_answer_marker(text: str) -> str | None: + """如果存在最终回答标记，返回标记后的正文。""" + matches = list(FINAL_ANSWER_RE.finditer(text)) + if not matches: + return None + return text[matches[-1].end() :].strip() + + +def _strip_leading_reasoning_section(text: str) -> str: + """删除以显式思考标记开头的推理段，保留后续最终正文。""" + final_text = _extract_after_final_answer_marker(text) + if final_text is not None: + return final_text + lines = text.splitlines() + first = _first_non_empty_line_index(lines) + if first is None or not _starts_with_reasoning_marker(lines[first]): + return text.strip() + last_reasoning = first + for index in range(first, len(lines)): + if _looks_like_reasoning_line(lines[index]): + last_reasoning = index + return "\n".join(lines[last_reasoning + 1 :]).strip() + + +def _first_non_empty_line_index(lines: list[str]) -> int | None: + """返回首个非空行下标。""" + for index, line in enumerate(lines): + if line.strip(): + return index + return None + + +def _looks_like_reasoning_line(line: str) -> bool: + """识别常见显式思考过程行。""" + stripped = line.strip() + if not stripped: + return False + if _starts_with_reasoning_marker(stripped): + return True + if REASONING_LINE_RE.search(stripped): + return True + return bool(re.match(r"^\s*\d+\.\s*\*\*[^*]+(?:request|response|answer|constraints|过程|回答)[^*]*\*\*", stripped, flags=re.IGNORECASE)) def _longest_suffix_prefix(text: str, targets: list[str]) -> int: diff --git a/tests/test_llm_text_filter.py b/tests/test_llm_text_filter.py index 3385958..81008ac 100644 --- a/tests/test_llm_text_filter.py +++ b/tests/test_llm_text_filter.py @@ -27,3 +27,57 @@ def test_filter_thinking_chunks_drops_unclosed_think_tail(): visible = list(filter_thinking_chunks(chunks)) assert "".join(visible) == "回答" + + +def test_strip_thinking_text_removes_explicit_thinking_process_without_tags(): + text = """Thinking Process: +1. **Analyze the Request:** +* Input: JSON object containing context and user_text ("你是谁？"). +* Role: PAM Deployment Agent Interaction Assistant. +* Constraints: + * Do NOT automatically trigger deployment, rollback, upgrade, script execution, or MCP calls. + * Do NOT output secrets. +2. **Determine the Response:** +* The user is asking about my identity. +* I need to introduce myself briefly. +3. **Drafting the Response:** +* Greeting/Identity: 我是 PAM 部署 Agent 的交互助手。 +* Function: 我可以回答普通问题、解释命令和部署流程。 +4. **Refining the Response:** +* Keep it concise and friendly. + +我是 PAM 部署 Agent 的交互助手。 +我可以回答普通问题、解释当前 Agent 的命令和部署流程。 +""" + + visible = strip_thinking_text(text) + + assert "Thinking Process" not in visible + assert "Analyze the Request" not in visible + assert "Determine the Response" not in visible + assert "Drafting the Response" not in visible + assert "我是 PAM 部署 Agent 的交互助手。" in visible + assert visible.startswith("我是 PAM 部署 Agent") + + +def test_strip_thinking_text_keeps_content_after_final_answer_marker(): + text = """Reasoning Process: +I should not expose this. +Final Answer: 可以，我只展示最终回答。 +""" + + assert strip_thinking_text(text) == "可以，我只展示最终回答。" + + +def test_filter_thinking_chunks_suppresses_explicit_reasoning_until_finish(): + chunks = [ + "Think", + "ing Process:\n", + "I should hide this reasoning.\n", + "Final Answer: ", + "这是最终回答。", + ] + + visible = list(filter_thinking_chunks(chunks)) + + assert "".join(visible) == "这是最终回答。"