python调用openai兼容模型api qwen实现ai智能对话生成动态题目表单并自动批改试卷代码-python代码-BFW代码库

代码语言：python

所属分类：其他

代码描述：python调用openai兼容模型api qwen实现ai智能对话生成动态题目表单并自动批改试卷代码，可设定提示词生成任意动态试卷，由ai进行答题批改解读。

代码标签： python 调用 openai 兼容模型api qwen ai 智能对话生成动态题目表

下面为部分代码预览，完整代码请点击下载或在bfwstudio webide中打开

#!/usr/local/python3/bin/python3
# -*- coding: utf-8 -*
import os
import uvicorn
import json
import re
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse
from openai import AsyncOpenAI
from pydantic import BaseModel
from typing import List, Dict, Any

# --- 1. 配置 ---

app = FastAPI()
client = AsyncOpenAI(api_key= "sk-",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")

# --- 2. Pydantic 模型定义 ---
class Message(BaseModel):
    role: str
    content: Any  # 保持 Any 以接收前端的灵活格式

class ChatRequest(BaseModel):
    messages: List[Message]
    model: str = "qwen-max"

class GradeRequest(BaseModel):
    exam: Dict[str, Any]
    answers: Dict[str, Any]
    history: List[Message]
    model: str = "qwen-max"

# --- 3. 后端逻辑 ---

def sanitize_messages(messages: List[Message]) -> List[Dict[str, Any]]:
    """将聊天记录转换为OpenAI API接受的格式"""
    sanitized = []
    for msg in messages:
        content = msg.content
        if isinstance(content, dict):
            content = json.dumps(content, ensure_ascii=False)
        
        sanitized.append({"role": msg.role, "content": content})
    return sanitized

# 【修复1】增强系统指令，确保只返回JSON
EXAM_GENERATION_SYSTEM_PROMPT = """
你是一个专业的试卷生成AI，你的唯一任务是根据用户需求生成符合指定格式的试卷JSON。

严格要求：
1. 必须且只能返回一个纯净的JSON对象，不包含任何JSON之外的内容
2. 不能有解释、说明、注释或Markdown标记
3. 确保JSON格式正确，可被标准JSON解析器解析
4. 严格遵循以下结构:
{
  "title": "试卷标题",
  "description": "对试卷的简短描述",
  "questions": [
    {"id": "q1", "type": "multiple_choice", "question_text": "问题文本", "options": ["选项A", "选项B", "选项C"]},
    {"id": "q2", "type": "short_answer", "question_text": "问题文本"},
    {"id": "q3", "type": "true_false", "question_text": "问题文本"}
  ]
}

支持的type: multiple_choice, short_answer, true_false
如果违反上述要求，你将无法完成任务。
"""

@app.post("/api/generate-exam")
async def generate_exam(request: ChatRequest):
    """接收用户请求，生成试卷JSON"""
    messages_for_api = sanitize_messages(request.messages)

    final_messages = [
        {"role": "system", "content": EXAM_GENERATION_SYSTEM_PROMPT}
    ]
    if messages_for_api:
       final_messages.append(messages_for_api[-1])
       
    try:
        response = await client.chat.completions.create(
            model=request.model,
            messages=final_messages,
            response_format={"type": "json_object"},
            temperature=0.1,
        )
        exam_json_str = response.choices[0].message.content
        
        # 【修复1】增加JSON提取逻辑，处理可能的额外内容
        if exam_json_str:
            # 尝试提取JSON部分
            json_match = re.search(r'\{.*\}', exam_json_str, re.DOTALL)
            if json_match:
                exam_json_str = json_match.group()
                
        exam_data = json.loads(exam_json_str)
        return JSONResponse(content=exam_data)
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}, 原始内容: {exam_json_str}")
        return JSONResponse(status_code=500, content={"error": f"生成的试卷格式无效: {str(e)}"})
    except Exception as e:
        print(f"生成试卷时发生错误: {e}")
        return JSONResponse(status_code=500, content={"error": f"AI模型未能生成有效的试卷格式: {str(e)}"})


@app.post("/api/grade-exam")
async def stream_grade_exam(request: GradeRequest):
    """接收试卷和答案，流式返回批改结果"""
    grading_prompt = f"""
    你是一位严格而公正的AI阅卷老师。以下是一份试卷的原始题目和一位学生的回答。
    原始试卷题目:
    {json.dumps(request.exam, ensure_ascii=False, indent=2)}
    学生的回答:
    {json.dumps(request.answers, ensure_ascii=False, indent=2)}
    请按以下要求完成批改任务：
    1. 对每道题进行评价，明确指出答案的对错与优缺点。
    2. 如果可能，给出一个总分或总体评价。
    3. 你的回答将直接展示给学生，请使用鼓励和友好的语气。
    4. 请使用Markdown格式进行回复，以便清晰展示。
    """
    
    messages_for_api = sanitize_messages(request.history)

    # 【修复2】调整消息顺序，系统提示应放在最前面
    final_messages = [
        {"role": "system", "content": grading_prompt}
    ] + messages_for_api
    
    # 添加用户当前请求
    fi.........完整代码请登录后点击上方下载按钮下载查看