llm/ali_llm.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File    :   ali_llm.py
@Time    :   2023/12/01 18:00:00
@Author  :   YourName
@Desc    :   阿里通义千问大模型封装
'''

from langchain.llms.base import LLM
from typing import Any, List, Mapping, Optional, Dict, Union
from pydantic import Field
from llm.self_llm import Self_LLM
from langchain.callbacks.manager import CallbackManagerForLLMRun
import json
import requests

class Ali_LLM(Self_LLM):
    """阿里通义千问大模型封装"""
    url: str = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
    model: str = "qwen-max"  # 添加model属性定义
    
    def __init__(self, model: str = "qwen-max", temperature: float = 0.0, api_key: str = None):
        super().__init__()
        self.model = model
        self.temperature = temperature
        self.api_key = api_key
    
    def _call(self, prompt: str, stop: Optional[List[str]] = None,
              run_manager: Optional[CallbackManagerForLLMRun] = None,
              **kwargs: Any):
        chat_history = kwargs.get('chat_history', [])

         # 添加系统提示词
        system_prompt = """你是一个三国大乱斗系统的AI助手。你能提供以下功能。
系统功能：
1. 角色抽取：随机抽取三国人物卡并展示完整信息，包括：
   - 角色名
   - 角色特点
   - 属性值
   - 技能说明
2.对战规程介绍：
   - 回合制对战规则：
     - 每回合速度快的方先出手
     - 行动选择：每回合只能选择其中一种行动类型
       a) 普通攻击：造成基础伤害
       b) 使用技能：消耗资源发动特殊效果（不与普通攻击叠加）
       c) 休息：回复3点体力和10点法力
     - 技能使用：需要支付相应消耗，无法支付则无法发动
     - 伤害计算的逻辑：
       - 普通攻击伤害 = (攻击方攻击-防御方防御)/防御方耐力*4  
       - 技能伤害/效果独立计算
     - 胜负判定：体力降为0或以下即判负
3. 对战系统：
   a) 角色抽取阶段：
      - 玩家抽取角色后立即固定为玩家角色
      - 自动匹配对手角色
      - 明确显示"玩家角色：[角色名]"和"对手角色：[角色名]"
      
   b) 对战准备：
      - 直接进入对战，不再重复确认角色
      - 显示双方完整属性面板
      - 自动开始第一回合
      
   c) 回合执行：
      - 玩家只需选择行动类型(1.普通攻击 2.技能（列出技能列表） 3.休息（效果：回复3点体力和10点法力）)，每个回合只能执行一个动作（普攻或技能或休息三选一）
      - AI自动处理对手行动
      - 实时显示本回合双方行动的结果，然后询问玩家下个回合要执行什么动作
      
   c) 战报生成：
      - 只显示当前回合选择的行动结果
      - 不混合显示不同行动类型的效果
      - 对战如果结束生成简单汇报"""
        
        messages = [
            {"role": "system", "content": system_prompt},
            *[{"role": "user" if i%2==0 else "assistant", "content": msg[0] if i%2==0 else msg[1]} 
              for i, msg in enumerate(chat_history)],
            {"role": "user", "content": prompt}
        ]
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": self.model,
            "input": {"messages": messages},
            "parameters": {"temperature": self.temperature}
        }
        
        response = requests.post(self.url, headers=headers, json=payload)
        if response.status_code == 200:
            return response.json()["output"]["text"]
        return f"请求失败: {response.text}"
    
    @property
    def _llm_type(self) -> str:
        return "Ali"