LLM Check & Refinement¶

# verify_agent.py
vmanager = StageManager(
    agent=agent,
    setting=setting,
    # ...
)

def init_stage(self, vstage: VerifyStage):
    """初始化阶段，创建 LLM 建议实例"""

    # 创建 Fail Refinement 实例
    if self.need_fail_llm_suggestion:
        vstage.llm_fail_suggestion = get_llm_check_instance(
            vstage=vstage,
            setting=self.setting['llm_suggestion']['check_fail_refinement'],
            agent_tools=self.tool_inspect_file  # 提供文件检查工具
        )

    # 创建 Pass Check 实例
    if self.need_pass_llm_suggestion:
        vstage.llm_pass_suggestion = get_llm_check_instance(
            vstage=vstage,
            setting=self.setting['llm_suggestion']['check_pass_refinement'],
            agent_tools=self.tool_inspect_file + [self.tool_approve_stage]  # 额外提供批准工具
        )

class VerifyStage:
    def __init__(self, ...):
        # LLM 建议相关属性
        self.need_fail_llm_suggestion: bool  # 是否需要失败建议
        self.need_pass_llm_suggestion: bool  # 是否需要通过验收
        self.continue_fail_count: int = 0    # 连续失败计数器（第 79 行）
        self._approved: bool = False          # LLM 批准状态（第 93 行）

        # LLM 建议实例
        self.llm_fail_suggestion = None
        self.llm_pass_suggestion = None

def gen_fail_suggestion(self, error_msg: str) -> Optional[str]:
    """生成失败建议"""
    vstage = self.get_current_stage()

    # 调用通用建议生成器
    return self.gen_llm_suggestion(
        vstage=vstage,
        llm_suggestion=vstage.llm_fail_suggestion,
        need_llm_suggestion=self.need_fail_llm_suggestion,
        prompts=(error_msg, error_msg)  # (任务提示, 建议提示)
    )

def gen_llm_suggestion(
    self,
    vstage: VerifyStage,
    llm_suggestion: BaseLLMSuggestion,
    need_llm_suggestion: bool,
    prompts: tuple
) -> Optional[str]:
    """通用 LLM 建议生成逻辑"""

    # 1. 检查是否启用
    if not need_llm_suggestion:
        return prompts[1]

    # 2. 阶段过滤：检查 bypass_stages
    if vstage.name in llm_suggestion.bypass_stages:
        return prompts[1]

    # 3. 阶段过滤：检查 target_stages
    if llm_suggestion.target_stages and vstage.name not in llm_suggestion.target_stages:
        return prompts[1]

    # 4. 阶段过滤：检查 default_apply_all_stages
    if not llm_suggestion.target_stages and not llm_suggestion.default_apply_all_stages:
        return prompts[1]

    # 5. 调用 LLM 建议方法
    return llm_suggestion.suggest(prompts[0])

def suggest(self, task_prompt: str) -> str:
    """生成失败优化建议"""

    # 1. 检查连续失败次数
    if self.vstage.continue_fail_count < self.min_fail_count:
        return self.suggestion_prompt  # 不足 min_fail_count 次，直接返回原始错误

    # 2. 切换阶段时清理对话历史
    if self.current_vstage_name != self.vstage.name:
        self._clear_history()
        self.current_vstage_name = self.vstage.name

    # 3. 构造任务信息
    task_info = self._build_task_info()
    test_info = self._extract_error_info()

    # 4. 构造完整提示
    full_prompt = f"{task_prompt}\n\n任务信息：\n{task_info}\n\n测试信息：\n{test_info}"

    # 5. 调用 LangChain Agent
    result = self._invoke_agent(full_prompt)

    # 6. 清理忽略标签并返回
    return self._remove_ignore_labels(result)

def _invoke_agent(self, prompt: str) -> str:
    """调用 LangChain Agent"""

    # 创建 Agent（带工具和中间件）
    agent = create_agent(
        model=self.model,
        tools=self.tools,  # ReadTextFile, ListDir, SearchText
        system_prompt=self.system_prompt
    )

    # 添加摘要中间件（防止对话过长）
    agent_with_middleware = agent | first_message_summary_middleware(
        trigger_tokens=self.summary_trigger_tokens,  # 32k
        keep_messages=self.summary_keep_messages     # 10
    )

    # 流式调用 Agent
    result = stream_call_langchain_agent(
        agent=agent_with_middleware,
        prompt=prompt,
        thread_id=self.vstage.name,  # 每个阶段独立的对话线程
        checkpointer=self.checkpointer
    )

    return result

