Method stream_chat

test/test_api/test_openai_api.py:41–68

(self, message: str, **kwargs)

            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]:
        """文本补全"""