use async_trait::async_trait;
use futures_util::Stream;
use std::pin::Pin;
use super::base::{BaseOutputParser, OutputParserError, OutputParserResult};
use crate::language_models::LLMResult;
use crate::runnables::{Runnable, RunnableConfig};
use crate::structured_output::parser::PartialJsonParser;
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 json = PartialJsonParser::strip_markdown_fence(text);
if json.is_empty() {
Ok(text.trim())
} else {
Ok(json.trim())
}
}
}
impl Default for JsonOutputParser {
fn default() -> Self {
Self::new()
}
}
fn preview_slice(s: &str, max_chars: usize) -> &str {
match s.char_indices().nth(max_chars) {
Some((i, _)) => &s[..i],
None => s,
}
}
#[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,
preview_slice(json_str, 200)
))
})
}
}
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 解析失败:{}",
preview_slice(text, 200)
)))
}
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<LLMResult, serde_json::Value> for JsonOutputParser {
type Error = OutputParserError;
async fn invoke(
&self,
input: LLMResult,
_config: Option<RunnableConfig>,
) -> Result<serde_json::Value, Self::Error> {
self.parse(&input.content).await
}
async fn stream(
&self,
input: LLMResult,
_config: Option<RunnableConfig>,
) -> Result<
Pin<Box<dyn Stream<Item = Result<serde_json::Value, Self::Error>> + Send>>,
Self::Error,
> {
let result = self.parse(&input.content).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_from_markdown_block_unclosed_fence() {
let parser = JsonOutputParser::new();
let input = "以下是结果:\n```json\n{\"status\": \"ok\"}";
let result = parser.parse(input).await.unwrap();
assert_eq!(result["status"], "ok");
}
#[tokio::test]
async fn test_json_parser_from_prose_prefix() {
let parser = JsonOutputParser::new();
let input = "结果是:\n{\"a\": 1}\n以上";
let result = parser.parse(input).await.unwrap();
assert_eq!(result["a"], 1);
}
#[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(
LLMResult {
content: r#"{"key": "value"}"#.to_string(),
..Default::default()
},
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);
}
#[tokio::test]
async fn test_json_parser_invalid_cjk_over_200_bytes() {
let parser = JsonOutputParser::new();
let long_cjk = "汉".repeat(200);
let bad = format!("{{\"名字\": {}", long_cjk);
let result = parser.parse(&bad).await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_json_parser_partial_invalid_cjk_over_200_bytes() {
let parser = JsonOutputParser::new_partial();
let long_cjk = "汉".repeat(200);
let bad = format!("{{\"名字\": {}", long_cjk);
let result = parser.parse(&bad).await;
assert!(result.is_err());
}
}