1use async_openai::{
2 config::OpenAIConfig,
3 error::OpenAIError,
4 types::{
5 chat::{ChatCompletionMessageToolCallChunk, FinishReason, Role},
6 stream::StreamResponse,
7 },
8 Client,
9};
10use serde::Deserialize;
11use serde_json::Value as JsonValue;
12
13use crate::event_bus::token_bus::TokenEvent;
14use crate::provider::{build_standard_request_json, ChatProvider, StreamChunkExt};
15
16#[derive(Debug, Deserialize)]
17pub struct Delta {
18 pub content: Option<String>,
19 #[serde(default)]
20 pub reasoning_content: Option<String>,
21 pub role: Option<Role>,
22 #[serde(default)]
23 pub tool_calls: Option<Vec<ChatCompletionMessageToolCallChunk>>,
24}
25
26#[derive(Debug, Deserialize)]
27pub struct Choice {
28 pub index: u32,
29 pub delta: Delta,
30 #[serde(default)]
31 pub finish_reason: Option<FinishReason>,
32}
33
34#[derive(Debug, Deserialize)]
35pub struct StreamChunk {
36 pub id: String,
37 pub choices: Vec<Choice>,
38 pub created: u32,
39 pub model: String,
40 #[serde(default)]
41 pub system_fingerprint: Option<String>,
42 pub object: String,
43}
44
45impl StreamChunkExt for StreamChunk {
46 fn extract_events(&self) -> Vec<TokenEvent> {
47 let mut events = Vec::new();
48 for choice in &self.choices {
49 if let Some(finish_reason) = choice.finish_reason {
50 events.push(TokenEvent::Finish(finish_reason));
51 }
52 if let Some(ref reasoning) = choice.delta.reasoning_content
53 && !reasoning.is_empty()
54 {
55 events.push(TokenEvent::Reasoning(reasoning.clone()));
56 }
57 match (choice.delta.content.as_deref(), choice.delta.tool_calls.as_ref()) {
58 (Some(text), Some(tool_calls)) => {
59 if !text.is_empty() {
60 events.push(TokenEvent::Text(text.to_string()));
61 }
62 for tc in tool_calls {
63 events.push(TokenEvent::ToolDelta {
64 index: tc.index as usize,
65 call_id: tc.id.clone().unwrap_or_default(),
66 name: tc.function.clone().and_then(|f| f.name),
67 args_chunk: tc.function.clone().and_then(|f| f.arguments),
68 });
69 }
70 }
71 (Some(text), None) => {
72 if !text.is_empty() {
73 events.push(TokenEvent::Text(text.to_string()));
74 }
75 }
76 (None, Some(tool_calls)) => {
77 for tc in tool_calls {
78 events.push(TokenEvent::ToolDelta {
79 index: tc.index as usize,
80 call_id: tc.id.clone().unwrap_or_default(),
81 name: tc.function.clone().and_then(|f| f.name),
82 args_chunk: tc.function.clone().and_then(|f| f.arguments),
83 });
84 }
85 }
86 (None, None) => {}
87 }
88 }
89 events
90 }
91}
92
93pub struct DeepSeekProvider;
94
95impl ChatProvider for DeepSeekProvider {
96 type Chunk = StreamChunk;
97
98 fn build_request_json(
99 model: &str,
100 messages: &[JsonValue],
101 skill_content: &str,
102 tools_json: &JsonValue,
103 ) -> JsonValue {
104 let mut json = build_standard_request_json(model, messages, skill_content, tools_json);
105
106 if let Some(msgs) = json["messages"].as_array_mut() {
109 let mut merged: Vec<JsonValue> = Vec::with_capacity(msgs.len());
110
111 let mut i = 0;
112 while i < msgs.len() {
113 let msg = &msgs[i];
114 let role = msg["role"].as_str().unwrap_or("");
115
116 if role == "assistant" && msg.get("tool_calls").and_then(|t| t.as_array()).is_some() {
117 let mut combined_calls = Vec::new();
118 let mut reasoning = None;
119
120 while i < msgs.len() {
121 let cur = &msgs[i];
122 if cur["role"].as_str() != Some("assistant")
123 || cur.get("tool_calls").and_then(|t| t.as_array()).is_none()
124 {
125 break;
126 }
127 if let Some(calls) = cur["tool_calls"].as_array() {
128 combined_calls.extend(calls.iter().cloned());
129 }
130 if reasoning.is_none() {
131 reasoning = cur.get("reasoning_content").and_then(|r| r.as_str()).map(|s| s.to_string());
132 }
133 i += 1;
134 }
135
136 let mut merged_msg = serde_json::json!({
137 "role": "assistant",
138 "content": null,
139 "tool_calls": combined_calls,
140 });
141 if let Some(rc) = reasoning {
142 merged_msg["reasoning_content"] = serde_json::json!(rc);
143 }
144 merged.push(merged_msg);
145 } else {
146 let mut m = msg.clone();
147 if role == "assistant"
148 && let Some(arr) = m.get("tool_calls").and_then(|t| t.as_array())
149 && !arr.is_empty() && !m.as_object().unwrap().contains_key("content")
150 {
151 m.as_object_mut().unwrap().insert("content".into(), JsonValue::Null);
152 }
153 merged.push(m);
154 i += 1;
155 }
156 }
157
158 *msgs = merged;
159 }
160
161 json.as_object_mut()
162 .unwrap()
163 .insert("thinking".into(), serde_json::json!({"type": "enabled"}));
164 json
165 }
166
167 async fn create_stream(
168 client: &Client<OpenAIConfig>,
169 request_json: JsonValue,
170 ) -> Result<StreamResponse<Self::Chunk>, OpenAIError> {
171 client
172 .chat()
173 .create_stream_byot::<JsonValue, Self::Chunk>(request_json)
174 .await
175 }
176}
177
178#[cfg(test)]
179mod tests {
180 use super::*;
181 use serde_json::json;
182
183 #[test]
184 fn build_request_merges_consecutive_tool_calls() {
185 let msgs = vec![
186 json!({"role": "user", "content": "Weather in Tokyo, Beijing, Paris?"}),
187 json!({"role": "assistant", "content": null, "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": r#"{"city":"Tokyo"}"#}}]}),
188 json!({"role": "assistant", "content": null, "tool_calls": [{"id": "call_2", "type": "function", "function": {"name": "get_weather", "arguments": r#"{"city":"Beijing"}"#}}]}),
189 json!({"role": "assistant", "content": null, "tool_calls": [{"id": "call_3", "type": "function", "function": {"name": "get_weather", "arguments": r#"{"city":"Paris"}"#}}]}),
190 json!({"role": "tool", "tool_call_id": "call_1", "content": "22°C"}),
191 json!({"role": "tool", "tool_call_id": "call_2", "content": "18°C"}),
192 json!({"role": "tool", "tool_call_id": "call_3", "content": "25°C"}),
193 ];
194 let result = DeepSeekProvider::build_request_json("test-model", &msgs, "", &json!([]));
195 let msgs_out = result["messages"].as_array().unwrap();
196 assert_eq!(msgs_out.len(), 5, "expected 5: user + 1 merged assistant + 3 tool");
197 assert_eq!(msgs_out[0]["role"], "user");
198 assert_eq!(msgs_out[1]["role"], "assistant");
199 let calls = msgs_out[1]["tool_calls"].as_array().unwrap();
200 assert_eq!(calls.len(), 3);
201 assert_eq!(calls[0]["id"], "call_1");
202 assert_eq!(calls[1]["id"], "call_2");
203 assert_eq!(calls[2]["id"], "call_3");
204 assert!(msgs_out[1]["content"].is_null());
205 assert_eq!(msgs_out[2]["role"], "tool");
206 assert_eq!(msgs_out[3]["role"], "tool");
207 assert_eq!(msgs_out[4]["role"], "tool");
208 }
209
210 #[test]
211 fn build_request_preserves_reasoning_on_merged() {
212 let msgs = vec![
213 json!({"role": "user", "content": "Weather in Tokyo?"}),
214 json!({"role": "assistant", "reasoning_content": "Let me think...", "content": null, "tool_calls": [{"id": "call_1", "function": {"name": "get_weather", "arguments": r#"{"city":"Tokyo"}"#}}]}),
215 json!({"role": "assistant", "content": null, "tool_calls": [{"id": "call_2", "function": {"name": "get_weather", "arguments": r#"{"city":"Osaka"}"#}}]}),
216 json!({"role": "tool", "tool_call_id": "call_1", "content": "22°C"}),
217 json!({"role": "tool", "tool_call_id": "call_2", "content": "20°C"}),
218 json!({"role": "assistant", "content": "Done."}),
219 ];
220 let result = DeepSeekProvider::build_request_json("test-model", &msgs, "", &json!([]));
221 let msgs_out = result["messages"].as_array().unwrap();
222 assert_eq!(msgs_out[1]["reasoning_content"].as_str().unwrap(), "Let me think...");
223 assert_eq!(msgs_out[1]["tool_calls"].as_array().unwrap().len(), 2);
224 }
225
226 #[test]
227 fn build_request_does_not_merge_non_consecutive() {
228 let msgs = vec![
229 json!({"role": "user", "content": "hi"}),
230 json!({"role": "assistant", "content": null, "tool_calls": [{"id": "c1", "function": {"name": "t1", "arguments": "{}"}}]}),
231 json!({"role": "assistant", "content": "I'll get that."}),
232 json!({"role": "assistant", "content": null, "tool_calls": [{"id": "c2", "function": {"name": "t2", "arguments": "{}"}}]}),
233 ];
234 let result = DeepSeekProvider::build_request_json("test-model", &msgs, "", &json!([]));
235 let msgs_out = result["messages"].as_array().unwrap();
236 assert_eq!(msgs_out.len(), 4, "non-consecutive should NOT merge");
237 let calls0 = msgs_out[1]["tool_calls"].as_array().unwrap();
238 assert_eq!(calls0.len(), 1);
239 let calls2 = msgs_out[3]["tool_calls"].as_array().unwrap();
240 assert_eq!(calls2.len(), 1);
241 }
242
243 #[test]
244 fn build_request_adds_content_null_to_tool_calls() {
245 let msgs = vec![
246 json!({"role": "user", "content": "hi"}),
247 json!({"role": "assistant", "tool_calls": [{"id": "c1", "function": {"name": "t1", "arguments": "{}"}}]}),
248 ];
249 let result = DeepSeekProvider::build_request_json("test-model", &msgs, "", &json!([]));
250 let msgs_out = result["messages"].as_array().unwrap();
251 assert!(msgs_out[1]["content"].is_null());
252 assert_eq!(msgs_out[1]["tool_calls"].as_array().unwrap().len(), 1);
253 }
254
255 #[test]
256 fn build_request_includes_thinking() {
257 let msgs = vec![json!({"role": "user", "content": "hi"})];
258 let result = DeepSeekProvider::build_request_json("test-model", &msgs, "", &json!([]));
259 assert_eq!(result["thinking"]["type"], "enabled");
260 }
261}