real_llm_server.py

"""
Real LLM Server using Moonshot (Kimi) API
真实的 LLM 服务器，调用 Kimi 大模型 API
"""

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

app = Flask(__name__)

# Moonshot API 配置
MOONSHOT_API_KEY = "sk-4VF7qW5HsuXjgOwiCl4boalx8HzY5U2HBSF8mIFPD8Hz8SLM"
MOONSHOT_BASE_URL = "https://api.moonshot.cn/v1"
MOONSHOT_MODEL = "kimi-k2-turbo-preview"

# 不同角色的系统提示词
SYSTEM_PROMPTS = {
    "coder": """你是一个专业的Python代码生成助手。
你的任务是根据用户需求生成可执行的Python代码。

要求：
1. 生成的代码必须是完整的、可执行的Python代码
2. 包含必要的导入语句
3. 代码要简洁清晰，包含适当的注释
4. 必须以JSON格式返回，包含三个字段：
   - prefix: 代码的简短描述
   - imports: 所有需要的导入语句（每行一个导入）
   - code: 实际的代码实现

示例输出格式：
{
    "prefix": "A function to add two numbers",
    "imports": "",
    "code": "def add(a, b):\\n    return a + b"
}

请严格按照这个JSON格式返回，不要包含其他文字。""",

    "debugger": """你是一个专业的Python代码调试专家。
你的任务是分析代码错误并提供修复后的代码。

要求：
1. 分析提供的代码和错误信息
2. 修复所有的错误（语法错误、逻辑错误、运行时错误）
3. 保持代码的原始逻辑和功能
4. 返回修复后的完整代码

返回格式必须是JSON：
{
    "prefix": "Fixed version with error corrections",
    "imports": "导入语句",
    "code": "修复后的代码",
    "has_errors": false,
    "error_info": ""
}

或者如果仍有错误：
{
    "has_errors": true,
    "error_info": "详细的错误说明",
    "fixed_code": {
        "prefix": "...",
        "imports": "...",
        "code": "..."
    }
}""",

    "checker": """你是一个代码验证专家。
你的任务是验证生成的代码是否符合用户的原始需求。

要求：
1. 仔细比较代码实现和用户的原始需求
2. 检查代码逻辑是否正确
3. 确保代码没有偏离原始意图
4. 提供详细的验证结果

返回格式必须是JSON：
{
    "is_valid": true/false,
    "message": "验证结果的简要说明",
    "errors": ["错误列表"],
    "requirements_met": ["已满足的需求"],
    "requirements_missing": ["未满足的需求"],
    "suggestions": "改进建议（如果有）",
    "suggested_code": {
        "prefix": "...",
        "imports": "...", 
        "code": "..."
    }
}

如果代码完全符合要求，is_valid 应该为 true，errors 和 requirements_missing 应该为空列表。
如果需要修改，提供 suggested_code。"""
}


def call_moonshot_api(messages: List[Dict[str, str]], temperature: float = 0.3) -> Dict[str, Any]:
    """调用 Moonshot API"""
    try:
        headers = {
            "Authorization": f"Bearer {MOONSHOT_API_KEY}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": MOONSHOT_MODEL,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": 2000
        }
        
        response = requests.post(
            f"{MOONSHOT_BASE_URL}/chat/completions",
            headers=headers,
            json=payload,
            timeout=60
        )
        
        if response.status_code == 200:
            return response.json()
        else:
            error_msg = f"API Error {response.status_code}: {response.text}"
            print(f"❌ {error_msg}")
            return {
                "error": error_msg,
                "status_code": response.status_code
            }
    
    except Exception as e:
        error_msg = f"Exception calling Moonshot API: {str(e)}"
        print(f"❌ {error_msg}")
        return {
            "error": error_msg
        }


