thal 0.0.1

Reactive semantic runtime — molecules, reactions, and effect actors for building LLM-backed applications as dataflow programs.
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

use super::EffectActor;
use crate::runtime::{Molecule, ReactorHandle};
use crate::value::Value;
use crate::Error;
use async_trait::async_trait;
use tokio::io::AsyncWriteExt;

pub struct TerminalWriteActor;

#[async_trait]
impl EffectActor for TerminalWriteActor {
    fn kind_name(&self) -> &'static str {
        "TerminalWrite"
    }

    async fn run(&self, request: Molecule, handle: ReactorHandle) -> Result<(), Error> {
        let stream = request
            .fields
            .get("stream")
            .ok_or_else(|| Error::Runtime("TerminalWrite missing stream".into()))?
            .as_string()?
            .to_string();
        let content = request
            .fields
            .get("content")
            .ok_or_else(|| Error::Runtime("TerminalWrite missing content".into()))?
            .as_string()?
            .to_string();
        let newline = request
            .fields
            .get("newline")
            .ok_or_else(|| Error::Runtime("TerminalWrite missing newline".into()))?
            .as_bool()?;
        let markdown = request
            .fields
            .get("markdown")
            .map(|v| v.as_bool().unwrap_or(false))
            .unwrap_or(false);

        if markdown {
            // termimad prints to stdout synchronously. Wrap in spawn_blocking
            // since it can do non-trivial work (parse + format).
            let body = content.clone();
            tokio::task::spawn_blocking(move || {
                let skin = termimad::MadSkin::default();
                skin.print_text(&body);
            })
            .await
            .map_err(|e| Error::Runtime(format!("termimad task join: {e}")))?;
        } else {
            let line = if newline {
                format!("{content}\n")
            } else {
                content
            };
            match stream.as_str() {
                "stderr" => {
                    let mut stderr = tokio::io::stderr();
                    stderr.write_all(line.as_bytes()).await.map_err(Error::Io)?;
                }
                _ => {
                    let mut stdout = tokio::io::stdout();
                    stdout.write_all(line.as_bytes()).await.map_err(Error::Io)?;
                }
            }
        }

        let mut updated = request.clone();
        updated
            .fields
            .insert("status".into(), Value::String("Done".into()));
        handle.emit(updated).await
    }
}