(url, messages, tools=None, stream=False)
| 80 | |
| 81 | |
| 82 | def chat_completion(url, messages, tools=None, stream=False): |
| 83 | payload = { |
| 84 | "messages": messages, |
| 85 | "stream": stream, |
| 86 | "max_tokens": 4096, |
| 87 | } |
| 88 | if tools: |
| 89 | payload["tools"] = tools |
| 90 | payload["tool_choice"] = "auto" |
| 91 | |
| 92 | try: |
| 93 | response = requests.post(url, json=payload, stream=stream) |
| 94 | response.raise_for_status() |
| 95 | except requests.exceptions.RequestException as e: |
| 96 | body = e.response.content if (e.response is not None) else b"" |
| 97 | print_fail(f"Request error: {e} | body: {body}") |
| 98 | return None |
| 99 | |
| 100 | full_content = "" |
| 101 | reasoning_content = "" |
| 102 | tool_calls: list[dict] = [] |
| 103 | |
| 104 | if stream: |
| 105 | for line in response.iter_lines(): |
| 106 | if not line: |
| 107 | continue |
| 108 | decoded = line.decode("utf-8") |
| 109 | if not decoded.startswith("data: "): |
| 110 | continue |
| 111 | data_str = decoded[6:] |
| 112 | if data_str == "[DONE]": |
| 113 | break |
| 114 | try: |
| 115 | data = json.loads(data_str) |
| 116 | except json.JSONDecodeError: |
| 117 | continue |
| 118 | choices = data.get("choices", []) |
| 119 | if not choices: |
| 120 | continue |
| 121 | delta = choices[0].get("delta", {}) |
| 122 | if delta.get("reasoning_content"): |
| 123 | reasoning_content += delta["reasoning_content"] |
| 124 | if delta.get("content"): |
| 125 | full_content += delta["content"] |
| 126 | print_model_output(delta["content"]) |
| 127 | for tc in delta.get("tool_calls", []): |
| 128 | idx = tc.get("index", 0) |
| 129 | while len(tool_calls) <= idx: |
| 130 | tool_calls.append( |
| 131 | { |
| 132 | "id": "", |
| 133 | "type": "function", |
| 134 | "function": {"name": "", "arguments": ""}, |
| 135 | } |
| 136 | ) |
| 137 | if "id" in tc: |
| 138 | tool_calls[idx]["id"] += tc["id"] |
| 139 | if "function" in tc: |
no test coverage detected