LangChain回调函数多次触发求助：如何让回调仅执行一次

你遇到的回调多次触发问题，是因为链中用到的字典结构{"app": itemgetter(...), ...}本质是RunnableMap，LangChain会为其中每个子Runnable单独触发on_chain_start事件，所以才会针对每个输入字段执行一次回调。

之前添加的has_run标志位仅首次有效，是因为GA4CallbackHandler实例会被复用，标志位设为True后没有在每次运行结束时重置，导致后续请求无法触发。

要实现回调仅在整个链运行时触发一次，可以通过判断事件对应的是根链而非内部子Runnable来实现，以下是两种可靠方式：

方式一：通过链的run_name过滤根链事件 #

利用你给整个链设置的run_name="feedback"，在回调中只处理根链的on_chain_start事件，忽略内部子Runnable的事件：

class GA4CallbackHandler(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, **kwargs):
        # 仅当触发事件的是根链（run_name为feedback）时执行逻辑
        if serialized.get("run_name") == "feedback":
            # 从inputs中获取真实请求数据，不要硬编码值
            response = track_feedback(
                "123456.7654321",
                inputs["app"],
                inputs["tool"],
                inputs["asset"],
                "standard",
                "user",
                str(inputs["score"])  # 转换为字符串匹配参数要求
            )
            print(response.status_code)
            print("feedback_sent")

链的配置保持不变即可：

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)

方式二：重置标志位适配实例复用 #

如果坚持用标志位的方式，需要在链运行结束或出错时重置标志位，确保每次请求都能正常触发：

class GA4CallbackHandler(BaseCallbackHandler):
    def __init__(self):
        self.has_run = False

    def on_chain_start(self, serialized, inputs, **kwargs):
        # 仅处理根链事件且未触发过的情况
        if not self.has_run and serialized.get("run_name") == "feedback":
            response = track_feedback(
                "123456.7654321",
                inputs["app"],
                inputs["tool"],
                inputs["asset"],
                "standard",
                "user",
                str(inputs["score"])
            )
            print(response.status_code)
            print("feedback_sent")
            self.has_run = True

    def on_chain_end(self, outputs, **kwargs):
        # 运行结束后重置标志位
        self.has_run = False

    def on_chain_error(self, error, **kwargs):
        # 出错时也要重置，避免影响下一次请求
        self.has_run = False

• 不要在回调中硬编码'app'这类固定值，必须从inputs参数中获取真实请求数据，否则无法正确跟踪每次运行的实际信息。
• 如果后续调整链结构，只要保证根链的run_name唯一，就能稳定过滤触发事件。

内容的提问来源于stack exchange，提问作者Carles Roch Arnau