(url, messages, tools=None, stream=False)
| 100 | |
| 101 | |
| 102 | def chat_completion(url, messages, tools=None, stream=False): |
| 103 | payload = { |
| 104 | "messages": messages, |
| 105 | "stream": stream, |
| 106 | "max_tokens": 4096, |
| 107 | } |
| 108 | if tools: |
| 109 | payload["tools"] = tools |
| 110 | payload["tool_choice"] = "auto" |
| 111 | |
| 112 | try: |
| 113 | response = requests.post(url, json=payload, stream=stream) |
| 114 | response.raise_for_status() |
| 115 | except requests.exceptions.RequestException as e: |
| 116 | body = e.response.content if (e.response is not None) else b"" |
| 117 | print_fail(f"Request error: {e} | body: {body}") |
| 118 | return None |
| 119 | |
| 120 | full_content = "" |
| 121 | reasoning_content = "" |
| 122 | tool_calls: list[dict] = [] |
| 123 | |
| 124 | if stream: |
| 125 | for line in response.iter_lines(): |
| 126 | if not line: |
| 127 | continue |
| 128 | decoded = line.decode("utf-8") |
| 129 | if not decoded.startswith("data: "): |
| 130 | continue |
| 131 | data_str = decoded[6:] |
| 132 | if data_str == "[DONE]": |
| 133 | break |
| 134 | try: |
| 135 | data = json.loads(data_str) |
| 136 | except json.JSONDecodeError: |
| 137 | continue |
| 138 | choices = data.get("choices", []) |
| 139 | if not choices: |
| 140 | continue |
| 141 | delta = choices[0].get("delta", {}) |
| 142 | if delta.get("reasoning_content"): |
| 143 | reasoning_content += delta["reasoning_content"] |
| 144 | if delta.get("content"): |
| 145 | full_content += delta["content"] |
| 146 | print_model_output(delta["content"]) |
| 147 | for tc in delta.get("tool_calls", []): |
| 148 | idx = tc.get("index", 0) |
| 149 | while len(tool_calls) <= idx: |
| 150 | tool_calls.append( |
| 151 | { |
| 152 | "id": "", |
| 153 | "type": "function", |
| 154 | "function": {"name": "", "arguments": ""}, |
| 155 | } |
| 156 | ) |
| 157 | if "id" in tc: |
| 158 | tool_calls[idx]["id"] += tc["id"] |
| 159 | if "function" in tc: |
no test coverage detected