Accessing Intermediate Steps in Chainlt LangChain Callback Handler

[bendoan-db]

Hello!

I'm currently working with a Chainlit and Langchain implementation, in which I have a series of generation steps and parsers. A basic example of the implementation can be found below:

def setup_lc_runnable():
    runnable = runnable_step_1() | StringOutputParser() | runnable_step_2 () | StringOutputParser() ....
    cl.user_session.set("runnable", full_chain)

async def on_chat_start():
    setup_lc_runnable()

@cl.on_message
async def on_message(message: cl.Message):
    memory = cl.user_session.get("memory")
    runnable = cl.user_session.get("runnable")  # type: Runnable

    res = cl.Message(content="")
    
    stream = runnable.astream(
            {"messages": [{"role": "user", "content": message.content}]},
            config={
                "configurable": {
                    "user_id":memory["user_id"],
                    "conversation_id":memory["conversation_id"]
                    }, 
                "callbacks": [cl.LangchainCallbackHandler(stream_final_answer=True)]
                }
    )
    
    async for chunk in stream:
        await res.stream_token(chunk)

This functionality works great! I can see the intermediate steps from runnable_step_1(), runnable_step_2(), etc. being executed in the UI as I run the chain.

However, I have an additional requirement to access the outputs from these intermediate steps and save them to an in-memory dictionary. What is the best way to implement this with the CustomDataLayer?

[bendoan-db]

hi @tpatel 😄 Any updates here?