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