LQAgentPlatform/utils_test/RAG_Test/debug_query_extract.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
调试QueryRewriteManager.query_extract方法
"""

import sys
import os
import time
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from foundation.ai.rag.retrieval.query_rewrite import QueryRewriteManager
from foundation.observability.logger.loggering import server_logger as logger


def debug_query_extract():
    """
    调试query_extract方法
    """
    print("="*60)
    print("调试QueryRewriteManager.query_extract方法")
    print("="*60)

    # 测试数据
    review_content = "深度大于3m的基坑开挖、有地下水侵扰的基坑清底封底，每个工作班至少巡查两遍。"
    print(f"原始输入内容: {review_content}")
    print(f"内容长度: {len(review_content)}")

    try:
        # 手动构建提示词模板进行调试
        from foundation.ai.rag.retrieval.query_rewrite import prompt_loader, generate_model_client
        import uuid
        import asyncio

        # 获取提示词模板
        task_prompt = prompt_loader.get_prompt_template(
            reviewer_type="query_extract",
            review_content=review_content
        )

        print(f"\n[DEBUG] 提示词模板类型: {type(task_prompt)}")

        # 尝试格式化消息
        try:
            messages = task_prompt.format_messages()
            print(f"[DEBUG] 消息数量: {len(messages)}")
            print(f"[DEBUG] 系统消息: {messages[0].content[:200]}...")
            print(f"[DEBUG] 用户消息: {messages[1].content[:300]}...")

            # 检查用户消息是否包含正确的review_content
            if review_content in messages[1].content:
                print("[OK] review_content 正确传递到提示词")
            else:
                print("[ERROR] review_content 未正确传递到提示词")
                print(f"[DEBUG] 期望内容: {review_content}")
                print(f"[DEBUG] 用户消息实际内容: {messages[1].content}")

        except Exception as e:
            print(f"[ERROR] 格式化消息失败: {e}")
            return

        # 构建任务提示信息
        task_prompt_info = {
            "task_prompt": task_prompt,
            "task_name": "query_extract"
        }

        # 生成trace_id
        trace_id = str(uuid.uuid4())
        print(f"[DEBUG] Trace ID: {trace_id}")

        # 调用模型生成接口
        print("[DEBUG] 开始调用模型...")
        model_response = asyncio.run(generate_model_client.get_model_generate_invoke(
            trace_id=trace_id,
            task_prompt_info=task_prompt_info
        ))

        print(f"[DEBUG] 模型响应: {model_response}")

        # 使用原始方法进行对比测试
        print("\n" + "="*40)
        print("使用原始QueryRewriteManager方法测试")
        print("="*40)

        query_rewrite_manager = QueryRewriteManager()
        start_time = time.time()
        result = query_rewrite_manager.query_extract(review_content)
        end_time = time.time()
        elapsed_time = end_time - start_time

        print(f"[OK] 原始方法提取完成，耗时: {elapsed_time:.2f}秒")
        print(f"[OK] 原始方法返回结果: {result}")

    except Exception as e:
        print(f"[ERROR] 调试失败: {str(e)}")
        import traceback
        traceback.print_exc()


def main():
    """
    主测试函数
    """
    print("开始调试 QueryRewriteManager.query_extract 方法")

    debug_query_extract()

    print("\n调试完成")


if __name__ == "__main__":
    main()