LQAgentPlatform/utils_test/RAG_Test/debug_messages.py

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

"""
详细调试传递给模型的消息内容
"""

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

from foundation.ai.rag.retrieval.query_rewrite import prompt_loader


def debug_messages():
    """
    调试实际传递给模型的消息
    """
    print("="*60)
    print("调试传递给模型的消息内容")
    print("="*60)

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

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

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

        # 检查模板的输入变量
        print(f"[DEBUG] 模板输入变量: {task_prompt.input_variables}")

        # 尝试不同的格式化方式
        print("\n=== 方法1: format_messages() (不传参数) ===")
        try:
            messages1 = task_prompt.format_messages()
            print(f"[DEBUG] 用户消息: {messages1[1].content}")
        except Exception as e:
            print(f"[ERROR] 方法1失败: {e}")

        print("\n=== 方法2: format_messages(review_content=...) ===")
        try:
            messages2 = task_prompt.format_messages(review_content=review_content)
            print(f"[DEBUG] 用户消息: {messages2[1].content}")
            print(f"[OK] 方法2成功!")
        except Exception as e:
            print(f"[ERROR] 方法2失败: {e}")

        print("\n=== 方法3: 手动构造消息 ===")
        try:
            # 获取提示词配置
            cache_key = "query_extract_query_extract"
            prompt_config = prompt_loader._cache[cache_key]

            system_prompt = prompt_config['system_prompt']
            user_template = prompt_config['user_prompt_template']

            # 手动替换变量
            user_prompt = user_template.replace("{{review_content}}", review_content)

            print(f"[DEBUG] 系统提示词: {system_prompt[:100]}...")
            print(f"[DEBUG] 用户提示词: {user_prompt}")
            print(f"[OK] 手动替换成功!")
        except Exception as e:
            print(f"[ERROR] 方法3失败: {e}")

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


def main():
    debug_messages()


if __name__ == "__main__":
    main()