Pipes library examples

[ThereIsNoPie]

Use case:
I'm trying to write a console app that communicates sends the conversation to chatgpt and writes the response to console (the actual application is a bit more complicated).

The current code is done imperatively like this (pseudo code for readability).

var conversationHistory = new List<LLMMessage>();
IMachine machine = GetMachine();

// get greeting message from machine.
IAsyncEnumerable<string> responseFromMachine = API.StreamNextResponse(
    conversationHistory,
    machine
);

while (true)
{
    // print response (streaming)
    conversationHistory.Add(new LLMMessage("assistant", responseFromMachine));
    var userMessage= Console.ReadLine();
    conversationHistory.Add(new LLMMessage("user", userMessage));
    responseFromMachine = API.StreamNextResponse(conversationHistory, machine);
}

Problem:
For Proxy<RT, UOut, UIn, DIn, DOut, A>, I can't find any samples that use DIn or UOut.

I assume I want to stream in both directions since both the console and the LLM (chatgpt) are stateful. I don't know what types I should use for Din, UOut, A, etc. I'm also unsure where IMachine fits (does RT contain both Console and LLM?).

Update:
The following code works for synchronous chat but there's a few issues:

1. I can't get streaming of the chat response back to work (there's a some commented out code in v4.4.9 that seems to promise to do what I want but hasn't been implemented yet).
2. I don't know what to do once I have an ongoing state that I need to send around (i.e. "conversationHistory" and the "machine"). Do I put that into the runtime or do I return these things in place of "Unit"?

public class ChatApplication2<RT>
    where RT : struct, HasCancel<RT>, HasConsole<RT>
{
    public record UserInput(string Text);

    public record LLMResponse(string Text);

    public static Aff<RT, Unit> Main =>
        from _ in Console<RT>.writeLine("Chat with AI.")
        from e in chatEffect
        select unit;

    static Effect<RT, Unit> chatEffect => chatServer | chatClient;

    static Server<RT, UserInput, LLMResponse, Unit> chatServer(UserInput userInput) =>
        from res in slift(simulateLLMResponse(userInput))
        from req in Server.respond<RT, UserInput, LLMResponse>(res)
        from _repeat in chatServer(req)
        select unit;

    static Client<RT, UserInput, LLMResponse, Unit> chatClient =>
        from _ in clift(Console<RT>.write("You: "))
        from req in clift(GetUserInput())
        from res in Client.request<RT, UserInput, LLMResponse>(req)
        from _2 in clift(Console<RT>.writeLine($"AI: {res.Text}"))
        from _repeat in chatClient
        select unit;

    static Eff<RT, UserInput> GetUserInput()
    {
        return from x in Console<RT>.readLine select new UserInput(x);
    }

    static Aff<RT, LLMResponse> simulateLLMResponse(UserInput input)
    {
        return Aff<RT, LLMResponse>(async _ =>
        {
            await Task.Delay(1000); // Simulate API call
            return new LLMResponse($"LLM response to: {input.Text}");
        });
    }

    private static Client<RT, UserInput, LLMResponse, A> clift<A>(Eff<RT, A> eff)
    {
        return Client.lift<RT, UserInput, LLMResponse, A>(eff);
    }

    private static Server<RT, UserInput, LLMResponse, A> slift<A>(Aff<RT, A> eff)
    {
        return Server.lift<RT, UserInput, LLMResponse, A>(eff);
    }

    static async IAsyncEnumerable<string> StreamSimulatedResponse(string input)
    {
        string[] responses =
        {
            $"Processing your input: {input}",
            "Analyzing...",
            "Generating response...",
            $"Final response to: {input}"
        };

        foreach (var response in responses)
        {
            await Task.Delay(500); // Simulate processing time
            yield return response;
        }
    }
}

[louthy]

1. I can't get streaming of the chat response back to work (there's a some commented out code in v4.4.9 that seems to promise to do what I want but hasn't been implemented yet).

Could you expand on that a little more? I'm not sure what you mean or what code is commented out?

2. I don't know what to do once I have an ongoing state that I need to send around (i.e. "conversationHistory" and the "machine"). Do I put that into the runtime or do I return these things in place of "Unit"?

v4 pipes is limited to lifting an Aff type into the Proxy whereas v5 pipes are genuine monad-transformers (like the Haskell original), meaning you can build a full monad transformer stack where you lift StateT or ReaderT into the Proxy. Outside of moving to v5 (which I wouldn't advise if you're working on production code) there aren't many options. You could, of course, use the runtime and add some helpers to create local states. Or, you could use Atom which is a mutable value with atomic updates. You could create an Atom per request, per client, per server, or whatever makes sense. You can also add an Atom value to the runtime if it makes sense - allowing for stateful values (rather than environmental).

It's a little bit of a hack I suppose, so not perfect, but without the full monad transformer capability you are a bit limited.

[ThereIsNoPie]

Could you expand on that a little more? I'm not sure what you mean or what code is commented out?

In the v4-latest branch there's this code in LanguageExt.Pipes.Client. There's something similar in Server.cs

        /*
        [Pure, MethodImpl(Proxy.mops)]
        internal static Client<RT, REQ, RES, X> enumerate<RT, REQ, RES, X>(EnumerateData<X> xs)
            where RT : struct, HasCancel<RT> =>
            new Enumerate2<RT, REQ, RES, Unit, Void, X, X>(xs, Pure<RT, REQ, RES, X>).ToClient();

        // 3 other overloads for enumerate here
            */

Outside of moving to v5 (which I wouldn't advise if you're working on production code) there aren't many options

Which of the hacky solutions do you think would be best? I'd be keen to just default to your suggestion since I haven't used any of these options before.

use the runtime and add some helpers to create local states

I'm looking at this approach but I'm unsure if I'm on the right track. Is this roughly what you're suggesting?

    public interface ConversationIO
    {
        Conversation Get();

        Unit AddMessage(string message);
    }

    public interface HasLLM<RT>
        where RT : struct
    {
        Eff<RT, ILLM> LlmEff { get; }
    }

    public interface HasConversation<RT>
        where RT : struct
    {
        Eff<RT, ConversationIO> ConversationEff { get; }
    }

    public readonly struct ExampleRuntime
        : HasCancel<ExampleRuntime>,
            HasConsole<ExampleRuntime>,
            HasLLM<ExampleRuntime>,
            HasConversation<ExampleRuntime>
    {
        private readonly RuntimeEnv env;

        public RuntimeEnv Env =>
            env
            ?? throw new InvalidOperationException(
                "Runtime Env not set.  Perhaps because of using default(Runtime) or new Runtime() rather than Runtime.New()"
            );

        public ExampleRuntime LocalCancel =>
            new ExampleRuntime(new RuntimeEnv(new CancellationTokenSource(), Env.Encoding));

        public CancellationToken CancellationToken => Env.Token;

        public CancellationTokenSource CancellationTokenSource => Env.Source;

        private ExampleRuntime(RuntimeEnv env)
        {
            this.env = env;
        }

        public Eff<ExampleRuntime, LanguageExt.Sys.Traits.ConsoleIO> ConsoleEff =>
            SuccessEff(LanguageExt.Sys.Live.ConsoleIO.Default);

        public Eff<ExampleRuntime, ILLM> LlmEff => SuccessEff(GPTLLM.Default);

        public Eff<ExampleRuntime, ConversationIO> ConversationEff =>
            SuccessEff(ConversationManager.Default);
    }