def gen_pass_suggestion(self, ck_info: str) -> tuple:
    """生成通过验收建议"""
    vstage = self.get_current_stage()

    # 默认设置为未批准
    vstage.set_approved(False)

    # 调用通用建议生成器
    suggestion = self.gen_llm_suggestion(
        vstage=vstage,
        llm_suggestion=vstage.llm_pass_suggestion,
        need_llm_suggestion=self.need_pass_llm_suggestion,
        prompts=(ck_info, ck_info)
    )

    # 检查 LLM 是否批准
    if vstage.get_approved():
        return (True, suggestion)  # 批准，阶段完成
    else:
        return (False, f"LLM 未批准完成，建议：{suggestion}")  # 未批准，返回失败

def suggest(self, task_prompt: str) -> str:
    """生成通过验收建议"""

    # 1. 切换阶段时清理对话历史
    if self.current_vstage_name != self.vstage.name:
        self._clear_history()
        self.current_vstage_name = self.vstage.name

    # 2. 构造任务信息和测试信息
    task_info = self._build_task_info()
    test_info = self._extract_error_info()

    full_prompt = f"{task_prompt}\n\n任务信息：\n{task_info}\n\n测试信息：\n{test_info}"

    # 3. 调用 LangChain Agent（带 ApproveStagePass 工具）
    result = self._invoke_agent(full_prompt)

    # 4. 返回建议（approved 状态由工具调用设置）
    return self._remove_ignore_labels(result)

class LLMApprovalTool:
    """LLM 批准工具"""

    def __init__(self, vstage: VerifyStage):
        self.vstage = vstage

    def __call__(self, approved: bool) -> str:
        """设置批准状态"""
        self.vstage.set_approved(approved)
        if approved:
            return "已批准阶段完成"
        else:
            return "未批准，需要继续改进"

self.tool_approve_stage = StructuredTool.from_function(
    func=LLMApprovalTool(vstage),
    name="ApproveStagePass",
    description="批准或拒绝当前阶段的完成。参数 approved: true 表示批准，false 表示拒绝。"
)

# 切换阶段时清理历史
if self.current_vstage_name != self.vstage.name:
    self._clear_history()
    self.current_vstage_name = self.vstage.name

def first_message_summary_middleware(trigger_tokens: int, keep_messages: int):
    """对话摘要中间件

    当 token 数超过 trigger_tokens 时，保留首条消息和最近 keep_messages 条消息，
    删除中间的历史消息。
    """
    # 实现逻辑：
    # 1. 计算当前对话的 token 数
    # 2. 如果超过阈值，保留首条消息（系统提示）和最近 N 条消息
    # 3. 删除中间的历史消息

class VerifyStage:
    def do_check(self, timeout: int, is_complete: bool = False) -> tuple:
        """执行检查"""

        # 调用 Checker
        ck_pass, ck_info = self.checker.do_check(timeout, is_complete)

        # 更新连续失败计数器（第 300 行附近）
        if ck_pass:
            self.continue_fail_count = 0  # 成功时重置
        else:
            self.continue_fail_count += 1  # 失败时累加

        return (ck_pass, ck_info)

def _extract_error_info(self) -> str:
    """提取错误信息"""

    # 从 vstage.check_info 中提取 last_msg.error 字段
    check_info = self.vstage.check_info
    if check_info and hasattr(check_info, 'last_msg'):
        error = check_info.last_msg.error
        return f"错误类型：{error.type}\n错误信息：{error.message}\n堆栈跟踪：{error.traceback}"

    return "无详细错误信息"

def _remove_ignore_labels(self, text: str) -> str:
    """移除忽略标签"""
    for start_label, end_label in self.ignore_labels:
        # 使用正则表达式移除 start_label 和 end_label 之间的内容
        text = re.sub(f"{re.escape(start_label)}.*?{re.escape(end_label)}", "", text, flags=re.DOTALL)
    return text