Refa: add result to callback for agent tool use. (#9137)

### What problem does this PR solve?


### Type of change

- [x] Refactoring

Author: KevinHuSh

agent/canvas.py

@@ -252,15 +252,6 @@ def _node_finished(cpn_obj):
                            "created_at": cpn_obj.output("_created_time"),
                        })
 
-        def _append_path(cpn_id):
-            if self.path[-1] == cpn_id:
-                return
-            self.path.append(cpn_id)
-
-        def _extend_path(cpn_ids):
-            for cpn_id in cpn_ids:
-                _append_path(cpn_id)
-
         self.error = ""
         idx = len(self.path) - 1
         partials = []
@@ -279,10 +270,11 @@ def _extend_path(cpn_ids):
             # post processing of components invocation
             for i in range(idx, to):
                 cpn = self.get_component(self.path[i])
-                if cpn["obj"].component_name.lower() == "message":
-                    if isinstance(cpn["obj"].output("content"), partial):
+                cpn_obj = self.get_component_obj(self.path[i])
+                if cpn_obj.component_name.lower() == "message":
+                    if isinstance(cpn_obj.output("content"), partial):
                         _m = ""
-                        for m in cpn["obj"].output("content")():
+                        for m in cpn_obj.output("content")():
                             if not m:
                                 continue
                             if m == "<think>":
@@ -292,48 +284,65 @@ def _extend_path(cpn_ids):
                             else:
                                 yield decorate("message", {"content": m})
                                 _m += m
-                        cpn["obj"].set_output("content", _m)
+                        cpn_obj.set_output("content", _m)
                     else:
-                        yield decorate("message", {"content": cpn["obj"].output("content")})
+                        yield decorate("message", {"content": cpn_obj.output("content")})
                     yield decorate("message_end", {"reference": self.get_reference()})
 
                     while partials:
-                        _cpn = self.get_component(partials[0])
-                        if isinstance(_cpn["obj"].output("content"), partial):
+                        _cpn_obj = self.get_component_obj(partials[0])
+                        if isinstance(_cpn_obj.output("content"), partial):
                             break
-                        yield _node_finished(_cpn["obj"])
+                        yield _node_finished(_cpn_obj)
                         partials.pop(0)
 
-                if cpn["obj"].error():
-                    ex = cpn["obj"].exception_handler()
-                    if ex and ex["comment"]:
-                        yield decorate("message", {"content": ex["comment"]})
-                        yield decorate("message_end", {})
+                other_branch = False
+                if cpn_obj.error():
+                    ex = cpn_obj.exception_handler()
                     if ex and ex["goto"]:
-                        self.path.append(ex["goto"])
-                    elif not ex or not ex["default_value"]:
-                        self.error = cpn["obj"].error()
+                        self.path.extend(ex["goto"])
+                        other_branch = True
+                    elif ex and ex["default_value"]:
+                        yield decorate("message", {"content": ex["default_value"]})
+                        yield decorate("message_end", {})
+                    else:
+                        self.error = cpn_obj.error()
 
-                if cpn["obj"].component_name.lower() != "iteration":
-                    if isinstance(cpn["obj"].output("content"), partial):
+                if cpn_obj.component_name.lower() != "iteration":
+                    if isinstance(cpn_obj.output("content"), partial):
                         if self.error:
-                            cpn["obj"].set_output("content", None)
-                            yield _node_finished(cpn["obj"])
+                            cpn_obj.set_output("content", None)
+                            yield _node_finished(cpn_obj)
                         else:
                             partials.append(self.path[i])
                     else:
-                        yield _node_finished(cpn["obj"])
-
-                if cpn["obj"].component_name.lower() == "iterationitem" and cpn["obj"].end():
-                    iter = cpn["obj"].get_parent()
+                        yield _node_finished(cpn_obj)
+
+                def _append_path(cpn_id):
+                    nonlocal other_branch
+                    if other_branch:
+                        return
+                    if self.path[-1] == cpn_id:
+                        return
+                    self.path.append(cpn_id)
+
+                def _extend_path(cpn_ids):
+                    nonlocal other_branch
+                    if other_branch:
+                        return
+                    for cpn_id in cpn_ids:
+                        _append_path(cpn_id)
+
+                if cpn_obj.component_name.lower() == "iterationitem" and cpn_obj.end():
+                    iter = cpn_obj.get_parent()
                     yield _node_finished(iter)
                     _extend_path(self.get_component(cpn["parent_id"])["downstream"])
