use super::{Client, ExtraConfig, Model, PromptType, QianwenClient, SendData}; use crate::{config::GlobalConfig, render::ReplyHandler, utils::PromptKind}; use anyhow::{anyhow, bail, Result}; use async_trait::async_trait; use futures_util::StreamExt; use reqwest::{Client as ReqwestClient, RequestBuilder}; use reqwest_eventsource::{Error as EventSourceError, Event, RequestBuilderExt}; use serde::Deserialize; use serde_json::{json, Value}; const API_URL: &str = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"; const MODELS: [(&str, usize); 3] = [ ("qwen-turbo", 6144), ("qwen-plus", 6144), ("qwen-max", 6144), ]; #[derive(Debug, Clone, Deserialize, Default)] pub struct QianwenConfig { pub name: Option, pub api_key: Option, pub extra: Option, } #[async_trait] impl Client for QianwenClient { fn config(&self) -> (&GlobalConfig, &Option) { (&self.global_config, &self.config.extra) } async fn send_message_inner(&self, client: &ReqwestClient, data: SendData) -> Result { let builder = self.request_builder(client, data)?; send_message(builder).await } async fn send_message_streaming_inner( &self, client: &ReqwestClient, handler: &mut ReplyHandler, data: SendData, ) -> Result<()> { let builder = self.request_builder(client, data)?; send_message_streaming(builder, handler).await } } impl QianwenClient { config_get_fn!(api_key, get_api_key); pub const PROMPTS: [PromptType<'static>; 1] = [("api_key", "API Key:", true, PromptKind::String)]; pub fn list_models(local_config: &QianwenConfig) -> Vec { let client_name = Self::name(local_config); MODELS .into_iter() .map(|(name, max_tokens)| { Model::new(client_name, name).set_max_tokens(Some(max_tokens)) }) .collect() } fn request_builder(&self, client: &ReqwestClient, data: SendData) -> Result { let api_key = self.get_api_key()?; let stream = data.stream; let body = build_body(data, self.model.name.clone()); debug!("Qianwen Request: {API_URL} {body}"); let mut builder = client.post(API_URL).bearer_auth(api_key).json(&body); if stream { builder = builder.header("X-DashScope-SSE", "enable"); } Ok(builder) } } async fn send_message(builder: RequestBuilder) -> Result { let data: Value = builder.send().await?.json().await?; check_error(&data)?; let output = data["output"]["text"] .as_str() .ok_or_else(|| anyhow!("Unexpected response {data}"))?; Ok(output.to_string()) } async fn send_message_streaming(builder: RequestBuilder, handler: &mut ReplyHandler) -> Result<()> { let mut es = builder.eventsource()?; while let Some(event) = es.next().await { match event { Ok(Event::Open) => {} Ok(Event::Message(message)) => { let data: Value = serde_json::from_str(&message.data)?; if let Some(text) = data["output"]["text"].as_str() { handler.text(text)?; } } Err(err) => { match err { EventSourceError::InvalidStatusCode(_, res) => { let data: Value = res.json().await?; check_error(&data)?; bail!("Request failed"); } EventSourceError::StreamEnded => {} _ => { bail!("{}", err); } } es.close(); } } } Ok(()) } fn check_error(data: &Value) -> Result<()> { if let Some(code) = data["code"].as_str() { if let Some(message) = data["message"].as_str() { bail!("{message}"); } else { bail!("{code}"); } } Ok(()) } fn build_body(data: SendData, model: String) -> Value { let SendData { messages, temperature, stream, } = data; let mut parameters = json!({}); if stream { parameters["incremental_output"] = true.into(); } if let Some(v) = temperature { parameters["temperature"] = v.into(); } json!({ "model": model, "input": json!({ "messages": messages, }), "parameters": parameters }) }