def generate_coder_response(messages: List[Dict[str, str]]) -> Dict[str, Any]:
    """生成 Coder 响应 - 调用真实 API"""
    print("\n🔨 Coder: 调用 Kimi API 生成代码...")
    
    # 构建完整的消息列表
    full_messages = [
        {"role": "system", "content": SYSTEM_PROMPTS["coder"]}
    ] + messages
    
    # 调用 API
    api_response = call_moonshot_api(full_messages, temperature=0.3)
    
    if "error" in api_response:
        # API 调用失败，返回错误
        return {
            "choices": [{
                "message": {
                    "content": json.dumps({
                        "prefix": "Error generating code",
                        "imports": "",
                        "code": f"# Error: {api_response['error']}"
                    }),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }
    
    # 提取 API 返回的内容
    try:
        content = api_response["choices"][0]["message"]["content"]
        print(f"✓ Kimi 返回: {content[:100]}...")
        
        # 尝试清理并解析 JSON
        # 移除可能的 markdown 代码块标记
        if "```json" in content:
            content = content.split("```json")[1].split("```")[0].strip()
        elif "```" in content:
            content = content.split("```")[1].split("```")[0].strip()
        
        # 尝试解析 JSON
        try:
            code_data = json.loads(content)
            # 验证必需的字段
            if "prefix" in code_data and "code" in code_data:
                print(f"✓ 成功解析代码")
            else:
                raise ValueError("Missing required fields")
        except:
            # JSON 解析失败，尝试从文本中提取
            print("⚠️  JSON 解析失败，使用文本提取")
            code_data = {
                "prefix": "Generated code from Kimi",
                "imports": "",
                "code": content
            }
        
        return {
            "choices": [{
                "message": {
                    "content": json.dumps(code_data),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }
    
    except Exception as e:
        print(f"❌ 处理响应失败: {e}")
        return {
            "choices": [{
                "message": {
                    "content": json.dumps({
                        "prefix": "Error processing response",
                        "imports": "",
                        "code": f"# Error: {str(e)}"
                    }),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }


def generate_debugger_response(messages: List[Dict[str, str]]) -> Dict[str, Any]:
    """生成 Debugger 响应 - 调用真实 API"""
    print("\n🐛 Debugger: 调用 Kimi API 调试代码...")
    
    # 构建完整的消息列表
    full_messages = [
        {"role": "system", "content": SYSTEM_PROMPTS["debugger"]}
    ] + messages
    
    # 调用 API
    api_response = call_moonshot_api(full_messages, temperature=0.2)
    
    if "error" in api_response:
        return {
            "choices": [{
                "message": {
                    "content": json.dumps({
                        "has_errors": True,
                        "error_info": api_response['error'],
                        "prefix": "Debug failed",
                        "imports": "",
                        "code": ""
                    }),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }
    
    try:
        content = api_response["choices"][0]["message"]["content"]
        print(f"✓ Kimi 返回: {content[:100]}...")
        
        # 清理并解析 JSON
        if "```json" in content:
            content = content.split("```json")[1].split("```")[0].strip()
        elif "```" in content:
            content = content.split("```")[1].split("```")[0].strip()
        
        try:
            debug_data = json.loads(content)
        except:
            print("⚠️  JSON 解析失败，构造默认响应")
            debug_data = {
                "has_errors": False,
                "error_info": "",
                "prefix": "Debugged code",
                "imports": "",
                "code": content
            }
        
        return {
            "choices": [{
                "message": {
                    "content": json.dumps(debug_data),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }
    
    except Exception as e:
        print(f"❌ 处理响应失败: {e}")
        return {
            "choices": [{
                "message": {
                    "content": json.dumps({
                        "has_errors": True,
                        "error_info": str(e)
                    }),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }


def generate_checker_response(messages: List[Dict[str, str]]) -> Dict[str, Any]:
    """生成 Checker 响应 - 调用真实 API"""
    print("\n✅ Checker: 调用 Kimi API 验证代码...")
    
    # 构建完整的消息列表
    full_messages = [
        {"role": "system", "content": SYSTEM_PROMPTS["checker"]}
    ] + messages
    
    # 调用 API
    api_response = call_moonshot_api(full_messages, temperature=0.2)
    
    if "error" in api_response:
        return {
            "choices": [{
                "message": {
                    "content": json.dumps({
                        "is_valid": False,
                        "message": f"Validation error: {api_response['error']}",
                        "errors": [api_response['error']],
                        "requirements_met": [],
                        "requirements_missing": ["Unknown"]
                    }),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }
    
    try:
        content = api_response["choices"][0]["message"]["content"]
        print(f"✓ Kimi 返回: {content[:100]}...")
        
        # 清理并解析 JSON
        if "```json" in content:
            content = content.split("```json")[1].split("```")[0].strip()
        elif "```" in content:
            content = content.split("```")[1].split("```")[0].strip()
        
        try:
            checker_data = json.loads(content)
            
            # 处理 Kimi 可能返回的不同格式
            # 如果有 requirements_met 字段但没有 is_valid，转换格式
            if "requirements_met" in checker_data and "is_valid" not in checker_data:
                # Kimi 返回的格式: {"requirements_met": true, "can_modify": false, ...}
                is_met = checker_data.get("requirements_met", False)
                checker_data["is_valid"] = bool(is_met)
                checker_data["message"] = checker_data.get("feedback", "Validation completed")
                checker_data["errors"] = [] if is_met else ["Some requirements not met"]
                if "requirements_met" not in checker_data or not isinstance(checker_data.get("requirements_met"), list):
                    checker_data["requirements_met"] = []
                if "requirements_missing" not in checker_data:
                    checker_data["requirements_missing"] = []
            
            # 确保有必需的字段
            if "is_valid" not in checker_data:
                checker_data["is_valid"] = False
            if "message" not in checker_data:
                checker_data["message"] = "Validation completed"
            if "errors" not in checker_data:
                checker_data["errors"] = []
        except:
            print("⚠️  JSON 解析失败，构造默认响应")
            # 尝试从文本判断
            is_valid = any(word in content.lower() for word in ["valid", "correct", "pass", "通过", "正确", "requirements_met\": true"])
            checker_data = {
                "is_valid": is_valid,
                "message": content[:200],
                "errors": [] if is_valid else ["Unable to parse detailed validation"],
                "requirements_met": [],
                "requirements_missing": []
            }
        
        return {
            "choices": [{
                "message": {
                    "content": json.dumps(checker_data),
                    "role": "assistant"
                },
                "finish_reason": "stop",
                "index": 0
            }]
        }
    
    except Exception as e:
        print(f"❌ 处理响应失败: {e}")
        return {
            "choices": [{
                "message": {
                    "content": json.dumps({
                        "is_valid": False,
                        "message": f"Error: {str(e)}",
                        "errors": [str(e)],
                        "requirements_met": [],
                        "requirements_missing": []
                    }),
                    "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')
        
        print(f"\n📨 收到请求 - 模型: {model}")
        print(f"   消息数: {len(messages)}")
        
        # 根据模型类型调用不同的处理函数
        if 'coder' in model.lower() or 'code' in model.lower():
            response = generate_coder_response(messages)
        elif 'debug' in model.lower() or 'fix' in model.lower():
            response = generate_debugger_response(messages)
        elif 'check' in model.lower() or 'valid' in model.lower():
            response = generate_checker_response(messages)
        else:
            # 默认使用 coder
            response = generate_coder_response(messages)
        
        # 添加响应元数据
        response.update({
            "model": model,
            "object": "chat.completion",
            "created": int(time.time()),
            "usage": {
                "prompt_tokens": sum(len(m.get("content", "")) for m in messages),
                "completion_tokens": len(str(response)),
                "total_tokens": sum(len(m.get("content", "")) for m in messages) + len(str(response))
            }
        })
        
        return jsonify(response)
    
    except Exception as e:
        print(f"❌ 请求处理失败: {e}")
        return jsonify({
            "error": {
                "message": str(e),
                "type": "internal_error",
                "code": "internal_error"
            }
        }), 500


@app.route('/health', methods=['GET'])
def health_check():
    """健康检查端点"""
    # 测试 API 连接
    try:
        test_response = call_moonshot_api([
            {"role": "user", "content": "test"}
        ])
        api_healthy = "error" not in test_response
    except:
        api_healthy = False
    
    return jsonify({
        "status": "healthy" if api_healthy else "degraded",
        "service": "real-llm-server",
        "api": "moonshot-kimi",
        "model": MOONSHOT_MODEL,
        "api_status": "connected" if api_healthy else "disconnected",
        "timestamp": int(time.time())
    })


@app.route('/models', methods=['GET'])
def list_models():
    """列出可用的模型"""
    return jsonify({
        "object": "list",
        "data": [
            {
                "id": "coder-model",
                "object": "model",
                "created": int(time.time()),
                "owned_by": "moonshot-kimi",
                "backend": MOONSHOT_MODEL
            },
            {
                "id": "checker-model",
                "object": "model",
                "created": int(time.time()),
                "owned_by": "moonshot-kimi",
                "backend": MOONSHOT_MODEL
            },
            {
                "id": "debugger-model",
                "object": "model",
                "created": int(time.time()),
                "owned_by": "moonshot-kimi",
                "backend": MOONSHOT_MODEL
            }
        ]
    })


if __name__ == '__main__':
    print("="*60)
    print("🚀 启动真实 LLM 服务器 (Moonshot Kimi API)")
    print("="*60)
    print(f"🤖 模型: {MOONSHOT_MODEL}")
    print(f"🌐 API: {MOONSHOT_BASE_URL}")
    print(f"🔑 API Key: {MOONSHOT_API_KEY[:20]}...")
    print("\n📡 端点:")
    print("   - POST /v1/chat/completions")
    print("   - GET  /health")
    print("   - GET  /models")
    print("\n支持的模型:")
    print("   - coder-model (代码生成)")
    print("   - checker-model (代码验证)")
    print("   - debugger-model (错误修复)")
    print("\n💡 使用真实的 Kimi API 进行推理")
    print("="*60)
    print("\n🔧 按 Ctrl+C 停止服务\n")
    
    # 启动服务器
    app.run(host='0.0.0.0', port=8000, debug=False, threaded=True)