-                elif cpn["obj"].component_name.lower() in ["categorize", "switch"]:
-                    _extend_path(cpn["obj"].output("_next"))
-                elif cpn["obj"].component_name.lower() == "iteration":
-                    _append_path(cpn["obj"].get_start())
-                elif not cpn["downstream"] and cpn["obj"].get_parent():
-                    _append_path(cpn["obj"].get_parent().get_start())
+                elif cpn_obj.component_name.lower() in ["categorize", "switch"]:
+                    _extend_path(cpn_obj.output("_next"))
+                elif cpn_obj.component_name.lower() == "iteration":
+                    _append_path(cpn_obj.get_start())
+                elif not cpn["downstream"] and cpn_obj.get_parent():
+                    _append_path(cpn_obj.get_parent().get_start())
                 else:
                     _extend_path(cpn["downstream"])
 
@@ -342,13 +351,13 @@ def _extend_path(cpn_ids):
                 break
             idx = to
 
-            if any([self.get_component(c)["obj"].component_name.lower() == "userfillup" for c in self.path[idx:]]):
+            if any([self.get_component_obj(c).component_name.lower() == "userfillup" for c in self.path[idx:]]):
                 path = [c for c in self.path[idx:] if self.get_component(c)["obj"].component_name.lower() == "userfillup"]
                 path.extend([c for c in self.path[idx:] if self.get_component(c)["obj"].component_name.lower() != "userfillup"])
                 another_inputs = {}
                 tips = ""
                 for c in path:
-                    o = self.get_component(c)["obj"]
+                    o = self.get_component_obj(c)
                     if o.component_name.lower() == "userfillup":
                         another_inputs.update(o.get_input_elements())
                         if o.get_param("enable_tips"):

agent/component/agent_with_tools.py

@@ -157,7 +157,8 @@ def _invoke(self, **kwargs):
         prompt, msg = self._prepare_prompt_variables()
 
         downstreams = self._canvas.get_component(self._id)["downstream"] if self._canvas.get_component(self._id) else []
-        if any([self._canvas.get_component_obj(cid).component_name.lower()=="message" for cid in downstreams]) and not self._param.output_structure:
+        ex = self.exception_handler()
+        if any([self._canvas.get_component_obj(cid).component_name.lower()=="message" for cid in downstreams]) and not self._param.output_structure and not (ex and ex["goto"]):
             self.set_output("content", partial(self.stream_output_with_tools, prompt, msg))
             return
 
@@ -169,7 +170,10 @@ def _invoke(self, **kwargs):
 
         if ans.find("**ERROR**") >= 0:
             logging.error(f"Agent._chat got error. response: {ans}")
-            self.set_output("_ERROR", ans)
+            if self.get_exception_default_value():
+                self.set_output("content", self.get_exception_default_value())
+            else:
+                self.set_output("_ERROR", ans)
             return
 
         self.set_output("content", ans)
@@ -182,6 +186,12 @@ def stream_output_with_tools(self, prompt, msg):
         answer_without_toolcall = ""
         use_tools = []
         for delta_ans,_ in self._react_with_tools_streamly(msg, use_tools):
+            if delta_ans.find("**ERROR**") >= 0:
+                if self.get_exception_default_value():
+                    self.set_output("content", self.get_exception_default_value())
+                    yield self.get_exception_default_value()
+                else:
+                    self.set_output("_ERROR", delta_ans)
             answer_without_toolcall += delta_ans
             yield delta_ans
 
@@ -204,8 +214,8 @@ def _react_with_tools_streamly(self, history: list[dict], use_tools):
         hist = deepcopy(history)
         last_calling = ""
         if len(hist) > 3:
-            self.callback("Multi-turn conversation optimization", {}, " running ...")
             user_request = full_question(messages=history, chat_mdl=self.chat_mdl)
+            self.callback("Multi-turn conversation optimization", {}, user_request)
         else:
             user_request = history[-1]["content"]
 
@@ -241,9 +251,6 @@ def complete():
                     cited = True
             yield "", token_count
 
-            if not cited and need2cite:
-                self.callback("gen_citations", {}, " running ...")
-
             _hist = hist
             if len(hist) > 12:
                 _hist = [hist[0], hist[1], *hist[-10:]]
@@ -255,17 +262,21 @@ def complete():
             if not need2cite or cited:
                 return
 
