Tags for specific invoke of langchain

[RechaviaAmit]

Hi,

First of all, thank you for the amazing tool you’ve built!

Until recently, my usage of Langfuse looked like this:

def invoke1(langfuse_handler):
    langfuse_handler.tags = ["invoke_1"]
    chain.invoke({"input": "<user_input>"}, config={"callbacks": [langfuse_handler]})
    
def invoke1(langfuse_handler):
    langfuse_handler.tags = ["invoke_2"]
    chain.invoke({"input": "<user_input>"}, config={"callbacks": [langfuse_handler]})

def main()
    from langfuse.callback import CallbackHandler
    langfuse_handler = CallbackHandler(
        secret_key="my_secret_key",
        public_key="my_public_key",
        host="my_host"
    )
    
    invoke1(langfuse_handler )
    invoke1(langfuse_handler )

Lately, we’ve started using asynchronous code when invoking the LLM, like this:

async def invoke1(langfuse_handler):
    langfuse_handler.tags = ["invoke_1"]
    await chain.ainvoke({"input": "<user_input>"}, config={"callbacks": [langfuse_handler]})
    
async def invoke1(langfuse_handler):
    langfuse_handler.tags = ["invoke_2"]
    await chain.ainvoke({"input": "<user_input>"}, config={"callbacks": [langfuse_handler]})

async def main()
    from langfuse.callback import CallbackHandler
    langfuse_handler = CallbackHandler(
        secret_key="my_secret_key",
        public_key="my_public_key",
        host="my_host"
    )
    
    await invoke1(langfuse_handler )
    await invoke1(langfuse_handler )

As a result, I’ve noticed that in certain cases of a race condition, a query is uploaded to Langfuse with the wrong tags (e.g., the query from invoke1 has the tag invoke_2). It’s clear why this happens, but I’m looking for a way to ensure the tags are set only for a specific query.

What is the best way to achieve this?

[dosubot[bot]]

Shalom, @RechaviaAmit! I'm here to assist you with any bugs, questions, or contributions you have regarding Langfuse. Let's tackle this issue together!

To ensure that tags are set correctly for each specific query in asynchronous environments, you should avoid modifying shared state like langfuse_handler.tags directly. Instead, pass the tags as part of the configuration for each invocation. This way, each invocation has its own set of tags, and you avoid race conditions. Here's how you can modify your code:

async def invoke1(langfuse_handler):
    await chain.ainvoke(
        {"input": "<user_input>"},
        config={"callbacks": [langfuse_handler], "tags": ["invoke_1"]}
    )

async def invoke2(langfuse_handler):
    await chain.ainvoke(
        {"input": "<user_input>"},
        config={"callbacks": [langfuse_handler], "tags": ["invoke_2"]}
    )

async def main():
    from langfuse.callback import CallbackHandler
    langfuse_handler = CallbackHandler(
        secret_key="my_secret_key",
        public_key="my_public_key",
        host="my_host"
    )
    
    await invoke1(langfuse_handler)
    await invoke2(langfuse_handler)

By passing the tags directly in the config dictionary for each ainvoke call, you ensure that each invocation uses the correct tags without interfering with others ^[1].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[RechaviaAmit]

This works only if the chain is runnable?

[marcklingen]

Setting tags on the handler is not recommended if you want to use different tags dynamically for each run as you will otherwise run into concurrency issues.

You can set tags dynamically for each run via the langchain run config (docs):

from langfuse.callback import CallbackHandler
 
handler = CallbackHandler()
tags = ["tag-1", "tag-2"]
 
chain.invoke(
    {"animal": "dog"},
    config={
        "callbacks": [handler],
        "tags": tags,
    },
)

[RechaviaAmit]

Unfortunately, I can't because it's company information. I can give you information about specific packages if you want.
or we can speak in private

but a simple snippet will be:

from langfuse.callback import CallbackHandler
handler = CallbackHandler()

model = AzureChatOpenAI(...)
model.invoke(
        {"animal": "dog"},
        config={
        "callbacks": [handler],
        "tags": tags,
    },
)

[marcklingen]

thank you, tracking this here: #4869

[RechaviaAmit]

thanks!

[RechaviaAmit]

B.T.W -
once we do -

from langfuse.callback import CallbackHandler
handler = CallbackHandler()

model = AzureChatOpenAI(...)
model = model.with_structured_output(PydanticModel)
model.invoke(
        {"animal": "dog"},
        config={
        "callbacks": [handler],
        "tags": tags,
    },
)

it works :)
I think it's because now the model is runnable...

[marcklingen]

interesting, thanks for sharing