以非流式方式调用LLM,并返回一次性完成的完整响应。 Args: system_prompt: 系统角色提示 user_prompt: 用户高优先级指令 **kwargs: 允许透传temperature/top_p等采样参数 Returns: 去除首尾空白后的LLM响应文本
(self, system_prompt: str, user_prompt: str, **kwargs)
| 67 | |
| 68 | @with_retry(LLM_RETRY_CONFIG) |
| 69 | def invoke(self, system_prompt: str, user_prompt: str, **kwargs) -> str: |
| 70 | """ |
| 71 | 以非流式方式调用LLM,并返回一次性完成的完整响应。 |
| 72 | |
| 73 | Args: |
| 74 | system_prompt: 系统角色提示 |
| 75 | user_prompt: 用户高优先级指令 |
| 76 | **kwargs: 允许透传temperature/top_p等采样参数 |
| 77 | |
| 78 | Returns: |
| 79 | 去除首尾空白后的LLM响应文本 |
| 80 | """ |
| 81 | messages = [ |
| 82 | {"role": "system", "content": system_prompt}, |
| 83 | {"role": "user", "content": user_prompt}, |
| 84 | ] |
| 85 | |
| 86 | allowed_keys = {"temperature", "top_p", "presence_penalty", "frequency_penalty", "stream"} |
| 87 | extra_params = {key: value for key, value in kwargs.items() if key in allowed_keys and value is not None} |
| 88 | |
| 89 | timeout = kwargs.pop("timeout", self.timeout) |
| 90 | |
| 91 | response = self.client.chat.completions.create( |
| 92 | model=self.model_name, |
| 93 | messages=messages, |
| 94 | timeout=timeout, |
| 95 | **extra_params, |
| 96 | ) |
| 97 | |
| 98 | if response.choices and response.choices[0].message: |
| 99 | return self.validate_response(response.choices[0].message.content) |
| 100 | return "" |
| 101 | |
| 102 | def stream_invoke(self, system_prompt: str, user_prompt: str, **kwargs) -> Generator[str, None, None]: |
| 103 | """ |
no test coverage detected