From eb428da14995ef7c0dbcb929786730e0f2453b54 Mon Sep 17 00:00:00 2001 From: unknown <3058704216@qq.com> Date: Tue, 14 Jul 2026 08:41:39 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(providers):=20=E4=B8=BA=20OpenAIModelC?= =?UTF-8?q?onfig=20=E6=B7=BB=E5=8A=A0=20extra=5Fbody=20=E5=AD=97=E6=AE=B5?= =?UTF-8?q?=EF=BC=8C=E5=85=81=E8=AE=B8=E7=94=A8=E6=88=B7=E4=BC=A0=E5=85=A5?= =?UTF-8?q?=E4=BB=BB=E6=84=8F=20API=20=E6=89=A9=E5=B1=95=E5=8F=82=E6=95=B0?= =?UTF-8?q?=EF=BC=88=E5=A6=82=20DeepSeek=20V4=20=E7=9A=84=20thinking?= =?UTF-8?q?=EF=BC=89=EF=BC=8C=E4=BD=BF=E7=94=A8=20serde(flatten)=20?= =?UTF-8?q?=E5=B0=86=20extra=5Fbody=20=E5=86=85=E5=AE=B9=E5=B1=95=E5=BC=80?= =?UTF-8?q?=E5=88=B0=E8=AF=B7=E6=B1=82=E6=A0=B9=E8=8A=82=E7=82=B9=EF=BC=8C?= =?UTF-8?q?=E8=A7=A3=E5=86=B3=E6=97=A0=E6=B3=95=E7=A6=81=E7=94=A8=E6=80=9D?= =?UTF-8?q?=E8=80=83=E6=A8=A1=E5=BC=8F=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/langgraph-providers/src/openai/model.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/crates/langgraph-providers/src/openai/model.rs b/crates/langgraph-providers/src/openai/model.rs index 34fb421..a267368 100644 --- a/crates/langgraph-providers/src/openai/model.rs +++ b/crates/langgraph-providers/src/openai/model.rs @@ -96,6 +96,9 @@ struct RawRequest { stream_options: Option, #[serde(skip_serializing_if = "Option::is_none")] response_format: Option, + #[serde(flatten)] + #[serde(skip_serializing_if = "Option::is_none")] + extra_body: Option, } fn is_false(b: &bool) -> bool { @@ -226,6 +229,8 @@ pub struct OpenAIModelConfig { pub presence_penalty: Option, /// Response format (e.g., {"type": "json_object"} or {"type": "json_schema", ...}). pub response_format: Option, + ///Extra Body + pub extra_body: Option, } impl Default for OpenAIModelConfig { @@ -240,6 +245,7 @@ impl Default for OpenAIModelConfig { frequency_penalty: None, presence_penalty: None, response_format: None, + extra_body:None, } } } @@ -468,6 +474,7 @@ impl BaseChatModel for OpenAIModel { stream: false, stream_options: None, response_format: self.config.response_format.clone(), + extra_body: self.config.extra_body.clone(), }; let response = self @@ -546,7 +553,9 @@ impl BaseChatModel for OpenAIModel { stream: true, stream_options: Some(StreamOptions { include_usage: true }), response_format: self.config.response_format.clone(), + extra_body: self.config.extra_body.clone(), }; + eprintln!("[REQUEST] {}", serde_json::to_string_pretty(&request).unwrap()); let es_builder = self .client From 43501081659792c80f81e24541f17ff71425e83b Mon Sep 17 00:00:00 2001 From: unknown <3058704216@qq.com> Date: Tue, 14 Jul 2026 08:56:02 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E5=88=A0=E9=99=A4=E8=B0=83=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/langgraph-providers/src/openai/model.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/langgraph-providers/src/openai/model.rs b/crates/langgraph-providers/src/openai/model.rs index a267368..9a6424c 100644 --- a/crates/langgraph-providers/src/openai/model.rs +++ b/crates/langgraph-providers/src/openai/model.rs @@ -555,7 +555,6 @@ impl BaseChatModel for OpenAIModel { response_format: self.config.response_format.clone(), extra_body: self.config.extra_body.clone(), }; - eprintln!("[REQUEST] {}", serde_json::to_string_pretty(&request).unwrap()); let es_builder = self .client From a00425667af4f4b391447ad516a904c4b2a00be7 Mon Sep 17 00:00:00 2001 From: unknown <3058704216@qq.com> Date: Sat, 18 Jul 2026 14:09:59 +0800 Subject: [PATCH 3/3] =?UTF-8?q?improved:=20=E5=9C=A8OpenAIModel=E4=B8=AD?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0santitize=5Fextra=5Fbody,=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E9=A2=9D=E5=A4=96=E8=AF=B7=E6=B1=82=E4=BD=93=E4=B8=8E=E5=B7=B2?= =?UTF-8?q?=E6=9C=89=E7=9A=84=E5=AD=97=E6=AE=B5=E4=BA=A7=E7=94=9F=E5=86=B2?= =?UTF-8?q?=E7=AA=81=E3=80=82=E6=B7=BB=E5=8A=A0=E4=BA=86=E5=AF=B9=E5=BA=94?= =?UTF-8?q?=E7=9A=84=E6=B5=8B=E8=AF=95=E7=94=A8=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/langgraph-derive/src/lib.rs | 1 - .../langgraph-providers/src/openai/model.rs | 328 ++++++++++++------ 2 files changed, 213 insertions(+), 116 deletions(-) diff --git a/crates/langgraph-derive/src/lib.rs b/crates/langgraph-derive/src/lib.rs index 44589e4..58ccd6c 100644 --- a/crates/langgraph-derive/src/lib.rs +++ b/crates/langgraph-derive/src/lib.rs @@ -338,7 +338,6 @@ fn impl_tool_macro(name_lit: &Option, desc_lit: &Option, func: &ItemFn let extractions: Vec = params.iter().map(|(name, ty)| { let name_str = name.to_string(); let err_invalid = format!("invalid parameter '{}': {{}}", name_str); - if is_option(ty) { quote! { let #name: #ty = match args.get(#name_str) { diff --git a/crates/langgraph-providers/src/openai/model.rs b/crates/langgraph-providers/src/openai/model.rs index 9a6424c..c5ef61c 100644 --- a/crates/langgraph-providers/src/openai/model.rs +++ b/crates/langgraph-providers/src/openai/model.rs @@ -245,7 +245,7 @@ impl Default for OpenAIModelConfig { frequency_penalty: None, presence_penalty: None, response_format: None, - extra_body:None, + extra_body: None, } } } @@ -441,6 +441,24 @@ impl OpenAIModel { cache_read_tokens: cached, } } + + fn sanitize_extra_body( + extra: Option, + reserved_keys: &[&str], + ) -> Option { + let mut extra = extra?; + if let Some(obj) = extra.as_object_mut() { + for key in reserved_keys { + obj.remove(*key); + } + if obj.is_empty() { + return None; + } + return Some(serde_json::Value::Object(obj.clone())); + } else { + return None; + } + } } #[async_trait] @@ -462,6 +480,24 @@ impl BaseChatModel for OpenAIModel { messages: &[Message], _config: &RunnableConfig, ) -> Result { + //构造extra_body + let extra_body = Self::sanitize_extra_body( + self.config.extra_body.clone(), + &[ + "model", + "messages", + "temperature", + "max_tokens", + "top_p", + "frequency_penalty", + "presence_penalty", + "tools", + "stream", + "stream_options", + "response_format", + ], + ); + let request = RawRequest { model: self.config.model.clone(), messages: self.build_messages(messages), @@ -474,7 +510,7 @@ impl BaseChatModel for OpenAIModel { stream: false, stream_options: None, response_format: self.config.response_format.clone(), - extra_body: self.config.extra_body.clone(), + extra_body, }; let response = self @@ -541,119 +577,125 @@ impl BaseChatModel for OpenAIModel { _config: &'a RunnableConfig, ) -> MessageStream<'a> { Box::pin(async_stream::stream! { - let request = RawRequest { - model: self.config.model.clone(), - messages: self.build_messages(messages), - temperature: self.config.temperature, - max_tokens: self.config.max_tokens, - top_p: self.config.top_p, - frequency_penalty: self.config.frequency_penalty, - presence_penalty: self.config.presence_penalty, - tools: self.build_tools(), - stream: true, - stream_options: Some(StreamOptions { include_usage: true }), - response_format: self.config.response_format.clone(), - extra_body: self.config.extra_body.clone(), - }; - - let es_builder = self - .client - .post(self.api_url()) - .header("Authorization", format!("Bearer {}", self.config.api_key)) - .header("Content-Type", "application/json") - .json(&request); - - let mut event_source = es_builder - .eventsource() - .map_err(|e| ModelError::Invocation(e.to_string()))?; - - let mut accumulated_content = String::new(); - let mut accumulated_thinking = String::new(); - let mut tool_call_buffers: Vec<(Option, String, String)> = Vec::new(); - let mut usage: Option = None; - - while let Some(event) = event_source.next().await { - let event = event.map_err(|e| ModelError::Invocation(e.to_string()))?; - - match event { - Event::Open => continue, - Event::Message(msg) => { - if msg.data == "[DONE]" { - break; - } - - let chunk: StreamChunk = serde_json::from_str(&msg.data) - .map_err(|e| ModelError::Invocation(e.to_string()))?; - - if let Some(u) = chunk.usage { - usage = Some(Self::extract_usage(&u)); - } - - if let Some(choice) = chunk.choices.first() { - let delta = &choice.delta; - - // Stream thinking delta — incremental only - if let Some(ref thinking) = delta.reasoning_content { - accumulated_thinking.push_str(thinking); - yield Ok(Message::ai_with_thinking("", thinking.clone())); - } - - // Stream answer delta — incremental only - if let Some(ref content) = delta.content { - accumulated_content.push_str(content); - yield Ok(Message::ai(content.clone())); - } - - // Accumulate tool call fragments (not yielded until done) - if let Some(calls) = &delta.tool_calls { - for tc in calls { - let idx = tc.index; - while tool_call_buffers.len() <= idx { - tool_call_buffers.push((None, String::new(), String::new())); - } - let buf = &mut tool_call_buffers[idx]; - if let Some(id) = &tc.id { - buf.0 = Some(id.clone()); - } - if let Some(func) = &tc.function { - if let Some(name) = &func.name { - if !name.is_empty() { - buf.1 = name.clone(); - } - } - if let Some(args) = &func.arguments { - buf.2.push_str(args); - } - } - } - } - } - } - } - } - - // After the SSE stream ends, yield ONE final chunk containing only - // the assembled tool calls (if any). Content/thinking are left empty - // because they have already been streamed incrementally above. - // Consumers that only need the final assembled Message (e.g. invoke) - // should call `ainvoke` instead. Consumers of `astream` that need - // tool calls can detect this chunk via `has_tool_calls()`. - if !tool_call_buffers.is_empty() { - let tool_calls: Vec = tool_call_buffers - .into_iter() - .filter(|(_, name, _)| !name.is_empty()) - .map(|(id, name, args)| { - let args_json = serde_json::from_str(&args) - .unwrap_or(serde_json::json!({})); - ToolCall { name, args: args_json, id } - }) - .collect(); - - yield Ok(common::build_ai_message(String::new(), tool_calls, None, usage)); - } else if usage.is_some() { - yield Ok(common::build_ai_message(String::new(), Vec::new(), None, usage)); - } - }) + let extra_body = Self::sanitize_extra_body(self.config.extra_body.clone(), + &["model", "messages", "temperature", "max_tokens", "top_p", + "frequency_penalty", "presence_penalty", "tools", "stream", + "stream_options", "response_format"]); + + + let request = RawRequest { + model: self.config.model.clone(), + messages: self.build_messages(messages), + temperature: self.config.temperature, + max_tokens: self.config.max_tokens, + top_p: self.config.top_p, + frequency_penalty: self.config.frequency_penalty, + presence_penalty: self.config.presence_penalty, + tools: self.build_tools(), + stream: true, + stream_options: Some(StreamOptions { include_usage: true }), + response_format: self.config.response_format.clone(), + extra_body, + }; + + let es_builder = self + .client + .post(self.api_url()) + .header("Authorization", format!("Bearer {}", self.config.api_key)) + .header("Content-Type", "application/json") + .json(&request); + + let mut event_source = es_builder + .eventsource() + .map_err(|e| ModelError::Invocation(e.to_string()))?; + + let mut accumulated_content = String::new(); + let mut accumulated_thinking = String::new(); + let mut tool_call_buffers: Vec<(Option, String, String)> = Vec::new(); + let mut usage: Option = None; + + while let Some(event) = event_source.next().await { + let event = event.map_err(|e| ModelError::Invocation(e.to_string()))?; + + match event { + Event::Open => continue, + Event::Message(msg) => { + if msg.data == "[DONE]" { + break; + } + + let chunk: StreamChunk = serde_json::from_str(&msg.data) + .map_err(|e| ModelError::Invocation(e.to_string()))?; + + if let Some(u) = chunk.usage { + usage = Some(Self::extract_usage(&u)); + } + + if let Some(choice) = chunk.choices.first() { + let delta = &choice.delta; + + // Stream thinking delta — incremental only + if let Some(ref thinking) = delta.reasoning_content { + accumulated_thinking.push_str(thinking); + yield Ok(Message::ai_with_thinking("", thinking.clone())); + } + + // Stream answer delta — incremental only + if let Some(ref content) = delta.content { + accumulated_content.push_str(content); + yield Ok(Message::ai(content.clone())); + } + + // Accumulate tool call fragments (not yielded until done) + if let Some(calls) = &delta.tool_calls { + for tc in calls { + let idx = tc.index; + while tool_call_buffers.len() <= idx { + tool_call_buffers.push((None, String::new(), String::new())); + } + let buf = &mut tool_call_buffers[idx]; + if let Some(id) = &tc.id { + buf.0 = Some(id.clone()); + } + if let Some(func) = &tc.function { + if let Some(name) = &func.name { + if !name.is_empty() { + buf.1 = name.clone(); + } + } + if let Some(args) = &func.arguments { + buf.2.push_str(args); + } + } + } + } + } + } + } + } + + // After the SSE stream ends, yield ONE final chunk containing only + // the assembled tool calls (if any). Content/thinking are left empty + // because they have already been streamed incrementally above. + // Consumers that only need the final assembled Message (e.g. invoke) + // should call `ainvoke` instead. Consumers of `astream` that need + // tool calls can detect this chunk via `has_tool_calls()`. + if !tool_call_buffers.is_empty() { + let tool_calls: Vec = tool_call_buffers + .into_iter() + .filter(|(_, name, _)| !name.is_empty()) + .map(|(id, name, args)| { + let args_json = serde_json::from_str(&args) + .unwrap_or(serde_json::json!({})); + ToolCall { name, args: args_json, id } + }) + .collect(); + + yield Ok(common::build_ai_message(String::new(), tool_calls, None, usage)); + } else if usage.is_some() { + yield Ok(common::build_ai_message(String::new(), Vec::new(), None, usage)); + } + }) } fn bind_tools(&self, tools: Vec) -> Box { @@ -774,4 +816,60 @@ mod tests { assert_eq!(msg.thinking(), Some("Let me think: 2+2=4")); assert_eq!(msg.text(), Some("The answer is 4")); } + + #[test] + fn test_extra_body_flatten() { + let config = OpenAIModelConfig { + model: "gpt-4".to_string(), + api_key: "test".to_string(), + extra_body: Some(serde_json::json!({ + "enable_thinking": false, + "custom_param": "value" + })), + ..Default::default() + }; + let model = OpenAIModel::new(config); + let request = RawRequest { + model: model.config.model.clone(), + messages: vec![], + temperature: Some(0.7), + max_tokens: None, + top_p: None, + frequency_penalty: None, + presence_penalty: None, + tools: None, + stream: false, + stream_options: None, + response_format: None, + extra_body: model.config.extra_body.clone(), + }; + let value = serde_json::to_value(&request).unwrap(); + assert_eq!(value["enable_thinking"], false); + assert_eq!(value["custom_param"], "value"); + assert!(value.get("extra_body").is_none()); + } + #[test] + fn test_extra_body_conflict_filter() { + let extra = Some(serde_json::json!({ + "temperature": 0.9, // 冲突 + "model": "gpt-5", // 冲突 + "custom": "keep" // 不冲突 + })); + let reserved = vec!["model", "temperature", "stream", "tools", "messages"]; + let filtered = OpenAIModel::sanitize_extra_body(extra, &reserved); + let filtered_value = filtered.unwrap(); + assert!(filtered_value.get("temperature").is_none()); + assert!(filtered_value.get("model").is_none()); + assert_eq!(filtered_value["custom"], "keep"); + } + #[test] + fn test_extra_body_empty_after_filter() { + let extra = Some(serde_json::json!({ + "temperature": 0.9, + "model": "gpt-5" + })); + let reserved = vec!["model", "temperature"]; + let filtered = OpenAIModel::sanitize_extra_body(extra, &reserved); + assert!(filtered.is_none()); + } }