lc_core/output_parsers/
structured_parser.rs1use async_trait::async_trait;
2use futures_util::Stream;
3use serde::de::DeserializeOwned;
4use std::collections::HashMap;
5use std::marker::PhantomData;
6use std::pin::Pin;
7
8use super::base::{BaseOutputParser, OutputParserError, OutputParserResult};
9use crate::language_models::LLMResult;
10use crate::runnables::{Runnable, RunnableConfig};
11
12pub struct StructuredOutputParser {
34 separator: char,
36}
37
38impl StructuredOutputParser {
39 pub fn new() -> Self {
41 Self { separator: ':' }
42 }
43
44 pub fn with_separator(separator: char) -> Self {
46 Self { separator }
47 }
48}
49
50impl Default for StructuredOutputParser {
51 fn default() -> Self {
52 Self::new()
53 }
54}
55
56#[async_trait]
57impl BaseOutputParser<HashMap<String, String>> for StructuredOutputParser {
58 async fn parse(&self, text: &str) -> OutputParserResult<HashMap<String, String>> {
59 let mut map = HashMap::new();
60
61 for line in text.lines() {
62 let line = line.trim();
63 if line.is_empty() {
64 continue;
65 }
66
67 if let Some(pos) = line.find(self.separator) {
68 let sep_len = self.separator.len_utf8();
71 let key = line[..pos].trim().to_string();
72 let value = line[pos + sep_len..].trim().to_string();
73
74 if !key.is_empty() {
75 map.insert(key, value);
76 }
77 }
78 }
79
80 Ok(map)
81 }
82
83 fn get_format_instructions(&self) -> String {
84 format!(
85 "请按以下格式输出(每行一个键值对,使用 '{}' 分隔):\n键{}值",
86 self.separator, self.separator
87 )
88 }
89}
90
91#[async_trait]
92impl Runnable<LLMResult, HashMap<String, String>> for StructuredOutputParser {
93 type Error = OutputParserError;
94
95 async fn invoke(
96 &self,
97 input: LLMResult,
98 _config: Option<RunnableConfig>,
99 ) -> Result<HashMap<String, String>, Self::Error> {
100 self.parse(&input.content).await
101 }
102
103 async fn stream(
104 &self,
105 input: LLMResult,
106 _config: Option<RunnableConfig>,
107 ) -> Result<
108 Pin<Box<dyn Stream<Item = Result<HashMap<String, String>, Self::Error>> + Send>>,
109 Self::Error,
110 > {
111 let result = self.parse(&input.content).await?;
112 let stream = futures_util::stream::once(async move { Ok(result) });
113 Ok(Box::pin(stream))
114 }
115}
116
117pub struct TypedOutputParser<T> {
140 _phantom: PhantomData<T>,
141}
142
143impl<T> TypedOutputParser<T> {
144 pub fn new() -> Self {
146 Self {
147 _phantom: PhantomData,
148 }
149 }
150}
151
152impl<T: DeserializeOwned> Default for TypedOutputParser<T> {
153 fn default() -> Self {
154 Self::new()
155 }
156}
157
158#[async_trait]
159impl<T: DeserializeOwned + Send + Sync + 'static> BaseOutputParser<T> for TypedOutputParser<T> {
160 async fn parse(&self, text: &str) -> OutputParserResult<T> {
161 let text = text.trim();
162
163 let json_str = Self::extract_from_markdown(text).unwrap_or(text);
165
166 serde_json::from_str::<serde_json::Value>(json_str)
168 .map_err(|e| OutputParserError::JsonError(format!("input is not valid JSON: {}", e)))?;
169
170 serde_json::from_str::<T>(json_str).map_err(|e| {
172 OutputParserError::TypeError(format!(
173 "type deserialization failed (check whether the JSON fields match): {}",
174 e
175 ))
176 })
177 }
178}
179
180impl<T: DeserializeOwned> TypedOutputParser<T> {
181 fn extract_from_markdown(text: &str) -> Option<&str> {
183 if let Some(start) = text.find("```json") {
185 let after = &text[start + 7..];
186 if let Some(end) = after.find("```") {
187 return Some(after[..end].trim());
188 }
189 }
190 if let Some(start) = text.find("```") {
192 let after = &text[start + 3..];
193 let after = after.trim();
194 let skip = after.find('\n').unwrap_or(0);
195 let after = &after[skip..].trim();
196 if let Some(end) = after.find("```") {
197 return Some(after[..end].trim());
198 }
199 }
200 None
201 }
202}
203
204#[async_trait]
205impl<T: DeserializeOwned + Send + Sync + 'static> Runnable<LLMResult, T> for TypedOutputParser<T> {
206 type Error = OutputParserError;
207
208 async fn invoke(
209 &self,
210 input: LLMResult,
211 _config: Option<RunnableConfig>,
212 ) -> Result<T, Self::Error> {
213 self.parse(&input.content).await
214 }
215
216 async fn stream(
217 &self,
218 input: LLMResult,
219 _config: Option<RunnableConfig>,
220 ) -> Result<Pin<Box<dyn Stream<Item = Result<T, Self::Error>> + Send>>, Self::Error> {
221 let result = self.parse(&input.content).await?;
222 let stream = futures_util::stream::once(async move { Ok(result) });
223 Ok(Box::pin(stream))
224 }
225}
226
227#[cfg(test)]
228mod tests {
229 use super::*;
230
231 #[tokio::test]
232 async fn test_structured_parser_default_colon() {
233 let parser = StructuredOutputParser::new();
234 let map = parser.parse("姓名: 张三\n年龄: 28").await.unwrap();
235 assert_eq!(map.get("姓名").unwrap(), "张三");
236 assert_eq!(map.get("年龄").unwrap(), "28");
237 }
238
239 #[tokio::test]
240 async fn test_structured_parser_fullwidth_separator() {
241 let parser = StructuredOutputParser::with_separator(':');
243 let map = parser.parse("姓名:张三\n年龄:28").await.unwrap();
244 assert_eq!(map.get("姓名").unwrap(), "张三");
245 assert_eq!(map.get("年龄").unwrap(), "28");
246 }
247
248 #[tokio::test]
249 async fn test_structured_parser_runnable_invoke() {
250 let parser = StructuredOutputParser::new();
251 let map = parser
252 .invoke(
253 LLMResult {
254 content: "状态: 成功".to_string(),
255 ..Default::default()
256 },
257 None,
258 )
259 .await
260 .unwrap();
261 assert_eq!(map.get("状态").unwrap(), "成功");
262 }
263}