|
|
#!/usr/bin/env python3
|
|
|
"""
|
|
|
简化的DeepSeek v3.1代码规约化测试脚本
|
|
|
|
|
|
直接使用aiohttp调用DeepSeek API，无需复杂的依赖
|
|
|
"""
|
|
|
|
|
|
import os
|
|
|
import asyncio
|
|
|
import aiohttp
|
|
|
import json
|
|
|
import time
|
|
|
|
|
|
|
|
|
class SimpleDeepSeekTest:
|
|
|
"""简化的DeepSeek测试类"""
|
|
|
|
|
|
def __init__(self, api_key=None, base_url=None):
|
|
|
self.api_key = api_key or os.getenv('SILICONFLOW_API_KEY')
|
|
|
self.base_url = base_url or 'https://api.siliconflow.cn/v1'
|
|
|
self.model = 'deepseek-ai/DeepSeek-V3.1'
|
|
|
|
|
|
if not self.api_key:
|
|
|
raise ValueError("需要设置 SILICONFLOW_API_KEY 环境变量")
|
|
|
|
|
|
async def generate_specification(self, function_info, verification_goals):
|
|
|
"""生成形式化规约"""
|
|
|
# 构建提示词
|
|
|
prompt = self._build_prompt(function_info, verification_goals)
|
|
|
|
|
|
# 准备请求数据
|
|
|
messages = [
|
|
|
{
|
|
|
"role": "system",
|
|
|
"content": "你是一个专业的形式化验证专家，专门为C/C++代码生成CBMC格式的形式化规约。请生成准确、完整的CBMC规约，只包含CBMC条款，不要包含其他解释。"
|
|
|
},
|
|
|
{
|
|
|
"role": "user",
|
|
|
"content": prompt
|
|
|
}
|
|
|
]
|
|
|
|
|
|
data = {
|
|
|
"model": self.model,
|
|
|
"messages": messages,
|
|
|
"temperature": 0.3,
|
|
|
"max_tokens": 1024,
|
|
|
"stream": False
|
|
|
}
|
|
|
|
|
|
headers = {
|
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
|
"Content-Type": "application/json"
|
|
|
}
|
|
|
|
|
|
start_time = time.time()
|
|
|
|
|
|
try:
|
|
|
async with aiohttp.ClientSession() as session:
|
|
|
async with session.post(
|
|
|
f"{self.base_url}/chat/completions",
|
|
|
headers=headers,
|
|
|
json=data,
|
|
|
timeout=aiohttp.ClientTimeout(total=60)
|
|
|
) as response:
|
|
|
|
|
|
if response.status == 200:
|
|
|
result = await response.json()
|
|
|
content = result['choices'][0]['message']['content']
|
|
|
tokens_used = result.get('usage', {}).get('total_tokens', 0)
|
|
|
|
|
|
# 后处理：提取CBMC规约
|
|
|
specification = self._extract_cbmc_spec(content)
|
|
|
|
|
|
generation_time = time.time() - start_time
|
|
|
|
|
|
return {
|
|
|
'specification': specification,
|
|
|
'raw_content': content,
|
|
|
'generation_time': generation_time,
|
|
|
'tokens_used': tokens_used,
|
|
|
'success': True
|
|
|
}
|
|
|
else:
|
|
|
error_text = await response.text()
|
|
|
return {
|
|
|
'success': False,
|
|
|
'error': f"API错误: {response.status} - {error_text}"
|
|
|
}
|
|
|
|
|
|
except Exception as e:
|
|
|
return {
|
|
|
'success': False,
|
|
|
'error': f"请求失败: {str(e)}"
|
|
|
}
|
|
|
|
|
|
def _build_prompt(self, function_info, verification_goals):
|
|
|
"""构建提示词"""
|
|
|
func_name = function_info.get('name', 'unknown')
|
|
|
return_type = function_info.get('return_type', 'void')
|
|
|
parameters = function_info.get('parameters', [])
|
|
|
source_code = function_info.get('source_code', '')
|
|
|
|
|
|
# 构建函数签名
|
|
|
param_str = ', '.join([f"{p['type']} {p['name']}" for p in parameters])
|
|
|
function_signature = f"{return_type} {func_name}({param_str})"
|
|
|
|
|
|
# 构建验证目标描述
|
|
|
goal_descriptions = {
|
|
|
'functional_correctness': '功能正确性验证',
|
|
|
'memory_safety': '内存安全性验证',
|
|
|
'pointer_validity': '指针有效性验证',
|
|
|
'integer_overflow': '整数溢出检查',
|
|
|
'array_bounds': '数组边界检查',
|
|
|
'error_handling': '错误处理验证'
|
|
|
}
|
|
|
|
|
|
goals_text = []
|
|
|
for goal in verification_goals:
|
|
|
description = goal_descriptions.get(goal, goal)
|
|
|
goals_text.append(f"- {description}")
|
|
|
|
|
|
prompt = f"""
|
|
|
请为以下C函数生成CBMC形式化验证规约：
|
|
|
|
|
|
函数签名：{function_signature}
|
|
|
|
|
|
源代码：
|
|
|
```c
|
|
|
{source_code}
|
|
|
```
|
|
|
|
|
|
验证目标：
|
|
|
{chr(10).join(goals_text)}
|
|
|
|
|
|
请生成完整的CBMC规约，包含：
|
|
|
1. \\requires 子句：前置条件
|
|
|
2. \\ensures 子句：后置条件
|
|
|
3. \\assigns 子句：赋值说明（如果需要）
|
|
|
|
|
|
只输出CBMC规约，不要包含其他解释。
|
|
|
"""
|
|
|
return prompt
|
|
|
|
|
|
def _extract_cbmc_spec(self, content):
|
|
|
"""从生成内容中提取CBMC规约"""
|
|
|
lines = content.split('\n')
|
|
|
cbmc_lines = []
|
|
|
|
|
|
for line in lines:
|
|
|
line = line.strip()
|
|
|
# 保留CBMC条款
|
|
|
if (line.startswith(r'\requires') or
|
|
|
line.startswith(r'\ensures') or
|
|
|
line.startswith(r'\assigns') or
|
|
|
line.startswith(r'\valid') or
|
|
|
line.startswith(r'\forall') or
|
|
|
line.startswith(r'\exists')):
|
|
|
cbmc_lines.append(line)
|
|
|
|
|
|
return '\n'.join(cbmc_lines) if cbmc_lines else content.strip()
|
|
|
|
|
|
async def health_check(self):
|
|
|
"""健康检查"""
|
|
|
try:
|
|
|
async with aiohttp.ClientSession() as session:
|
|
|
async with session.post(
|
|
|
f"{self.base_url}/chat/completions",
|
|
|
headers={
|
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
|
"Content-Type": "application/json"
|
|
|
},
|
|
|
json={
|
|
|
"model": self.model,
|
|
|
"messages": [{"role": "user", "content": "test"}],
|
|
|
"max_tokens": 5
|
|
|
},
|
|
|
timeout=aiohttp.ClientTimeout(total=10)
|
|
|
) as response:
|
|
|
|
|
|
return {
|
|
|
'status': 'healthy' if response.status == 200 else 'unhealthy',
|
|
|
'response_time': 0.0, # 简化版本
|
|
|
'status_code': response.status
|
|
|
}
|
|
|
except Exception as e:
|
|
|
return {
|
|
|
'status': 'unhealthy',
|
|
|
'error': str(e)
|
|
|
}
|
|
|
|
|
|
|
|
|
def create_test_functions():
|
|
|
"""创建测试函数"""
|
|
|
return [
|
|
|
{
|
|
|
'name': 'add_numbers',
|
|
|
'return_type': 'int',
|
|
|
'parameters': [
|
|
|
{'name': 'a', 'type': 'int'},
|
|
|
{'name': 'b', 'type': 'int'}
|
|
|
],
|
|
|
'source_code': 'int add_numbers(int a, int b) { return a + b; }',
|
|
|
'verification_goals': ['functional_correctness']
|
|
|
},
|
|
|
{
|
|
|
'name': 'safe_divide',
|
|
|
'return_type': 'bool',
|
|
|
'parameters': [
|
|
|
{'name': 'a', 'type': 'int'},
|
|
|
{'name': 'b', 'type': 'int'},
|
|
|
{'name': 'result', 'type': 'int*'}
|
|
|
],
|
|
|
'source_code': '''
|
|
|
bool safe_divide(int a, int b, int *result) {
|
|
|
if (b == 0 || result == NULL) {
|
|
|
return false;
|
|
|
}
|
|
|
*result = a / b;
|
|
|
return true;
|
|
|
}''',
|
|
|
'verification_goals': ['functional_correctness', 'memory_safety', 'error_handling']
|
|
|
}
|
|
|
]
|
|
|
|
|
|
|
|
|
async def main():
|
|
|
"""主测试函数"""
|
|
|
print("🚀 DeepSeek v3.1 代码规约化简化测试")
|
|
|
print("=" * 50)
|
|
|
|
|
|
# 检查API密钥
|
|
|
if not os.getenv('SILICONFLOW_API_KEY'):
|
|
|
print("❌ 错误: 未设置 SILICONFLOW_API_KEY 环境变量")
|
|
|
print("请设置: export SILICONFLOW_API_KEY=your_api_key")
|
|
|
return
|
|
|
|
|
|
try:
|
|
|
# 创建测试实例
|
|
|
tester = SimpleDeepSeekTest()
|
|
|
print(f"📡 连接模型: {tester.model}")
|
|
|
|
|
|
# 健康检查
|
|
|
print("\n🔍 检查API状态...")
|
|
|
health = await tester.health_check()
|
|
|
if health['status'] == 'healthy':
|
|
|
print("✅ API健康")
|
|
|
else:
|
|
|
print(f"❌ API不健康: {health.get('error', '未知错误')}")
|
|
|
return
|
|
|
|
|
|
# 测试函数
|
|
|
test_functions = create_test_functions()
|
|
|
|
|
|
for i, func in enumerate(test_functions, 1):
|
|
|
print(f"\n{'='*20} 测试 {i} {'='*20}")
|
|
|
print(f"📝 函数: {func['name']}")
|
|
|
print(f"🎯 目标: {', '.join(func['verification_goals'])}")
|
|
|
|
|
|
print("\n⏳ 正在生成规约...")
|
|
|
result = await tester.generate_specification(func, func['verification_goals'])
|
|
|
|
|
|
if result['success']:
|
|
|
print(f"✅ 生成成功!")
|
|
|
print(f"⏱️ 时间: {result['generation_time']:.2f}s")
|
|
|
print(f"🔤 Token: {result['tokens_used']}")
|
|
|
|
|
|
print(f"\n📋 生成的CBMC规约:")
|
|
|
print("-" * 40)
|
|
|
print(result['specification'])
|
|
|
print("-" * 40)
|
|
|
|
|
|
print(f"\n📄 原始输出:")
|
|
|
print("-" * 40)
|
|
|
print(result['raw_content'])
|
|
|
print("-" * 40)
|
|
|
else:
|
|
|
print(f"❌ 生成失败: {result['error']}")
|
|
|
|
|
|
print(f"\n🎉 测试完成!")
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"❌ 测试失败: {str(e)}")
|
|
|
import traceback
|
|
|
traceback.print_exc()
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
asyncio.run(main()) |
