import logging import os from typing import Any, Dict, List, Mapping, Optional from langchain.llms.base import BaseLLM, LLM from langchain.schema import LLMResult from langchain.utils import get_from_dict_or_env from langchain.callbacks.manager import CallbackManagerForLLMRun, Callbacks from enum import Enum from pydantic import root_validator, Field from .xinghuo import SparkApi from .xinghuo.ws import SparkAPI logger = logging.getLogger(__name__) text = [] # length = 0 def getText(role, content): jsoncon = {"role": role, "content": content} text.append(jsoncon) return text def getlength(text): length = 0 for content in text: temp = content["content"] leng = len(temp) length += leng return length def checklen(_text): while getlength(_text) > 8000: del _text[0] return _text class SparkLLM(LLM): """ ErnieLLM is a LLM that uses Ernie to generate text. """ appid: str = Field( None, description="APPID", ) api_key: str = Field( None, description="API_KEY", ) api_secret: str = Field( None, description="API_SECRET", ) version: str = Field( None, description="version", ) api: SparkAPI = Field( None, description="api", ) @root_validator() def validate_environment(self, values: Dict) -> Dict: """Validate the environment.""" # print(values) appid = get_from_dict_or_env(values, "appid", "XH_APPID", "") api_key = get_from_dict_or_env(values, "api_key", "XH_API_KEY", "") api_secret = get_from_dict_or_env(values, "api_secret", "XH_API_SECRET", "") version = values.get("version", "v1") if not appid: raise ValueError("No appid provided.") if not api_key: raise ValueError("No api_key provided.") if not api_secret: raise ValueError("No api_secret provided.") values["appid"] = appid values["api_key"] = api_key values["api_secret"] = api_secret api = SparkAPI(appid, api_key, api_secret, version) values["api"] = api return values def _call(self, prompt: str, stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs) -> str: question = self.getText("user", prompt) try: # 你的代码 # SparkApi.main(self.appid,self.api_key,self.api_secret,self.Spark_url,self.domain,question) self.api.call(question) response = self.api.answer return response except Exception as e: # 处理异常 print("exception:", e) raise e def getText(self, role, content): text = [] jsoncon = {} jsoncon["role"] = role jsoncon["content"] = content text.append(jsoncon) return text @property def _llm_type(self) -> str: """Return type of llm.""" return "xinghuo"