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?
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.
Carles Roch Arnau is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.