LQAgentPlatform/utils_test/Prompt_Test/semantic_test_prompt.py

"""
测试提示词脚本
使用requests库测试语法审查API
支持本地模型和在线模型两种调用方式
"""
import requests
import json
from openai import OpenAI


class PromptTester:
    """提示词测试类"""
    
    def __init__(self, api_url, api_key=None, is_local=True):
        """
        初始化测试器
        
        Args:
            api_url: API地址
            api_key: API密钥（在线模型需要）
            is_local: 是否使用本地模型（True: 本地模型使用requests, False: 在线模型使用OpenAI SDK）
        """
        self.api_url = api_url
        self.api_key = api_key
        self.is_local = is_local
        
        # 如果是在线模型，初始化OpenAI客户端
        if not is_local and api_key:
            self.client = OpenAI(
                base_url=api_url.replace('/chat/completions', ''),  # 移除末尾的路径
                api_key=api_key
            )
        else:
            self.client = None
        self.system_prompt = """
# 角色定位 (Role)
你是建筑施工方案语义逻辑审查专家，专注于识别文本中的逻辑矛盾、语义冲突和表述混乱问题。
你的职责是发现真实存在的逻辑错误，而非挑剔文字表达或提出主观性建议。

# 审查范围 (Scope)
仅审查以下三类明确的语义逻辑问题：

1. **逻辑矛盾** - 同一文档内出现相互冲突的陈述
   示例：前文说"采用A方法"，后文说"不采用A方法"
   
2. **因果关系错误** - 原因与结果之间不存在合理的逻辑关联
   示例："因为天气晴朗，所以混凝土强度不足"（因果无关）
   
3. **条件与结论不匹配** - 给定条件无法推导出所述结论
   示例："当温度低于5℃时，可正常施工"（违背常识）
# 工作流程 (Workflow)

步骤1：通读全文，理解整体语境和专业背景
步骤2：识别关键陈述、条件判断、因果关系
步骤3：检查是否存在上述三类明确的逻辑问题
步骤4：对于疑似问题，进行二次验证：
       - 是否确实存在逻辑冲突？
       - 是否可能是专业术语的正常表达？
       - 是否可能是上下文理解不足导致的误判？
       - 如果你的上文中的内容是另外一个板块的规则，不可用于下一个板块的逻辑判断。
步骤5：仅输出确认无误的问题
# 严格规则 (Strict Rules)

禁止行为：
1. 禁止对正确内容提出修改建议
2. 禁止对专业术语的标准表达提出质疑
3. 禁止对文字风格、表达习惯提出意见
4. 禁止对表格格式、制表符进行检查
5. 禁止对同一问题重复输出
6. 禁止在没有明确逻辑错误时输出问题
7. 禁止将"表达不够完美"当作"逻辑错误"
8. 禁止将"可以优化"当作"必须修改"
9. 禁止在无上下文的情况猜测规则，禁止臆想规则。你的无上下文的规则不应该作为评判的依据。

必须遵守：
1. 必须基于客观事实和逻辑规则判断
2. 必须确保问题的真实性和严重性
3. 必须结合建筑施工专业知识理解内容
4. 必须区分"逻辑错误"与"表达习惯"
5. 必须在不确定时选择"无明显问题"
6. 必须保持高标准：宁缺毋滥

# 风险等级判定标准 (Risk Level)

- **高风险**：存在明显逻辑矛盾，可能导致施工错误或安全隐患
  示例：安全措施前后矛盾、施工顺序逻辑错误
  
- **中风险**：存在因果关系错误，可能导致理解偏差
  示例：原因分析不合理、条件判断不准确
  
- **低风险**：存在轻微的表述不一致，不影响实质理解
  示例：用词前后略有差异但不影响理解

# 判断原则 (Principles)

1. **疑罪从无原则**：当无法确定是否为错误时，判定为无问题
2. **专业优先原则**：尊重建筑施工行业的专业术语和表达习惯
3. **语境理解原则**：必须在完整语境中理解句子含义
4. **实质审查原则**：关注实质性逻辑错误，忽略形式问题
5. **严格证据原则**：必须有明确证据证明存在逻辑错误
6. **上下章节隔离原则**: 上下不同章节之间的信息不应该一同参考，即使有冲突也不应该一同讨论，章节条例应当独立
审查参考:
{review_references}
"""
        self.user_prompt_template = """
# 审查任务
请对以下施工方案内容进行语义逻辑审查，严格按照系统提示词中的三类问题范围进行检查。

## 待审查内容：
{review_content}


## 审查要求：
1. 仅识别明确的逻辑矛盾、因果错误、条件结论不匹配问题
2. 必须确保问题的真实性，不得对正确内容提出修改
3. 不确定是否为问题时，必须输出"无明显问题"
4. 必须结合建筑施工专业知识和完整语境判断

## 输出格式：

**情况1：未发现明确的语义逻辑问题**
直接输出：“无明显问题”，不需要解释。

**情况2：发现明确的语义逻辑问题**
严格按照以下JSON格式输出（仅输出确认无误的问题）：
```json
{{
  "issue_point": "[问题类型]具体问题描述（问题类型必须是：逻辑矛盾/因果关系错误/条件结论不匹配之一）",
  "location": "问题所在的原始条款内容及位置（如：三、施工方法 (页码: 12)），包含必要的上下文",
  "suggestion": "基于逻辑规则的具体修改建议（必须是纠正逻辑错误，而非优化表达）",
  "reason": "详细说明为何这是一个逻辑错误，包括：1)矛盾点在哪里 2)为何不符合逻辑 3)可能产生的后果",
  "risk_level": "高风险/中风险/低风险（严格按照系统提示词中的标准判定）"
}}
```

## 特别提醒：
- 如果内容表达虽不完美但逻辑正确，输出"无明显问题"
- 如果是专业术语的标准表达，输出"无明显问题"  
- 如果只是表达习惯差异，输出"无明显问题"
- 保持高标准：宁可漏报，不可误报
- 上下章节隔离: 上下不同章节之间的信息不应该一同参考，即使有冲突也不应该一同讨论，章节条例应当独立
"""

    def test_prompt(self, review_content, model="Qwen3-8B", temperature=0.3, max_tokens=2000):
        """
        测试提示词
        
        Args:
            review_content: 待审查的文本内容
            model: 模型名称
            temperature: 温度参数
            max_tokens: 最大token数
            
        Returns:
            API响应结果
        """
        # 填充用户提示词
        user_prompt = self.user_prompt_template.format(review_content=review_content)
        
        # 打印请求信息
        print("=" * 80)
        print(f"调用方式: {'本地模型 (requests)' if self.is_local else '在线模型 (OpenAI SDK)'}")
        print("发送请求到:", self.api_url)
        print("=" * 80)
        print("System Prompt:")
        print("-" * 80)
        print(self.system_prompt)
        print("=" * 80)
        print("User Prompt:")
        print("-" * 80)
        print(user_prompt)
        print("=" * 80)
        
        # 根据is_local选择不同的调用方式
        if self.is_local:
            # 本地模型：使用requests
            return self._call_with_requests(user_prompt, model, temperature, max_tokens)
        else:
            # 在线模型：使用OpenAI SDK
            return self._call_with_openai(user_prompt, model, temperature, max_tokens)
    
    def _call_with_requests(self, user_prompt, model, temperature, max_tokens):
        """使用requests库调用本地模型"""
        # 构建请求数据
        payload = {
            "model": model,
            "messages": [
                {
                    "role": "system",
                    "content": self.system_prompt
                },
                {
                    "role": "user",
                    "content": user_prompt
                }
            ],
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        try:
            # 构建请求头
            headers = {"Content-Type": "application/json"}
            if self.api_key:
                headers["Authorization"] = f"Bearer {self.api_key}"
            
            # 发送POST请求
            response = requests.post(
                self.api_url,
                headers=headers,
                json=payload,
                timeout=60
            )
            
            # 检查响应状态
            response.raise_for_status()
            
            # 解析响应
            result = response.json()
            
            print("响应状态码:", response.status_code)
            print("=" * 80)
            print("API响应结果:")
            print("-" * 80)
            print(json.dumps(result, ensure_ascii=False, indent=2))
            print("=" * 80)
            
            # 提取AI回复内容
            if "choices" in result and len(result["choices"]) > 0:
                ai_response = result["choices"][0].get("message", {}).get("content", "")
                print("AI回复内容:")
                print("-" * 80)
                print(ai_response)
                print("=" * 80)
            
            return result
            
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            return None
        except json.JSONDecodeError as e:
            print(f"JSON解析失败: {e}")
            print("原始响应:", response.text)
            return None
    
    def _call_with_openai(self, user_prompt, model, temperature, max_tokens):
        """使用OpenAI SDK调用在线模型"""
        if not self.client:
            print("错误: OpenAI客户端未初始化")
            return None
        
        try:
            # 使用OpenAI SDK调用
            extra_body = {
                # enable thinking, set to False to disable test
                "enable_thinking": False,
                # use thinking_budget to contorl num of tokens used for thinking
                # "thinking_budget": 4096
            }
            response = self.client.chat.completions.create(
                model=model,
                messages=[
                    {
                        "role": "system",
                        "content": self.system_prompt
                    },
                    {
                        "role": "user",
                        "content": user_prompt
                    }
                ],
                temperature=temperature,
                max_tokens=max_tokens,
                extra_body=extra_body
            )
            
            # 转换为字典格式
            result = {
                "id": response.id,
                "object": response.object,
                "created": response.created,
                "model": response.model,
                "choices": [
                    {
                        "index": choice.index,
                        "message": {
                            "role": choice.message.role,
                            "content": choice.message.content
                        },
                        "finish_reason": choice.finish_reason
                    }
                    for choice in response.choices
                ],
                "usage": {
                    "prompt_tokens": response.usage.prompt_tokens,
                    "completion_tokens": response.usage.completion_tokens,
                    "total_tokens": response.usage.total_tokens
                }
            }
            
            print("=" * 80)
            print("API响应结果:")
            print("-" * 80)
            print(json.dumps(result, ensure_ascii=False, indent=2))
            print("=" * 80)
            
            # 提取AI回复内容
            if result["choices"] and len(result["choices"]) > 0:
                ai_response = result["choices"][0]["message"]["content"]
                print("AI回复内容:")
                print("-" * 80)
                print(ai_response)
                print("=" * 80)
            
            return result
            
        except Exception as e:
            print(f"OpenAI SDK调用失败: {e}")
            return None


def main():
    """主函数"""
    
    # ==================== 配置区域 ====================
    # 是否使用本地模型（True: 本地模型, False: 在线模型）
    is_local = True
    
    # 本地模型配置
    local_config = {
        "api_url": "http://192.168.91.253:8003/v1/chat/completions",
        "api_key": "sk-123456",
        "model": "qwen3-30b",
        "temperature": 0.7,
        "max_tokens": 2000
    }
    
    # 在线模型配置（ModelScope）
    online_config = {
        "api_url": "https://api-inference.modelscope.cn/v1",
        "api_key": "ms-c0349a0a-8f15-466b-96be-4f96d001d8f2",  # ModelScope Token
        "model": "Qwen/Qwen3-14B",
        "temperature": 0.7,
        "max_tokens": 2000
    }
    # ==================== 配置区域结束 ====================
    
    # 根据is_local选择配置
    config = local_config if is_local else online_config
    
    print("=" * 80)
    print(f"当前使用: {'本地模型' if is_local else '在线模型'}")
    print(f"API地址: {config['api_url']}")
    print(f"模型名称: {config['model']}")
    print("=" * 80)
    
    # 创建测试器
    tester = PromptTester(config["api_url"], config["api_key"], is_local)
    
    custom_content = """
1、验收程序
（1）建立完整的质量保证体系，按施工合同约定的质量要求与质量管理程序进行作业，保证施工质量。
（2）每道工序开工前施工单位进行施工技术交底，重要工序或分部工程其内容须书面报送监理工程师审查。
（3）工序施工完毕，实行各道工序的操作人员“自检”、“互检”和专职质量管理人员“专检”相结合的“三检”程序，并签署有完整的检验记录。不合格，则自行返工。
（4）在“自检”、“互检”和“专检”合格基础上，备齐自检资料和报验申请表，提前申请验收。
（5）工程监理采用巡视、旁站、平行检查方式，及平常的工地例行检查，不能作为对施工方的施工质量的检查验收。
（6）施工工序属于隐蔽工程，需提前通知监理单位验收，未经监理工程师批准，任何工程工序均不能擅自隐蔽。隐蔽验收须按施工工序步骤逐步进行，并按工序步骤记录在验收表上，验收合格后才准许隐蔽。
（7）监理方接到施工方验收申请，积极组织监理方人员检查，对工程中存在的质量问题和安全隐患问题以书面的形式，要求施工单位进行整改回复、检查、再申请验收。
（8）监理方按设计和施工验收规范检查，确认无质量、安全问题，组织业主、施工单位联合检验。
（9）为确保工程质量，为设计和施工验收提供可靠依据，施工过程中各项试验，应严格按国家相关标准的规定进行见证抽样送检。未经监理见证送检的材料，一律不准用于工程实体。
图7-1 验收程序
"""
    tester.test_prompt(
        custom_content,
        model=config["model"],
        temperature=config["temperature"],
        max_tokens=config["max_tokens"]
    )


if __name__ == "__main__":
    main()