drop error when llm service is unavailable (#90)

crates/ai/src/llm/openai.rs

@@ -56,7 +56,7 @@ impl LLMModel for OpenAI {
         let mut url = url.join("chat/completions")?;
         url.set_query(query);
 
-        let (tx, mut rx) = mpsc::channel(512);
+        let (tx, mut rx) = mpsc::channel::<anyhow::Result<Option<String>>>(512);
 
         tracing::debug!("openai url: {:?}", url);
 
@@ -201,15 +201,37 @@ impl LLMModel for OpenAI {
                     }
                     Err(e) => {
                         tracing::error!("failed to handle event source: {}", e);
-                        break;
+                        // break;  // 不能 break，原因见下面的注释
+                        if let Err(e) = tx.send(Err(e.into())).await {
+                            tracing::error!("failed to send error: {}", e);
+                        }
                     }
                 }
             }
         });
 
         let stream = async_stream::stream! {
-            while let Some(result) = rx.recv().await {
-                yield result;
+            // while let Some(result) = rx.recv().await {
+            //     yield result;
+            // }
+            // 当所有 tx 都被 drop 的时候，rx.recv() 会收到一次 None
+            // 比如前面如果请求失败 (failed to handle event source) 以后 break 的结果是 rx.recv() 收到 None
+            // 这样不大好，这就区分不出来是请求失败还是所有 stream 请求都结束了
+            // 应该在请求失败的时候 tx.send 一个 Err
+            loop {
+                match rx.recv().await {
+                    Some(Ok(result)) => {
+                        yield Ok(result);
+                    }
+                    Some(Err(e)) => {
+                        // yield Ok(Some(format!("error {:}", e)));
+                        yield Err(e);
+                    }
+                    None => {
+                        // yield Ok(None);
+                        break;
+                    }
+                }
             }
         };
 

crates/ai/src/traits/llm.rs

@@ -37,6 +37,8 @@ impl LLMOutput {
 
     pub async fn to_string(&mut self) -> anyhow::Result<String> {
         let mut output = String::new();
+        // LLMOutput 是 LLMModel::get_completion 返回的结果
+        // 这里会消费 async_stream::stream! 生成的流并且取 yield 的结果
         while let Some(item) = self.next().await {
             let item = item?;
             if let Some(item) = item {
@@ -93,7 +95,11 @@ impl LLMModel {
                     result
                 }
             },
-            |mut v| async move { v.to_string().await },
+            |mut v| async move {
+                // 这个 convert_output 方法在 AIModel::create_reference 里调用,
+                // LLMOutput 是一个异步闭包，调用 convert_output 的时候才会执行 to_string() 并获取 LLM 服务返回的结果
+                v.to_string().await
+            },
         )
     }
 }