use async_trait::async_trait;
use futures_util::Stream;
use std::pin::Pin;
use super::base::{BaseOutputParser, OutputParserError, OutputParserResult};
use crate::core::runnables::{Runnable, RunnableConfig};
pub struct JsonOutputParser {
partial: bool,
}
impl JsonOutputParser {
pub fn new() -> Self {
Self { partial: false }
}
pub fn new_partial() -> Self {
Self { partial: true }
}
fn extract_json_str<'a>(&self, text: &'a str) -> OutputParserResult<&'a str> {
let text = text.trim();
if let Some(start) = text.find("```json") {
let content = &text[start + 7..];
if let Some(end) = content.find("```") {
return Ok(content[..end].trim());
}
}
if let Some(start) = text.find("```") {
let content = &text[start + 3..];
let content = content.trim();
let skip_to_newline = content.find('\n').unwrap_or(0);
let content = &content[skip_to_newline..];
if let Some(end) = content.find("```") {
return Ok(content[..end].trim());
}
}
Ok(text)
}
}
impl Default for JsonOutputParser {
fn default() -> Self {
Self::new()
}
}
#[async_trait]
impl BaseOutputParser<serde_json::Value> for JsonOutputParser {
async fn parse(&self, text: &str) -> OutputParserResult<serde_json::Value> {
let json_str = self.extract_json_str(text)?;
if self.partial {
self.parse_partial_json(json_str)
} else {
serde_json::from_str(json_str).map_err(|e| {
OutputParserError::JsonError(format!(
"JSON 解析失败(位置 {}:{}):{},输入:{}",
e.line(),
e.column(),
e,
&json_str[..std::cmp::min(200, json_str.len())]
))
})
}
}
fn get_format_instructions(&self) -> String {
"请使用 JSON 格式输出,例如:{\"key\": \"value\"}。确保 JSON 是合法的。".to_string()
}
}
impl JsonOutputParser {
fn parse_partial_json(&self, text: &str) -> OutputParserResult<serde_json::Value> {
if let Ok(value) = serde_json::from_str::<serde_json::Value>(text) {
return Ok(value);
}
let repaired = self.repair_partial_json(text);
if let Ok(value) = serde_json::from_str::<serde_json::Value>(&repaired) {
return Ok(value);
}
Err(OutputParserError::JsonError(format!(
"部分 JSON 解析失败:{}",
&text[..std::cmp::min(200, text.len())]
)))
}
fn repair_partial_json(&self, text: &str) -> String {
let mut repaired = text.trim().to_string();
if let Some(stripped) = Self::strip_incomplete_token(&repaired) {
repaired = stripped;
}
let mut in_string = false;
let mut escape_next = false;
let mut open_braces = 0usize;
let mut close_braces = 0usize;
let mut open_brackets = 0usize;
let mut close_brackets = 0usize;
for ch in repaired.chars() {
if escape_next {
escape_next = false;
continue;
}
if ch == '\\' && in_string {
escape_next = true;
continue;
}
if ch == '"' {
in_string = !in_string;
continue;
}
if !in_string {
match ch {
'{' => open_braces += 1,
'}' => close_braces += 1,
'[' => open_brackets += 1,
']' => close_brackets += 1,
_ => {}
}
}
}
for _ in close_braces..open_braces {
repaired.push('}');
}
for _ in close_brackets..open_brackets {
repaired.push(']');
}
let mut in_string = false;
let mut escape_next = false;
let mut last_open_quote_pos: Option<usize> = None;
for (i, ch) in repaired.char_indices() {
if escape_next {
escape_next = false;
continue;
}
if ch == '\\' && in_string {
escape_next = true;
continue;
}
if ch == '"' {
if in_string {
in_string = false;
last_open_quote_pos = None;
} else {
in_string = true;
last_open_quote_pos = Some(i);
}
continue;
}
}
if in_string {
if let Some(open_pos) = last_open_quote_pos {
let after_quote = &repaired[open_pos + 1..];
if after_quote.contains('\n') {
let newline_pos = repaired[open_pos + 1..].find('\n').unwrap() + open_pos + 1;
repaired.truncate(newline_pos);
repaired.push('"');
}
}
}
repaired
}
fn strip_incomplete_token(s: &str) -> Option<String> {
let trimmed = s.trim_end();
let chars: Vec<char> = trimmed.chars().collect();
if chars.is_empty() {
return None;
}
let mut i = chars.len();
while i > 0 {
i -= 1;
match chars[i] {
',' | ':' | '{' | '[' | '}' | ']' => {
let truncate_at: usize = trimmed
.char_indices()
.nth(i + 1)
.map(|(pos, _)| pos)
.unwrap_or(trimmed.len());
if truncate_at < s.len() {
let result = trimmed[..truncate_at].to_string();
if result != s.trim_end() {
return Some(result);
}
}
return None;
}
'"' => {
return None;
}
_ => {}
}
}
None
}
}
#[async_trait]
impl Runnable<String, serde_json::Value> for JsonOutputParser {
type Error = OutputParserError;
async fn invoke(
&self,
input: String,
_config: Option<RunnableConfig>,
) -> Result<serde_json::Value, Self::Error> {
self.parse(&input).await
}
async fn stream(
&self,
input: String,
_config: Option<RunnableConfig>,
) -> Result<
Pin<Box<dyn Stream<Item = Result<serde_json::Value, Self::Error>> + Send>>,
Self::Error,
> {
let result = self.parse(&input).await?;
let stream = futures_util::stream::once(async move { Ok(result) });
Ok(Box::pin(stream))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_json_parser_standard_obj() {
let parser = JsonOutputParser::new();
let result = parser
.parse(r#"{"name": "Rust", "year": 2015}"#)
.await
.unwrap();
assert_eq!(result["name"], "Rust");
assert_eq!(result["year"], 2015);
}
#[tokio::test]
async fn test_json_parser_from_markdown_block() {
let parser = JsonOutputParser::new();
let input = "以下是结果:\n```json\n{\"status\": \"ok\"}\n```\n";
let result = parser.parse(input).await.unwrap();
assert_eq!(result["status"], "ok");
}
#[tokio::test]
async fn test_json_parser_array() {
let parser = JsonOutputParser::new();
let result = parser.parse("[1, 2, 3]").await.unwrap();
assert_eq!(result[0], 1);
assert_eq!(result[2], 3);
}
#[tokio::test]
async fn test_json_parser_invalid_json() {
let parser = JsonOutputParser::new();
let result = parser.parse("{invalid}").await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_json_parser_format_instructions() {
let parser = JsonOutputParser::new();
let instructions = parser.get_format_instructions();
assert!(!instructions.is_empty());
}
#[tokio::test]
async fn test_json_parser_invoke_runnable() {
let parser = JsonOutputParser::new();
let result = parser
.invoke(r#"{"key": "value"}"#.to_string(), None)
.await
.unwrap();
assert_eq!(result["key"], "value");
}
#[tokio::test]
async fn test_json_parser_partial_success() {
let parser = JsonOutputParser::new_partial();
let result = parser.parse(r#"{"a": 1}"#).await.unwrap();
assert_eq!(result["a"], 1);
}
}