Code -同步调用大模型API 处理文本

发表于   |   分类于 技术 , Python   |   暂无评论

核心第一步——同步调用大模型API（理解API交互）

import requests
import os
from dotenv import load_dotenv

1. 环境配置

load_dotenv()
LLM_API_KEY = os.getenv("LLM_API_KEY")
if not LLM_API_KEY:

raise ValueError("❌ 未设置 LLM_API_KEY，请检查 .env 文件")

LLM_API_URL = (

"https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"

)

2. 同步调用API

def call_llm_sync(text: str):

"""同步调用 DashScope Qwen 模型进行文本分类"""
try:
    # 构造 Prompt（关键：让模型知道要做什么）
    categories = ["政策通知", "群众咨询", "投诉举报", "政务公告", "其他"]
    prompt = (
        f"你是一个政务文本分类器。请将以下文本严格分类为：{', '.join(categories)}。\n"
        f"只输出类别名称，不要任何解释或标点。\n\n"
        f"文本：{text}\n"
        f"类别："
    )

    # 构造请求头和请求体
    headers = {
        "Authorization": f"Bearer {LLM_API_KEY}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": "qwen-turbo-latest",  # 或 qwen-max / qwen-plus
        "input": {"messages": [{"role": "user", "content": prompt}]},
        "parameters": {
            "max_tokens": 10,
            "temperature": 0.1,  # 降低随机性，提高稳定性
            "seed": 12345,  # 可选：固定种子
        },
    }

    # 发送 POST 请求
    response = requests.post(
        url=LLM_API_URL, headers=headers, json=payload, timeout=30
    )
    response.raise_for_status()  # 抛出 HTTP 异常（如 401, 400, 500）

    result = response.json()

    # 提取模型输出
    output_text = result.get("output", {}).get("text", "").strip()

    # 简单置信度模拟（实际无法获取真实 confidence，可设为 1.0 或基于规则）
    confidence = 1.0 if output_text in categories else 0.0

    return {
        "text": text,
        "category": output_text if confidence > 0 else "",
        "confidence": confidence,
        "status": "成功" if confidence > 0 else "失败",
        "error_msg": "" if confidence > 0 else f"模型返回无效类别: '{output_text}'",
    }

except Exception as e:
    return {
        "text": text,
        "category": "",
        "confidence": 0.0,
        "status": "失败",
        "error_msg": str(e),
    }

测试代码

if name == "__main__":

test_text = "群众投诉小区垃圾清运不及时，影响生活环境"
result = call_llm_sync(test_text)
print("API调用结果：", result)
添加新评论