Method stream_function_call

test/test_api/test_openai_api.py:136–180

stream Function calling

(self, message: str, tools: List[Dict], tool_choice: str = "auto", **kwargs)

            raise Exception(f"API调用失败: {response.status_code} - {response.text}")

    def stream_function_call(self, message: str, tools: List[Dict], tool_choice: str = "auto", **kwargs):
        """stream Function calling"""
        data = {
            "model": self.model_name,
            "messages": [{"role": "user", "content": message}],
            "tools": tools,
            "tool_choice": tool_choice,
            "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:
            content_buffer = ""
            tool_calls_buffer = []

            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)
                            delta = chunk["choices"][0]["delta"]

                            # 处理内容
                            if delta.get("content"):
                                content_buffer += delta["content"]
                                yield {"type": "content", "data": delta["content"]}

                            # 处理函数调用
                            if delta.get("tool_calls"):
                                for tool_call in delta["tool_calls"]:
                                    tool_calls_buffer.append(tool_call)
                                    yield {"type": "tool_call", "data": tool_call}

                        except json.JSONDecodeError:
                            continue
        else:
            raise Exception(f"API调用失败: {response.status_code} - {response.text}")

    def completions_with_tokens(self, token_ids: List[int], **kwargs) -> Dict[str, Any]:
        """使用token数组进行文本补全"""