OiO.lk Community platform!

Oio.lk is an excellent forum for developers, providing a wide range of resources, discussions, and support for those in the developer community. Join oio.lk today to connect with like-minded professionals, share insights, and stay updated on the latest trends and technologies in the development field.
  You need to log in or register to access the solved answers to this problem.
  • You have reached the maximum number of guest views allowed
  • Please register below to remove this limitation

Callback attached to runnable runs multiple times

  • Thread starter Thread starter Carles Roch Arnau
  • Start date Start date
C

Carles Roch Arnau

Guest
I am coding a backend server in python using lancgain and I need to attach a callback to track each time the chain has been run using track_feedback(). I am using the "callbacks" argument on the with_config() in the self.runnable. But it is getting triggered many times per runnable execution. Could the children chains be somehow inheriting the call?

Code:
class BaseInputs(BaseModel):
    """Base inputs model."""
    app: str 
    run_id: str
    score: int
    tool: str
    asset: str

class GA4CallbackHandler(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        if not self.has_run:
            print(track_feedback("123456.7654321", 'app', 'tool', 'asset', 'standard', "user", "0").status_code)
            print("feedback_sent")

class FeedbackChain(Chain):
    
    def __init__(self) -> None:                
        self.runnable = ( 
            {
                "app" : itemgetter("app"),
                "run_id" : itemgetter("run_id"),
                "score" : itemgetter("score"),
                "tool": itemgetter("tool"),
                "asset": itemgetter("asset")                
            } 

            | RunnableLambda(self.send_feedback)
        ).with_config(run_name="feedback",callbacks=[GA4CallbackHandler()]).with_types(input_type=BaseInputs)

The reults is many triggers of the status code and "feedback_sent" string on console for each feedback sent. I printed the "serialized" parameter and it seems to be executing for each input (so once for "app", another for "run_id"...) How can I make this execute only once at the start?

I tried using a flag on the handler class but then it only executes once and subsequent executions never run. I also tried attaching the callback to the RunnableLambda inside the self.runnable chain, that worked but it looks dirty. I'd like it to be attached to the whole chain.
<p>I am coding a backend server in python using lancgain and I need to attach a callback to track each time the chain has been run using <code>track_feedback()</code>. I am using the "callbacks" argument on the <code>with_config()</code> in the <code>self.runnable</code>. But it is getting triggered many times per runnable execution. Could the children chains be somehow inheriting the call?</p>
<pre><code>class BaseInputs(BaseModel):
"""Base inputs model."""
app: str
run_id: str
score: int
tool: str
asset: str

class GA4CallbackHandler(BaseCallbackHandler):
def on_chain_start(self, serialized, inputs, **kwargs):
if not self.has_run:
print(track_feedback("123456.7654321", 'app', 'tool', 'asset', 'standard', "user", "0").status_code)
print("feedback_sent")

class FeedbackChain(Chain):

def __init__(self) -> None:
self.runnable = (
{
"app" : itemgetter("app"),
"run_id" : itemgetter("run_id"),
"score" : itemgetter("score"),
"tool": itemgetter("tool"),
"asset": itemgetter("asset")
}

| RunnableLambda(self.send_feedback)
).with_config(run_name="feedback",callbacks=[GA4CallbackHandler()]).with_types(input_type=BaseInputs)
</code></pre>
<p>The reults is many triggers of the status code and "feedback_sent" string on console for each feedback sent. I printed the "serialized" parameter and it seems to be executing for each input (so once for "app", another for "run_id"...)
How can I make this execute only once at the start?</p>
<p>I tried using a flag on the handler class but then it only executes once and subsequent executions never run.
I also tried attaching the callback to the RunnableLambda inside the self.runnable chain, that worked but it looks dirty. I'd like it to be attached to the whole chain.</p>
 

Latest posts

B
Replies
0
Views
1
Blundering Ecologist
B
Top