llm 1.3.8

A Rust library unifying multiple LLM backends.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

use bytes::Bytes;
use futures::stream::StreamExt;

use super::create_sse_stream;
use crate::error::LLMError;

#[tokio::test]
async fn test_create_sse_stream_handles_split_utf8() {
    let test_data = "data: Positive reactions\n\n".as_bytes();

    let chunks: Vec<Result<Bytes, reqwest::Error>> = vec![
        Ok(Bytes::from(&test_data[..10])),
        Ok(Bytes::from(&test_data[10..])),
    ];

    let mock_response = create_mock_response(chunks);

    let parser = |event: &str| -> Result<Option<String>, LLMError> {
        if let Some(content) = event.strip_prefix("data: ") {
            let content = content.trim();
            if content.is_empty() {
                return Ok(None);
            }
            return Ok(Some(content.to_string()));
        }
        Ok(None)
    };

    let mut stream = create_sse_stream(mock_response, parser);

    let mut results = Vec::new();
    while let Some(result) = stream.next().await {
        results.push(result);
    }

    assert_eq!(results.len(), 1);
    assert_eq!(results[0].as_ref().unwrap(), "Positive reactions");
}

#[tokio::test]
async fn test_create_sse_stream_handles_split_sse_events() {
    let event1 = "data: First event\n\n";
    let event2 = "data: Second event\n\n";
    let combined = format!("{event1}{event2}");
    let test_data = combined.as_bytes().to_vec();

    let split_point = event1.len() + 5;
    let chunks: Vec<Result<Bytes, reqwest::Error>> = vec![
        Ok(Bytes::from(test_data[..split_point].to_vec())),
        Ok(Bytes::from(test_data[split_point..].to_vec())),
    ];

    let mock_response = create_mock_response(chunks);

    let parser = |event: &str| -> Result<Option<String>, LLMError> {
        if let Some(content) = event.strip_prefix("data: ") {
            let content = content.trim();
            if content.is_empty() {
                return Ok(None);
            }
            return Ok(Some(content.to_string()));
        }
        Ok(None)
    };

    let mut stream = create_sse_stream(mock_response, parser);

    let mut results = Vec::new();
    while let Some(result) = stream.next().await {
        results.push(result);
    }

    assert_eq!(results.len(), 2);
    assert_eq!(results[0].as_ref().unwrap(), "First event");
    assert_eq!(results[1].as_ref().unwrap(), "Second event");
}

#[tokio::test]
async fn test_create_sse_stream_handles_multibyte_utf8_split() {
    let multibyte_char = "✨";
    let event = format!("data: Star {multibyte_char}\n\n");
    let test_data = event.as_bytes().to_vec();

    let emoji_start = event.find(multibyte_char).unwrap();
    let split_in_emoji = emoji_start + 1;

    let chunks: Vec<Result<Bytes, reqwest::Error>> = vec![
        Ok(Bytes::from(test_data[..split_in_emoji].to_vec())),
        Ok(Bytes::from(test_data[split_in_emoji..].to_vec())),
    ];

    let mock_response = create_mock_response(chunks);

    let parser = |event: &str| -> Result<Option<String>, LLMError> {
        if let Some(content) = event.strip_prefix("data: ") {
            let content = content.trim();
            if content.is_empty() {
                return Ok(None);
            }
            return Ok(Some(content.to_string()));
        }
        Ok(None)
    };

    let mut stream = create_sse_stream(mock_response, parser);

    let mut results = Vec::new();
    while let Some(result) = stream.next().await {
        results.push(result);
    }

    assert_eq!(results.len(), 1);
    assert_eq!(
        results[0].as_ref().unwrap(),
        &format!("Star {multibyte_char}")
    );
}

fn create_mock_response(chunks: Vec<Result<Bytes, reqwest::Error>>) -> reqwest::Response {
    use http_body_util::StreamBody;
    use reqwest::Body;

    let frame_stream = futures::stream::iter(
        chunks
            .into_iter()
            .map(|chunk| chunk.map(hyper::body::Frame::data)),
    );

    let body = StreamBody::new(frame_stream);
    let body = Body::wrap(body);

    let http_response = http::Response::builder().status(200).body(body).unwrap();

    http_response.into()
}