+            txt = ""
             for delta_ans in self._gen_citations(entire_txt):
                 yield delta_ans, 0
+                txt += delta_ans
+
+            self.callback("gen_citations", {}, txt)
 
         def append_user_content(hist, content):
             if hist[-1]["role"] == "user":
                 hist[-1]["content"] += content
             else:
                 hist.append({"role": "user", "content": content})
 
-        self.callback("analyze_task", {}, " running ...")
         task_desc = analyze_task(self.chat_mdl, user_request, tool_metas)
+        self.callback("analyze_task", {}, task_desc)
         for _ in range(self._param.max_rounds + 1):
             response, tk = next_step(self.chat_mdl, hist, tool_metas, task_desc)
             # self.callback("next_step", {}, str(response)[:256]+"...")

agent/component/base.py

@@ -44,7 +44,6 @@ def __init__(self):
         self.delay_after_error = 2.0
         self.exception_method = None
         self.exception_default_value = None
-        self.exception_comment = None
         self.exception_goto = None
         self.debug_inputs = {}
 
@@ -97,6 +96,14 @@ def __str__(self):
     def as_dict(self):
         def _recursive_convert_obj_to_dict(obj):
             ret_dict = {}
+            if isinstance(obj, dict):
+                for k,v in obj.items():
+                    if isinstance(v, dict) or (v and type(v).__name__ not in dir(builtins)):
+                        ret_dict[k] = _recursive_convert_obj_to_dict(v)
+                    else:
+                        ret_dict[k] = v
+                return ret_dict
+
             for attr_name in list(obj.__dict__):
                 if attr_name in [_FEEDED_DEPRECATED_PARAMS, _DEPRECATED_PARAMS, _USER_FEEDED_PARAMS, _IS_RAW_CONF]:
                     continue
@@ -105,7 +112,7 @@ def _recursive_convert_obj_to_dict(obj):
                 if isinstance(attr, pd.DataFrame):
                     ret_dict[attr_name] = attr.to_dict()
                     continue
-                if attr and type(attr).__name__ not in dir(builtins):
+                if isinstance(attr, dict) or (attr and type(attr).__name__ not in dir(builtins)):
                     ret_dict[attr_name] = _recursive_convert_obj_to_dict(attr)
                 else:
                     ret_dict[attr_name] = attr
@@ -415,7 +422,10 @@ def invoke(self, **kwargs) -> dict[str, Any]:
         try:
             self._invoke(**kwargs)
         except Exception as e:
-            self._param.outputs["_ERROR"] = {"value": str(e)}
+            if self.get_exception_default_value():
+                self.set_exception_default_value()
+            else:
+                self.set_output("_ERROR", str(e))
             logging.exception(e)
         self._param.debug_inputs = {}
         self.set_output("_elapsed_time", time.perf_counter() - self.output("_created_time"))
@@ -427,7 +437,7 @@ def _invoke(self, **kwargs):
 
     def output(self, var_nm: str=None) -> Union[dict[str, Any], Any]:
         if var_nm:
-            return self._param.outputs.get(var_nm, {}).get("value")
+            return self._param.outputs.get(var_nm, {}).get("value", "")
         return {k: o.get("value") for k,o in self._param.outputs.items()}
 
     def set_output(self, key: str, value: Any):
@@ -520,7 +530,7 @@ def get_upstream(self) -> List[str]:
     def string_format(content: str, kv: dict[str, str]) -> str:
         for n, v in kv.items():
             content = re.sub(
-                r"\{%s\}" % re.escape(n), re.escape(v), content
+                r"\{%s\}" % re.escape(n), v, content
             )
         return content
 
@@ -529,13 +539,17 @@ def exception_handler(self):
             return
         return {
             "goto": self._param.exception_goto,
-            "comment": self._param.exception_comment,
             "default_value": self._param.exception_default_value
         }
 
     def get_exception_default_value(self):
+        if self._param.exception_method != "comment":
+            return ""
         return self._param.exception_default_value
 
+    def set_exception_default_value(self):
+        self.set_output("result", self.get_exception_default_value())
+
     @abstractmethod
     def thoughts(self) -> str:
         ...

agent/component/begin.py

@@ -46,4 +46,4 @@ def _invoke(self, **kwargs):
             self.set_input_value(k, v)
 
     def thoughts(self) -> str:
-        return "☕ Here we go..."
+        return ""