Class LightLLMClient

test/test_api/test_openai_api.py:15–344

LightLLM OpenAI API test cases

class LightLLMClient:
    """LightLLM OpenAI API test cases"""

    def __init__(self, base_url: str = "http://localhost:8000", model_name: str = "your_model_name"):
        self.base_url = base_url
        self.model_name = model_name
        self.headers = {"Content-Type": "application/json"}
        self.conversation_history = []

    def simple_chat(self, message: str, **kwargs) -> Dict[str, Any]:
        """简单对话"""
        data = {
            "model": self.model_name,
            "messages": [{"role": "user", "content": message}],
            "temperature": kwargs.get("temperature", 0.7),
            "max_tokens": kwargs.get("max_tokens", 1000),
            **kwargs,
        }

        response = requests.post(f"{self.base_url}/v1/chat/completions", headers=self.headers, json=data)

        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f"API调用失败: {response.status_code} - {response.text}")

    def stream_chat(self, message: str, **kwargs):
        data = {
            "model": self.model_name,
            "messages": [{"role": "user", "content": message}],
            "stream": True,
            "temperature": kwargs.get("temperature", 0.7),
            "max_tokens": kwargs.get("max_tokens", 1000),
            **kwargs,
        }

        response = requests.post(f"{self.base_url}/v1/chat/completions", headers=self.headers, json=data, stream=True)

        if response.status_code == 200:
            for line in response.iter_lines():
                if line:
                    line = line.decode("utf-8")
                    if line.startswith("data: "):
                        data_str = line[6:]
                        if data_str == "[DONE]":
                            break
                        try:
                            chunk = json.loads(data_str)
                            if chunk["choices"][0]["delta"].get("content"):
                                yield chunk["choices"][0]["delta"]["content"]
                        except json.JSONDecodeError:
                            continue
        else:
            raise Exception(f"API调用失败: {response.status_code} - {response.text}")

    def completions(self, prompt: str, **kwargs) -> Dict[str, Any]:
        """文本补全"""
        data = {