buggy_llm_server.py

"""
Buggy LLM Server - 故意返回有错误的代码用于测试
"""

from flask import Flask, request, jsonify
import json
import time
from typing import Dict, Any, List

app = Flask(__name__)

# 错误计数器 - 用于控制在第几次调用时返回正确代码
call_counts = {
    "coder": 0,
    "debugger": 0,
    "checker": 0
}


def generate_buggy_coder_response(messages: List[Dict[str, str]]) -> Dict[str, Any]:
    """生成有错误的代码"""
    call_counts["coder"] += 1
    
    user_content = ""
    for msg in reversed(messages):
        if msg.get("role") == "user":
            user_content = msg.get("content", "")
            break
    
    # 第一次调用：返回有语法错误的代码
    if "divide" in user_content.lower():
        code_data = {
            "prefix": "A function to divide two numbers",
            "imports": "",
            "code": """def divide(a, b):
    # 故意的语法错误 - 缺少引号
    if b == 0:
        raise ValueError(Cannot divide by zero)
    return a / b"""
        }
    elif "square root" in user_content.lower():
        code_data = {
            "prefix": "A function to calculate square root",
            "imports": "import math",
            "code": """def sqrt_safe(n):
    # 故意的逻辑错误 - 没有检查负数
    return math.sqrt(n)"""
        }
    else:
        # 默认：返回有缩进错误的代码
        code_data = {
            "prefix": "A simple function",
            "imports": "",
            "code": """def my_function():
print('Hello')  # 缩进错误
    return True"""
        }
    
    return {
        "choices": [{
            "message": {
                "content": json.dumps(code_data),
                "role": "assistant"
            },
            "finish_reason": "stop",
            "index": 0
        }]
    }


def generate_buggy_debugger_response(messages: List[Dict[str, str]], iteration: int) -> Dict[str, Any]:
    """生成调试后的代码 - 可能仍有错误"""
    call_counts["debugger"] += 1
    
    user_content = ""
    for msg in reversed(messages):
        if msg.get("role") == "user":
            user_content = msg.get("content", "")
            break
    
    # 根据迭代次数返回不同质量的代码
    if call_counts["debugger"] == 1:
        # 第一次：修复了语法错误，但有新的逻辑错误
        code_data = {
            "prefix": "Fixed syntax errors",
            "imports": "",
            "code": """def divide(a, b):
    if b = 0:  # 还有错误：应该用 == 而不是 =
        raise ValueError("Cannot divide by zero")
    return a / b"""
        }
    elif call_counts["debugger"] == 2:
        # 第二次：修复了比较运算符，但返回类型错误
        code_data = {
            "prefix": "Fixed comparison operator",
            "imports": "",
            "code": """def divide(a, b):
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return str(a / b)  # 错误：返回字符串而不是数字"""
        }
    else:
        # 第三次及以后：返回正确的代码
        code_data = {
            "prefix": "Fully corrected version",
            "imports": "",
            "code": """def divide(a, b):
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b"""
        }
    
    return {
        "choices": [{
            "message": {
                "content": json.dumps(code_data),
                "role": "assistant"
            },
            "finish_reason": "stop",
            "index": 0
        }]
    }


def generate_buggy_checker_response(messages: List[Dict[str, str]], iteration: int) -> Dict[str, Any]:
    """生成 checker 响应 - 前几次返回失败"""
    call_counts["checker"] += 1
    
    if call_counts["checker"] <= 2:
        # 前两次：验证失败
        checker_data = {
            "is_valid": False,
            "message": f"Code has issues (check iteration {call_counts['checker']})",
            "errors": [
                "Logic does not fully match requirements",
                "Missing edge case handling"
            ],
            "requirements_met": [],
            "requirements_missing": ["Proper error handling", "Complete implementation"]
        }
    else:
        # 第三次：验证通过
        checker_data = {
            "is_valid": True,
            "message": "Code now meets all requirements",
            "errors": [],
            "requirements_met": ["Division functionality", "Zero division handling"],
            "requirements_missing": []
        }
    
    return {
        "choices": [{
            "message": {
                "content": json.dumps(checker_data),
                "role": "assistant"
            },
            "finish_reason": "stop",
            "index": 0
        }]
    }


@app.route('/v1/chat/completions', methods=['POST'])
def chat_completions():
    """处理聊天补全请求"""
    try:
        data = request.get_json()
        messages = data.get('messages', [])
        model = data.get('model', 'default')
        
        time.sleep(0.1)  # 模拟处理延迟
        
        # 根据模型类型生成不同的响应
        if 'coder' in model.lower() or 'code' in model.lower():
            response = generate_buggy_coder_response(messages)
        elif 'debug' in model.lower() or 'fix' in model.lower():
            response = generate_buggy_debugger_response(messages, call_counts["debugger"])
        elif 'check' in model.lower() or 'valid' in model.lower():
            response = generate_buggy_checker_response(messages, call_counts["checker"])
        else:
            response = generate_buggy_coder_response(messages)
        
        response.update({
            "model": model,
            "object": "chat.completion",
            "created": int(time.time()),
            "usage": {
                "prompt_tokens": len(str(messages)),
                "completion_tokens": len(str(response)),
                "total_tokens": len(str(messages)) + len(str(response))
            }
        })
        
        return jsonify(response)
    
    except Exception as e:
        return jsonify({
            "error": {
                "message": str(e),
                "type": "internal_error",
                "code": "internal_error"
            }
        }), 500


@app.route('/health', methods=['GET'])
def health_check():
    """健康检查"""
    return jsonify({
        "status": "healthy",
        "service": "buggy-llm-server",
        "timestamp": int(time.time()),
        "call_counts": call_counts
    })


@app.route('/reset', methods=['POST'])
def reset_counters():
    """重置调用计数器"""
    global call_counts
    call_counts = {"coder": 0, "debugger": 0, "checker": 0}
    return jsonify({"status": "reset", "call_counts": call_counts})


if __name__ == '__main__':
    print("="*60)
    print("🐛 启动 Buggy LLM Server (用于测试错误处理)")
    print("="*60)
    print("此服务器会故意返回有错误的代码来测试系统的错误处理能力")
    print("\n端点:")
    print("  - POST /v1/chat/completions")
    print("  - GET  /health")
    print("  - POST /reset (重置调用计数)")
    print("\n特性:")
    print("  - Coder: 返回有语法错误的代码")
    print("  - Debugger: 前2次返回仍有错误的代码，第3次返回正确代码")
    print("  - Checker: 前2次返回验证失败，第3次返回通过")
    print("="*60)
    print("\n🔧 按 Ctrl+C 停止服务\n")
    
    app.run(host='0.0.0.0', port=8000, debug=False, threaded=True)