zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-07 20:10:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix: code supports matplotlib (#13724)

Browse Source 
### What problem does this PR solve?

Code as "final" node: 

![img_v3_02vs_aece4caf-8403-4939-9e68-9845a22c2cfg](https://github.com/user-attachments/assets/9d87b8df-da6b-401c-bf6d-8b807fe92c22)

Code as "mid" node:

![img_v3_02vv_f74f331f-d755-44ab-a18c-96fff8cbd34g](https://github.com/user-attachments/assets/c94ef3f9-2a6c-47cb-9d2b-19703d2752e4)


### Type of change

- [x] New Feature (non-breaking change which adds functionality)
This commit is contained in:
Yongteng Lei
2026-03-20 20:32:00 +08:00
committed by GitHub
parent 0507463f4e
commit dd839f30e8
20 changed files with 905 additions and 482 deletions
Download Patch File Download Diff File Expand all files Collapse all files

526
agent/component/agent_with_tools.py
View File

@@ -20,20 +20,20 @@ import os
import re
from copy import deepcopy
from functools import partial
from timeit import default_timer as timer
from typing import Any
import json_repair
from timeit import default_timer as timer
from agent.tools.base import LLMToolPluginCallSession, ToolParamBase, ToolBase, ToolMeta
from api.db.services.llm_service import LLMBundle
from api.db.services.tenant_llm_service import TenantLLMService
from api.db.services.mcp_server_service import MCPServerService
from agent.component.llm import LLM, LLMParam
from agent.tools.base import LLMToolPluginCallSession, ToolBase, ToolMeta, ToolParamBase
from api.db.joint_services.tenant_model_service import get_model_config_by_type_and_name
from api.db.services.llm_service import LLMBundle
from api.db.services.mcp_server_service import MCPServerService
from api.db.services.tenant_llm_service import TenantLLMService
from common.connection_utils import timeout
from rag.prompts.generator import next_step_async, COMPLETE_TASK, \
citation_prompt, kb_prompt, citation_plus, full_question, message_fit_in, structured_output_prompt
from common.mcp_tool_call_conn import MCPToolCallSession, mcp_tool_metadata_to_openai_tool
from agent.component.llm import LLMParam, LLM
from rag.prompts.generator import citation_plus, citation_prompt, full_question, kb_prompt, message_fit_in, structured_output_prompt
class AgentParam(LLMParam, ToolParamBase):
@@ -42,35 +42,25 @@ class AgentParam(LLMParam, ToolParamBase):
"""
def __init__(self):
self.meta:ToolMeta = {
"name": "agent",
"description": "This is an agent for a specific task.",
"parameters": {
"user_prompt": {
"type": "string",
"description": "This is the order you need to send to the agent.",
"default": "",
"required": True
},
"reasoning": {
"type": "string",
"description": (
"Supervisor's reasoning for choosing the this agent. "
"Explain why this agent is being invoked and what is expected of it."
),
"required": True
},
"context": {
"type": "string",
"description": (
"All relevant background information, prior facts, decisions, "
"and state needed by the agent to solve the current query. "
"Should be as detailed and self-contained as possible."
),
"required": True
},
}
}
self.meta: ToolMeta = {
"name": "agent",
"description": "This is an agent for a specific task.",
"parameters": {
"user_prompt": {"type": "string", "description": "This is the order you need to send to the agent.", "default": "", "required": True},
"reasoning": {
"type": "string",
"description": ("Supervisor's reasoning for choosing the this agent. Explain why this agent is being invoked and what is expected of it."),
"required": True,
},
"context": {
"type": "string",
"description": (
"All relevant background information, prior facts, decisions, and state needed by the agent to solve the current query. Should be as detailed and self-contained as possible."
),
"required": True,
},
},
}
super().__init__()
self.function_name = "agent"
self.tools = []
@@ -92,12 +82,14 @@ class Agent(LLM, ToolBase):
indexed_name = f"{original_name}_{idx}"
self.tools[indexed_name] = cpn
chat_model_config = get_model_config_by_type_and_name(self._canvas.get_tenant_id(), TenantLLMService.llm_id2llm_type(self._param.llm_id), self._param.llm_id)
self.chat_mdl = LLMBundle(self._canvas.get_tenant_id(), chat_model_config,
max_retries=self._param.max_retries,
retry_interval=self._param.delay_after_error,
max_rounds=self._param.max_rounds,
verbose_tool_use=True
)
self.chat_mdl = LLMBundle(
self._canvas.get_tenant_id(),
chat_model_config,
max_retries=self._param.max_retries,
retry_interval=self._param.delay_after_error,
max_rounds=self._param.max_rounds,
verbose_tool_use=False,
)
self.tool_meta = []
for indexed_name, tool_obj in self.tools.items():
original_meta = tool_obj.get_meta()
@@ -114,10 +106,30 @@ class Agent(LLM, ToolBase):
self.tools[tnm] = tool_call_session
self.callback = partial(self._canvas.tool_use_callback, id)
self.toolcall_session = LLMToolPluginCallSession(self.tools, self.callback)
#self.chat_mdl.bind_tools(self.toolcall_session, self.tool_metas)
if self.tool_meta:
self.chat_mdl.bind_tools(self.toolcall_session, self.tool_meta)
def _fit_messages(self, prompt: str, msg: list[dict]) -> list[dict]:
_, fitted_messages = message_fit_in(
[{"role": "system", "content": prompt}, *msg],
int(self.chat_mdl.max_length * 0.97),
)
return fitted_messages
@staticmethod
def _append_system_prompt(msg: list[dict], extra_prompt: str) -> None:
if extra_prompt and msg and msg[0]["role"] == "system":
msg[0]["content"] += "\n" + extra_prompt
@staticmethod
def _clean_formatted_answer(ans: str) -> str:
ans = re.sub(r"^.*</think>", "", ans, flags=re.DOTALL)
ans = re.sub(r"^.*```json", "", ans, flags=re.DOTALL)
return re.sub(r"```\n*$", "", ans, flags=re.DOTALL)
def _load_tool_obj(self, cpn: dict) -> object:
from agent.component import component_class
tool_name = cpn["component_name"]
param = component_class(tool_name + "Param")()
param.update(cpn["params"])
@@ -130,7 +142,7 @@ class Agent(LLM, ToolBase):
return component_class(cpn["component_name"])(self._canvas, cpn_id, param)
def get_meta(self) -> dict[str, Any]:
self._param.function_name= self._id.split("-->")[-1]
self._param.function_name = self._id.split("-->")[-1]
m = super().get_meta()
if hasattr(self._param, "user_prompt") and self._param.user_prompt:
m["function"]["parameters"]["properties"]["user_prompt"] = self._param.user_prompt
@@ -139,10 +151,7 @@ class Agent(LLM, ToolBase):
def get_input_form(self) -> dict[str, dict]:
res = {}
for k, v in self.get_input_elements().items():
res[k] = {
"type": "line",
"name": v["name"]
}
res[k] = {"type": "line", "name": v["name"]}
for cpn in self._param.tools:
if not isinstance(cpn, LLM):
continue
@@ -175,7 +184,7 @@ class Agent(LLM, ToolBase):
def _invoke(self, **kwargs):
return asyncio.run(self._invoke_async(**kwargs))
@timeout(int(os.environ.get("COMPONENT_EXEC_TIMEOUT", 20*60)))
@timeout(int(os.environ.get("COMPONENT_EXEC_TIMEOUT", 20 * 60)))
async def _invoke_async(self, **kwargs):
if self.check_if_canceled("Agent processing"):
return
@@ -204,19 +213,17 @@ class Agent(LLM, ToolBase):
schema = json.dumps(output_schema, ensure_ascii=False, indent=2)
schema_prompt = structured_output_prompt(schema)
downstreams = self._canvas.get_component(self._id)["downstream"] if self._canvas.get_component(self._id) else []
component = self._canvas.get_component(self._id)
downstreams = component["downstream"] if component else []
ex = self.exception_handler()
if any([self._canvas.get_component_obj(cid).component_name.lower()=="message" for cid in downstreams]) and not (ex and ex["goto"]) and not output_schema:
has_message_downstream = any(self._canvas.get_component_obj(cid).component_name.lower() == "message" for cid in downstreams)
if has_message_downstream and not (ex and ex["goto"]) and not output_schema:
self.set_output("content", partial(self.stream_output_with_tools_async, prompt, deepcopy(msg), user_defined_prompt))
return
_, msg = message_fit_in([{"role": "system", "content": prompt}, *msg], int(self.chat_mdl.max_length * 0.97))
use_tools = []
ans = ""
async for delta_ans, _tk in self._react_with_tools_streamly_async_simple(prompt, msg, use_tools, user_defined_prompt,schema_prompt=schema_prompt):
if self.check_if_canceled("Agent processing"):
return
ans += delta_ans
msg = self._fit_messages(prompt, msg)
self._append_system_prompt(msg, schema_prompt)
ans = await self._generate_async(msg)
if ans.find("**ERROR**") >= 0:
logging.error(f"Agent._chat got error. response: {ans}")
@@ -230,14 +237,8 @@ class Agent(LLM, ToolBase):
error = ""
for _ in range(self._param.max_retries + 1):
try:
def clean_formated_answer(ans: str) -> str:
ans = re.sub(r"^.*</think>", "", ans, flags=re.DOTALL)
ans = re.sub(r"^.*```json", "", ans, flags=re.DOTALL)
return re.sub(r"```\n*$", "", ans, flags=re.DOTALL)
obj = json_repair.loads(clean_formated_answer(ans))
obj = json_repair.loads(self._clean_formatted_answer(ans))
self.set_output("structured", obj)
if use_tools:
self.set_output("use_tools", use_tools)
return obj
except Exception:
error = "The answer cannot be parsed as JSON"
@@ -248,333 +249,118 @@ class Agent(LLM, ToolBase):
self.set_output("_ERROR", error)
return
attachment_content = self._collect_tool_attachment_content(existing_text=ans)
if attachment_content:
ans += "\n\n" + attachment_content
artifact_md = self._collect_tool_artifact_markdown(existing_text=ans)
if artifact_md:
ans += "\n\n" + artifact_md
self.set_output("content", ans)
if use_tools:
self.set_output("use_tools", use_tools)
return ans
async def stream_output_with_tools_async(self, prompt, msg, user_defined_prompt={}):
_, msg = message_fit_in([{"role": "system", "content": prompt}, *msg], int(self.chat_mdl.max_length * 0.97))
answer_without_toolcall = ""
use_tools = []
async for delta_ans, _ in self._react_with_tools_streamly_async_simple(prompt, msg, use_tools, user_defined_prompt):
if len(msg) > 3:
st = timer()
user_request = await full_question(messages=msg, chat_mdl=self.chat_mdl)
self.callback("Multi-turn conversation optimization", {}, user_request, elapsed_time=timer() - st)
msg = [*msg[:-1], {"role": "user", "content": user_request}]
msg = self._fit_messages(prompt, msg)
need2cite = self._param.cite and self._canvas.get_reference()["chunks"] and self._id.find("-->") < 0
cited = False
if need2cite and len(msg) < 7:
self._append_system_prompt(msg, citation_prompt())
cited = True
answer = ""
async for delta in self._generate_streamly(msg):
if self.check_if_canceled("Agent streaming"):
return
if delta_ans.find("**ERROR**") >= 0:
if delta.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)
return
answer_without_toolcall += delta_ans
yield delta_ans
self.set_output("content", answer_without_toolcall)
if use_tools:
self.set_output("use_tools", use_tools)
async def _react_with_tools_streamly_async_simple(self, prompt, history: list[dict], use_tools, user_defined_prompt={}, schema_prompt: str = ""):
token_count = 0
tool_metas = self.tool_meta
hist = deepcopy(history)
last_calling = ""
if len(hist) > 3:
st = timer()
user_request = await full_question(messages=history, chat_mdl=self.chat_mdl)
self.callback("Multi-turn conversation optimization", {}, user_request, elapsed_time=timer()-st)
else:
user_request = history[-1]["content"]
def build_task_desc(prompt: str, user_request: str, user_defined_prompt: dict | None = None) -> str:
"""Build a minimal task_desc by concatenating prompt, query, and tool schemas."""
user_defined_prompt = user_defined_prompt or {}
task_desc = (
"### Agent Prompt\n"
f"{prompt}\n\n"
"### User Request\n"
f"{user_request}\n\n"
)
if user_defined_prompt:
udp_json = json.dumps(user_defined_prompt, ensure_ascii=False, indent=2)
task_desc += "\n### User Defined Prompts\n" + udp_json + "\n"
return task_desc
async def use_tool_async(name, args):
nonlocal hist, use_tools, last_calling
logging.info(f"{last_calling=} == {name=}")
last_calling = name
tool_response = await self.toolcall_session.tool_call_async(name, args)
use_tools.append({
"name": name,
"arguments": args,
"results": tool_response
})
return name, tool_response
async def complete():
nonlocal hist
need2cite = self._param.cite and self._canvas.get_reference()["chunks"] and self._id.find("-->") < 0
if schema_prompt:
need2cite = False
cited = False
if hist and hist[0]["role"] == "system":
if schema_prompt:
hist[0]["content"] += "\n" + schema_prompt
if need2cite and len(hist) < 7:
hist[0]["content"] += citation_prompt()
cited = True
yield "", token_count
_hist = hist
if len(hist) > 12:
_hist = [hist[0], hist[1], *hist[-10:]]
entire_txt = ""
async for delta_ans in self._generate_streamly(_hist):
if not need2cite or cited:
yield delta_ans, 0
entire_txt += delta_ans
if not need2cite or cited:
self.set_output("_ERROR", delta)
return
if not need2cite or cited:
yield delta
answer += delta
st = timer()
txt = ""
async for delta_ans in self._gen_citations_async(entire_txt):
if self.check_if_canceled("Agent streaming"):
return
yield delta_ans, 0
txt += delta_ans
self.callback("gen_citations", {}, txt, elapsed_time=timer()-st)
def build_observation(tool_call_res: list[tuple]) -> str:
"""
Build a Observation from tool call results.
No LLM involved.
"""
if not tool_call_res:
return ""
lines = ["Observation:"]
for name, result in tool_call_res:
lines.append(f"[{name} result]")
lines.append(str(result))
return "\n".join(lines)
def append_user_content(hist, content):
if hist[-1]["role"] == "user":
hist[-1]["content"] += content
else:
hist.append({"role": "user", "content": content})
if not need2cite or cited:
attachment_content = self._collect_tool_attachment_content(existing_text=answer)
if attachment_content:
yield "\n\n" + attachment_content
answer += "\n\n" + attachment_content
artifact_md = self._collect_tool_artifact_markdown(existing_text=answer)
if artifact_md:
yield "\n\n" + artifact_md
answer += "\n\n" + artifact_md
self.set_output("content", answer)
return
st = timer()
task_desc = build_task_desc(prompt, user_request, user_defined_prompt)
self.callback("analyze_task", {}, task_desc, elapsed_time=timer()-st)
for _ in range(self._param.max_rounds + 1):
cited_answer = ""
async for delta in self._gen_citations_async(answer):
if self.check_if_canceled("Agent streaming"):
return
response, tk = await next_step_async(self.chat_mdl, hist, tool_metas, task_desc, user_defined_prompt)
# self.callback("next_step", {}, str(response)[:256]+"...")
token_count += tk or 0
hist.append({"role": "assistant", "content": response})
try:
# Remove markdown code fences properly
cleaned_response = re.sub(r"^.*```json\s*", "", response, flags=re.DOTALL)
cleaned_response = re.sub(r"```\s*$", "", cleaned_response, flags=re.DOTALL)
functions = json_repair.loads(cleaned_response)
if not isinstance(functions, list):
raise TypeError(f"List should be returned, but `{functions}`")
for f in functions:
if not isinstance(f, dict):
raise TypeError(f"An object type should be returned, but `{f}`")
tool_tasks = []
for func in functions:
name = func["name"]
args = func["arguments"]
if name == COMPLETE_TASK:
append_user_content(hist, f"Respond with a formal answer. FORGET(DO NOT mention) about `{COMPLETE_TASK}`. The language for the response MUST be as the same as the first user request.\n")
async for txt, tkcnt in complete():
yield txt, tkcnt
return
tool_tasks.append(asyncio.create_task(use_tool_async(name, args)))
results = await asyncio.gather(*tool_tasks) if tool_tasks else []
st = timer()
reflection = build_observation(results)
append_user_content(hist, reflection)
self.callback("reflection", {}, str(reflection), elapsed_time=timer()-st)
except Exception as e:
logging.exception(msg=f"Wrong JSON argument format in LLM ReAct response: {e}")
e = f"\nTool call error, please correct the input parameter of response format and call it again.\n *** Exception ***\n{e}"
append_user_content(hist, str(e))
logging.warning( f"Exceed max rounds: {self._param.max_rounds}")
final_instruction = f"""
{user_request}
IMPORTANT: You have reached the conversation limit. Based on ALL the information and research you have gathered so far, please provide a DIRECT and COMPREHENSIVE final answer to the original request.
Instructions:
1. SYNTHESIZE all information collected during this conversation
2. Provide a COMPLETE response using existing data - do not suggest additional research
3. Structure your response as a FINAL DELIVERABLE, not a plan
4. If information is incomplete, state what you found and provide the best analysis possible with available data
5. DO NOT mention conversation limits or suggest further steps
6. Focus on delivering VALUE with the information already gathered
Respond immediately with your final comprehensive answer.
"""
if self.check_if_canceled("Agent final instruction"):
return
append_user_content(hist, final_instruction)
async for txt, tkcnt in complete():
yield txt, tkcnt
# async def _react_with_tools_streamly_async(self, prompt, history: list[dict], use_tools, user_defined_prompt={}, schema_prompt: str = ""):
# token_count = 0
# tool_metas = self.tool_meta
# hist = deepcopy(history)
# last_calling = ""
# if len(hist) > 3:
# st = timer()
# user_request = await full_question(messages=history, chat_mdl=self.chat_mdl)
# self.callback("Multi-turn conversation optimization", {}, user_request, elapsed_time=timer()-st)
# else:
# user_request = history[-1]["content"]
# async def use_tool_async(name, args):
# nonlocal hist, use_tools, last_calling
# logging.info(f"{last_calling=} == {name=}")
# last_calling = name
# tool_response = await self.toolcall_session.tool_call_async(name, args)
# use_tools.append({
# "name": name,
# "arguments": args,
# "results": tool_response
# })
# # self.callback("add_memory", {}, "...")
# #self.add_memory(hist[-2]["content"], hist[-1]["content"], name, args, str(tool_response), user_defined_prompt)
# return name, tool_response
# async def complete():
# nonlocal hist
# need2cite = self._param.cite and self._canvas.get_reference()["chunks"] and self._id.find("-->") < 0
# if schema_prompt:
# need2cite = False
# cited = False
# if hist and hist[0]["role"] == "system":
# if schema_prompt:
# hist[0]["content"] += "\n" + schema_prompt
# if need2cite and len(hist) < 7:
# hist[0]["content"] += citation_prompt()
# cited = True
# yield "", token_count
# _hist = hist
# if len(hist) > 12:
# _hist = [hist[0], hist[1], *hist[-10:]]
# entire_txt = ""
# async for delta_ans in self._generate_streamly(_hist):
# if not need2cite or cited:
# yield delta_ans, 0
# entire_txt += delta_ans
# if not need2cite or cited:
# return
# st = timer()
# txt = ""
# async for delta_ans in self._gen_citations_async(entire_txt):
# if self.check_if_canceled("Agent streaming"):
# return
# yield delta_ans, 0
# txt += delta_ans
# self.callback("gen_citations", {}, txt, elapsed_time=timer()-st)
# def append_user_content(hist, content):
# if hist[-1]["role"] == "user":
# hist[-1]["content"] += content
# else:
# hist.append({"role": "user", "content": content})
# st = timer()
# task_desc = await analyze_task_async(self.chat_mdl, prompt, user_request, tool_metas, user_defined_prompt)
# self.callback("analyze_task", {}, task_desc, elapsed_time=timer()-st)
# for _ in range(self._param.max_rounds + 1):
# if self.check_if_canceled("Agent streaming"):
# return
# response, tk = await next_step_async(self.chat_mdl, hist, tool_metas, task_desc, user_defined_prompt)
# # self.callback("next_step", {}, str(response)[:256]+"...")
# token_count += tk or 0
# hist.append({"role": "assistant", "content": response})
# try:
# functions = json_repair.loads(re.sub(r"```.*", "", response))
# if not isinstance(functions, list):
# raise TypeError(f"List should be returned, but `{functions}`")
# for f in functions:
# if not isinstance(f, dict):
# raise TypeError(f"An object type should be returned, but `{f}`")
# tool_tasks = []
# for func in functions:
# name = func["name"]
# args = func["arguments"]
# if name == COMPLETE_TASK:
# append_user_content(hist, f"Respond with a formal answer. FORGET(DO NOT mention) about `{COMPLETE_TASK}`. The language for the response MUST be as the same as the first user request.\n")
# async for txt, tkcnt in complete():
# yield txt, tkcnt
# return
# tool_tasks.append(asyncio.create_task(use_tool_async(name, args)))
# results = await asyncio.gather(*tool_tasks) if tool_tasks else []
# st = timer()
# reflection = await reflect_async(self.chat_mdl, hist, results, user_defined_prompt)
# append_user_content(hist, reflection)
# self.callback("reflection", {}, str(reflection), elapsed_time=timer()-st)
# except Exception as e:
# logging.exception(msg=f"Wrong JSON argument format in LLM ReAct response: {e}")
# e = f"\nTool call error, please correct the input parameter of response format and call it again.\n *** Exception ***\n{e}"
# append_user_content(hist, str(e))
# logging.warning( f"Exceed max rounds: {self._param.max_rounds}")
# final_instruction = f"""
# {user_request}
# IMPORTANT: You have reached the conversation limit. Based on ALL the information and research you have gathered so far, please provide a DIRECT and COMPREHENSIVE final answer to the original request.
# Instructions:
# 1. SYNTHESIZE all information collected during this conversation
# 2. Provide a COMPLETE response using existing data - do not suggest additional research
# 3. Structure your response as a FINAL DELIVERABLE, not a plan
# 4. If information is incomplete, state what you found and provide the best analysis possible with available data
# 5. DO NOT mention conversation limits or suggest further steps
# 6. Focus on delivering VALUE with the information already gathered
# Respond immediately with your final comprehensive answer.
# """
# if self.check_if_canceled("Agent final instruction"):
# return
# append_user_content(hist, final_instruction)
# async for txt, tkcnt in complete():
# yield txt, tkcnt
yield delta
cited_answer += delta
attachment_content = self._collect_tool_attachment_content(existing_text=cited_answer)
if attachment_content:
yield "\n\n" + attachment_content
cited_answer += "\n\n" + attachment_content
artifact_md = self._collect_tool_artifact_markdown(existing_text=cited_answer)
if artifact_md:
yield "\n\n" + artifact_md
cited_answer += "\n\n" + artifact_md
self.callback("gen_citations", {}, cited_answer, elapsed_time=timer() - st)
self.set_output("content", cited_answer)
async def _gen_citations_async(self, text):
retrievals = self._canvas.get_reference()
retrievals = {"chunks": list(retrievals["chunks"].values()), "doc_aggs": list(retrievals["doc_aggs"].values())}
formated_refer = kb_prompt(retrievals, self.chat_mdl.max_length, True)
async for delta_ans in self._generate_streamly([{"role": "system", "content": citation_plus("\n\n".join(formated_refer))},
{"role": "user", "content": text}
]):
async for delta_ans in self._generate_streamly([{"role": "system", "content": citation_plus("\n\n".join(formated_refer))}, {"role": "user", "content": text}]):
yield delta_ans
def _collect_tool_artifact_markdown(self, existing_text: str = "") -> str:
md_parts = []
for tool_obj in self.tools.values():
if not hasattr(tool_obj, "_param") or not hasattr(tool_obj._param, "outputs"):
continue
artifacts_meta = tool_obj._param.outputs.get("_ARTIFACTS", {})
artifacts = artifacts_meta.get("value") if isinstance(artifacts_meta, dict) else None
if not artifacts:
continue
for art in artifacts:
if not isinstance(art, dict):
continue
url = art.get("url", "")
if url and (f"![]({url})" in existing_text or f"![{art.get('name', '')}]({url})" in existing_text):
continue
if art.get("mime_type", "").startswith("image/"):
md_parts.append(f"![{art['name']}]({url})")
else:
md_parts.append(f"[Download {art['name']}]({url})")
return "\n\n".join(md_parts)
def _collect_tool_attachment_content(self, existing_text: str = "") -> str:
text_parts = []
for tool_obj in self.tools.values():
if not hasattr(tool_obj, "_param") or not hasattr(tool_obj._param, "outputs"):
continue
content_meta = tool_obj._param.outputs.get("_ATTACHMENT_CONTENT", {})
content = content_meta.get("value") if isinstance(content_meta, dict) else None
if not content or not isinstance(content, str):
continue
content = content.strip()
if not content or content in existing_text:
continue
text_parts.append(content)
return "\n\n".join(text_parts)
def reset(self, only_output=False):
"""
Reset all tools if they have a reset method. This avoids errors for tools like MCPToolCallSession.

8
agent/sandbox/README.md
View File

@@ -189,7 +189,11 @@ Currently, the following languages are officially supported:
### 🐍 Python
To add Python dependencies, simply edit the following file:
Pre-installed packages: `requests`, `numpy`, `pandas`, `matplotlib`.
> `matplotlib` uses the `Agg` (non-interactive) backend by default in the sandbox (`MPLBACKEND=Agg`). No display server is available, so always save figures to files (e.g. `fig.savefig("artifacts/chart.png")`) rather than calling `plt.show()`.
To add more dependencies, edit:
```bash
sandbox_base_image/python/requirements.txt
@@ -199,6 +203,8 @@ Add any additional packages you need, one per line (just like a normal pip requi
### 🟨 Node.js
Pre-installed packages: `axios`.
To add Node.js dependencies:
1. Navigate to the Node.js base image directory:

10
agent/sandbox/executor_manager/models/schemas.py
View File

@@ -21,6 +21,13 @@ from pydantic import BaseModel, Field, field_validator
from models.enums import ResourceLimitType, ResultStatus, RuntimeErrorType, SupportLanguage, UnauthorizedAccessType
class ArtifactItem(BaseModel):
name: str
mime_type: str
size: int
content_b64: str
class CodeExecutionResult(BaseModel):
status: ResultStatus
stdout: str
@@ -37,6 +44,9 @@ class CodeExecutionResult(BaseModel):
unauthorized_access_type: Optional[UnauthorizedAccessType] = None
runtime_error_type: Optional[RuntimeErrorType] = None
# File artifacts produced by code execution (images, PDFs, CSVs, etc.)
artifacts: list[ArtifactItem] = []
class CodeExecutionRequest(BaseModel):
code_b64: str = Field(..., description="Base64 encoded code string")

86
agent/sandbox/executor_manager/services/execution.py
View File

@@ -24,7 +24,7 @@ from core.config import TIMEOUT
from core.container import allocate_container_blocking, release_container
from core.logger import logger
from models.enums import ResourceLimitType, ResultStatus, RuntimeErrorType, SupportLanguage, UnauthorizedAccessType
from models.schemas import CodeExecutionRequest, CodeExecutionResult
from models.schemas import ArtifactItem, CodeExecutionRequest, CodeExecutionResult
from utils.common import async_run_command
@@ -59,8 +59,12 @@ async def execute_code(req: CodeExecutionRequest):
f.write("""import json
import os
import sys
os.makedirs(os.path.join(os.getcwd(), "artifacts"), exist_ok=True)
sys.path.insert(0, os.path.dirname(__file__))
from main import main
if __name__ == "__main__":
args = json.loads(sys.argv[1])
result = main(**args)
@@ -180,12 +184,14 @@ if (fs.existsSync(mainPath)) {
logger.info(f"{args_json=}")
if returncode == 0:
artifacts = await _collect_artifacts(container, task_id, workdir)
return CodeExecutionResult(
status=ResultStatus.SUCCESS,
stdout=str(stdout),
stderr=stderr,
exit_code=0,
time_used_ms=time_used_ms,
artifacts=artifacts,
)
elif returncode == 124:
return CodeExecutionResult(
@@ -229,6 +235,84 @@ if (fs.existsSync(mainPath)) {
await release_container(container, language)
ALLOWED_ARTIFACT_EXTENSIONS = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".svg": "image/svg+xml",
".pdf": "application/pdf",
".csv": "text/csv",
".json": "application/json",
".html": "text/html",
}
MAX_ARTIFACT_COUNT = 10
MAX_ARTIFACT_SIZE = 10 * 1024 * 1024 # 10MB per file
async def _collect_artifacts(container: str, task_id: str, host_workdir: str) -> list[ArtifactItem]:
artifacts_path = f"/workspace/{task_id}/artifacts"
# List files in the artifacts directory inside the container
returncode, stdout, _ = await async_run_command(
"docker", "exec", container, "find", artifacts_path,
"-maxdepth", "1", "-type", "f", timeout=5,
)
if returncode != 0 or not stdout.strip():
return []
raw_names = [line.split("/")[-1] for line in stdout.strip().splitlines() if line.strip()]
# Sanitize: reject names with path traversal or control characters
filenames = [n for n in raw_names if n and "/" not in n and "\\" not in n and ".." not in n and not n.startswith(".")]
if not filenames:
return []
items: list[ArtifactItem] = []
for fname in filenames[:MAX_ARTIFACT_COUNT]:
ext = os.path.splitext(fname)[1].lower()
mime_type = ALLOWED_ARTIFACT_EXTENSIONS.get(ext)
if not mime_type:
logger.warning(f"Skipping artifact with disallowed extension: {fname}")
continue
file_path = f"{artifacts_path}/{fname}"
# Check file size inside the container
returncode, size_str, _ = await async_run_command(
"docker", "exec", container, "stat", "-c", "%s", file_path, timeout=5,
)
if returncode != 0:
logger.warning(f"Failed to stat artifact {fname}")
continue
file_size = int(size_str.strip())
if file_size > MAX_ARTIFACT_SIZE:
logger.warning(f"Artifact {fname} too large ({file_size} bytes), skipping")
continue
if file_size == 0:
continue
# Read file content via docker exec (docker cp doesn't work with gVisor tmpfs)
returncode, content_b64, stderr = await async_run_command(
"docker", "exec", container, "base64", file_path, timeout=30,
)
if returncode != 0:
logger.warning(f"Failed to read artifact {fname}: {stderr}")
continue
content_b64 = content_b64.replace("\n", "").strip()
items.append(ArtifactItem(
name=fname,
mime_type=mime_type,
size=file_size,
content_b64=content_b64,
))
logger.info(f"Collected artifact: {fname} ({file_size} bytes, {mime_type})")
return items
def analyze_error_result(stderr: str, exit_code: int) -> CodeExecutionResult:
"""Analyze the error result and classify it"""
if "Permission denied" in stderr:

1
agent/sandbox/providers/self_managed.py
View File

@@ -199,6 +199,7 @@ class SelfManagedProvider(SandboxProvider):
"memory_used_kb": result.get("memory_used_kb"),
"detail": result.get("detail"),
"instance_id": instance_id,
"artifacts": result.get("artifacts", []),
}
)

3
agent/sandbox/sandbox_base_image/python/Dockerfile
View File

@@ -2,12 +2,15 @@ FROM python:3.11-slim-bookworm
COPY --from=ghcr.io/astral-sh/uv:0.7.5 /uv /uvx /bin/
ENV UV_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple
ENV MPLBACKEND=Agg
ENV MPLCONFIGDIR=/tmp/matplotlib
COPY requirements.txt .
RUN grep -rl 'deb.debian.org' /etc/apt/ | xargs sed -i 's|http[s]*://deb.debian.org|https://mirrors.tuna.tsinghua.edu.cn|g' && \
apt-get update && \
apt-get install -y curl gcc && \
mkdir -p /tmp/matplotlib && \
uv pip install --system -r requirements.txt
WORKDIR /workspace

1
agent/sandbox/sandbox_base_image/python/requirements.txt
View File

@@ -1,3 +1,4 @@
numpy
pandas
matplotlib
requests

21
agent/tools/base.py
View File

@@ -57,17 +57,19 @@ class LLMToolPluginCallSession(ToolCallSession):
async def tool_call_async(self, name: str, arguments: dict[str, Any]) -> Any:
assert name in self.tools_map, f"LLM tool {name} does not exist"
logging.info(f"[ToolCall] invoke name={name} arguments={str(arguments)[:200]}")
st = timer()
tool_obj = self.tools_map[name]
if isinstance(tool_obj, MCPToolCallSession):
resp = await thread_pool_exec(tool_obj.tool_call, name, arguments, 60)
elif hasattr(tool_obj, "invoke_async") and asyncio.iscoroutinefunction(tool_obj.invoke_async):
resp = await tool_obj.invoke_async(**arguments)
else:
if hasattr(tool_obj, "invoke_async") and asyncio.iscoroutinefunction(tool_obj.invoke_async):
resp = await tool_obj.invoke_async(**arguments)
else:
resp = await thread_pool_exec(tool_obj.invoke, **arguments)
resp = await thread_pool_exec(tool_obj.invoke, **arguments)
self.callback(name, arguments, resp, elapsed_time=timer()-st)
elapsed = timer() - st
logging.info(f"[ToolCall] done name={name} elapsed={elapsed:.2f}s result={str(resp)[:200]}")
self.callback(name, arguments, resp, elapsed_time=elapsed)
return resp
def get_tool_obj(self, name):
@@ -101,13 +103,8 @@ class ToolParamBase(ComponentParamBase):
if "enum" in p:
params[k]["enum"] = p["enum"]
desc = self.meta["description"]
if hasattr(self, "description"):
desc = self.description
function_name = self.meta["name"]
if hasattr(self, "function_name"):
function_name = self.function_name
desc = getattr(self, "description", None) or self.meta["description"]
function_name = getattr(self, "function_name", self.meta["name"])
return {
"type": "function",

233
agent/tools/code_exec.py
View File

@@ -18,6 +18,7 @@ import base64
import json
import logging
import os
import uuid
from abc import ABC
from typing import Optional
@@ -25,8 +26,10 @@ from pydantic import BaseModel, Field, field_validator
from strenum import StrEnum
from agent.tools.base import ToolBase, ToolMeta, ToolParamBase
from api.db.services.file_service import FileService
from common import settings
from common.connection_utils import timeout
from common.constants import SANDBOX_ARTIFACT_BUCKET, SANDBOX_ARTIFACT_EXPIRE_DAYS
class Language(StrEnum):
@@ -70,6 +73,7 @@ class CodeExecParam(ToolParamBase):
"name": "execute_code",
"description": """
This tool has a sandbox that can execute code written in 'Python'/'Javascript'. It receives a piece of code and return a Json string.
Here's a code example for Python(`main` function MUST be included):
def main() -> dict:
\"\"\"
@@ -84,6 +88,26 @@ def main() -> dict:
"result": fibonacci_recursive(100),
}
To generate charts or files (images, PDFs, CSVs, etc.), save them to the `artifacts/` directory (relative to the working directory). The sandbox will automatically collect these files and return them. Example:
def main() -> dict:
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import pandas as pd
df = pd.DataFrame({"x": [1, 2, 3, 4], "y": [10, 20, 25, 30]})
fig, ax = plt.subplots()
ax.plot(df["x"], df["y"])
ax.set_title("Sample Chart")
fig.savefig("artifacts/chart.png", dpi=150, bbox_inches="tight")
plt.close(fig)
return {"summary": "Chart saved to artifacts/chart.png"}
Available Python packages: pandas, numpy, matplotlib, requests.
Supported artifact file types: .png, .jpg, .jpeg, .svg, .pdf, .csv, .json, .html
Collected artifacts are also parsed automatically and appended to the stable text output `content`. The content includes sections like `attachment1 (image): ...`, `attachment2 (pdf): ...`, so downstream nodes can consume a single text output without depending on unstable attachment-specific variables.
Here's a code example for Javascript(`main` function MUST be included and exported):
const axios = require('axios');
async function main(args) {
@@ -125,6 +149,7 @@ module.exports = { main };
class CodeExec(ToolBase, ABC):
component_name = "CodeExec"
_lifecycle_configured = False
@timeout(int(os.environ.get("COMPONENT_EXEC_TIMEOUT", 10 * 60)))
def _invoke(self, **kwargs):
@@ -148,6 +173,8 @@ class CodeExec(ToolBase, ABC):
if self.check_if_canceled("CodeExec execution"):
return self.output()
timeout_seconds = int(os.environ.get("COMPONENT_EXEC_TIMEOUT", 10 * 60))
try:
# Try using the new sandbox provider system first
try:
@@ -157,25 +184,13 @@ class CodeExec(ToolBase, ABC):
return
# Execute code using the provider system
result = sandbox_execute_code(
code=code,
language=language,
timeout=int(os.environ.get("COMPONENT_EXEC_TIMEOUT", 10 * 60)),
arguments=arguments
)
result = sandbox_execute_code(code=code, language=language, timeout=timeout_seconds, arguments=arguments)
if self.check_if_canceled("CodeExec execution"):
return
# Process the result
if result.stderr:
self.set_output("_ERROR", result.stderr)
return
parsed_stdout = self._deserialize_stdout(result.stdout)
logging.info(f"[CodeExec]: Provider system -> {parsed_stdout}")
self._populate_outputs(parsed_stdout, result.stdout)
return
artifacts = result.metadata.get("artifacts", []) if result.metadata else []
return self._process_execution_result(result.stdout, result.stderr, "Provider system", artifacts)
except (ImportError, RuntimeError) as provider_error:
# Provider system not available or not configured, fall back to HTTP
@@ -196,7 +211,7 @@ class CodeExec(ToolBase, ABC):
self.set_output("_ERROR", "Task has been canceled")
return self.output()
resp = requests.post(url=f"http://{settings.SANDBOX_HOST}:9385/run", json=code_req, timeout=int(os.environ.get("COMPONENT_EXEC_TIMEOUT", 10 * 60)))
resp = requests.post(url=f"http://{settings.SANDBOX_HOST}:9385/run", json=code_req, timeout=timeout_seconds)
logging.info(f"http://{settings.SANDBOX_HOST}:9385/run, code_req: {code_req}, resp.status_code {resp.status_code}:")
if self.check_if_canceled("CodeExec execution"):
@@ -206,14 +221,12 @@ class CodeExec(ToolBase, ABC):
resp.raise_for_status()
body = resp.json()
if body:
stderr = body.get("stderr")
if stderr:
self.set_output("_ERROR", stderr)
return self.output()
raw_stdout = body.get("stdout", "")
parsed_stdout = self._deserialize_stdout(raw_stdout)
logging.info(f"[CodeExec]: http://{settings.SANDBOX_HOST}:9385/run -> {parsed_stdout}")
self._populate_outputs(parsed_stdout, raw_stdout)
return self._process_execution_result(
body.get("stdout", ""),
body.get("stderr"),
f"http://{settings.SANDBOX_HOST}:9385/run",
body.get("artifacts", []),
)
else:
self.set_output("_ERROR", "There is no response from sandbox")
return self.output()
@@ -226,6 +239,100 @@ class CodeExec(ToolBase, ABC):
return self.output()
def _process_execution_result(self, stdout: str, stderr: str | None, source: str, artifacts: list | None = None):
if stderr and not stdout and not artifacts:
self.set_output("_ERROR", stderr)
return self.output()
# Clear any stale error from previous runs or base class initialization
self.set_output("_ERROR", "")
if stderr:
logging.warning(f"[CodeExec]: stderr (non-fatal): {stderr[:500]}")
parsed_stdout = self._deserialize_stdout(stdout)
logging.info(f"[CodeExec]: {source} -> {parsed_stdout}")
self._populate_outputs(parsed_stdout, stdout)
content_parts = []
base_content = self._build_content_text(parsed_stdout, raw_stdout=stdout)
if base_content:
content_parts.append(base_content)
if artifacts:
artifact_urls = self._upload_artifacts(artifacts)
if artifact_urls:
self.set_output("_ARTIFACTS", artifact_urls)
attachment_text = self._build_attachment_content(artifacts, artifact_urls)
self.set_output("_ATTACHMENT_CONTENT", attachment_text)
if attachment_text:
content_parts.append(attachment_text)
else:
self.set_output("_ATTACHMENT_CONTENT", "")
self.set_output("content", "\n\n".join([part for part in content_parts if part]).strip())
return self.output()
@classmethod
def _ensure_bucket_lifecycle(cls):
if cls._lifecycle_configured:
return
try:
storage = settings.STORAGE_IMPL
# Only MinIO/S3 backends expose .conn for lifecycle config
if not hasattr(storage, "conn") or storage.conn is None:
cls._lifecycle_configured = True
return
if not storage.conn.bucket_exists(SANDBOX_ARTIFACT_BUCKET):
storage.conn.make_bucket(SANDBOX_ARTIFACT_BUCKET)
from minio.commonconfig import Filter
from minio.lifecycleconfig import Expiration, LifecycleConfig, Rule
rule = Rule(
rule_id="auto-expire",
status="Enabled",
rule_filter=Filter(prefix=""),
expiration=Expiration(days=SANDBOX_ARTIFACT_EXPIRE_DAYS),
)
storage.conn.set_bucket_lifecycle(SANDBOX_ARTIFACT_BUCKET, LifecycleConfig([rule]))
logging.info(f"[CodeExec]: Set {SANDBOX_ARTIFACT_EXPIRE_DAYS}-day lifecycle on bucket '{SANDBOX_ARTIFACT_BUCKET}'")
cls._lifecycle_configured = True
except Exception as e:
# Do NOT set _lifecycle_configured so we retry next time
logging.warning(f"[CodeExec]: Failed to set bucket lifecycle: {e}")
def _upload_artifacts(self, artifacts: list) -> list[dict]:
self._ensure_bucket_lifecycle()
uploaded = []
for art in artifacts:
try:
name = art.get("name", "") if isinstance(art, dict) else getattr(art, "name", "")
content_b64 = art.get("content_b64", "") if isinstance(art, dict) else getattr(art, "content_b64", "")
mime_type = art.get("mime_type", "") if isinstance(art, dict) else getattr(art, "mime_type", "")
size = art.get("size", 0) if isinstance(art, dict) else getattr(art, "size", 0)
if not content_b64 or not name:
continue
ext = os.path.splitext(name)[1].lower()
storage_name = f"{uuid.uuid4().hex}{ext}"
binary = base64.b64decode(content_b64)
settings.STORAGE_IMPL.put(SANDBOX_ARTIFACT_BUCKET, storage_name, binary)
url = f"/v1/document/artifact/{storage_name}"
uploaded.append(
{
"name": name,
"url": url,
"mime_type": mime_type,
"size": size,
}
)
logging.info(f"[CodeExec]: Uploaded artifact {name} -> {url}")
except Exception as e:
logging.warning(f"[CodeExec]: Failed to upload artifact: {e}")
return uploaded
def _encode_code(self, code: str) -> str:
return base64.b64encode(code.encode("utf-8")).decode("utf-8")
@@ -357,6 +464,84 @@ class CodeExec(ToolBase, ABC):
logging.info(f"[CodeExec]: populate scalar key='{key}' raw='{val}' coerced='{coerced}'")
self.set_output(key, coerced)
def _build_attachment_content(self, artifacts: list, artifact_urls: list[dict] | None = None) -> str:
sections = []
artifact_urls = artifact_urls or []
for idx, art in enumerate(artifacts, start=1):
key = f"attachment{idx}"
try:
name = art.get("name", "") if isinstance(art, dict) else getattr(art, "name", "")
content_b64 = art.get("content_b64", "") if isinstance(art, dict) else getattr(art, "content_b64", "")
mime_type = art.get("mime_type", "") if isinstance(art, dict) else getattr(art, "mime_type", "")
if not name or not content_b64:
continue
blob = base64.b64decode(content_b64)
parsed = FileService.parse(
name,
blob,
False,
tenant_id=self._canvas.get_tenant_id(),
)
attachment_type = self._normalize_attachment_type(name, mime_type)
section = self._format_attachment_section(key, attachment_type, name, parsed)
sections.append(section)
logging.info(f"[CodeExec]: parse attachment section key='{key}' from artifact='{name}'")
except Exception as e:
logging.warning(f"[CodeExec]: Failed to parse artifact for content section '{key}': {e}")
fallback_type = self._normalize_attachment_type(
art.get("name", "") if isinstance(art, dict) else getattr(art, "name", ""),
art.get("mime_type", "") if isinstance(art, dict) else getattr(art, "mime_type", ""),
)
fallback_name = art.get("name", "") if isinstance(art, dict) else getattr(art, "name", "")
fallback_url = ""
if idx - 1 < len(artifact_urls):
fallback_url = artifact_urls[idx - 1].get("url", "")
fallback_text = "Artifact generated but parse failed."
if fallback_url:
fallback_text += f" Download: {fallback_url}"
sections.append(self._format_attachment_section(key, fallback_type, fallback_name, fallback_text))
if sections:
return f"attachment_count: {len(sections)}\n\n" + "\n\n".join(sections)
return "attachment_count: 0"
def _normalize_attachment_type(self, name: str, mime_type: str) -> str:
mime_type = str(mime_type or "").strip().lower()
if mime_type.startswith("image/"):
return "image"
if mime_type == "application/pdf":
return "pdf"
if mime_type == "text/csv":
return "csv"
if mime_type == "application/json":
return "json"
if mime_type == "text/html":
return "html"
ext = os.path.splitext(name or "")[1].lower().lstrip(".")
return ext or "file"
def _format_attachment_section(self, key: str, attachment_type: str, name: str, parsed: str) -> str:
title = f"{key} ({attachment_type})"
if name:
title += f": {name}"
body = parsed if isinstance(parsed, str) else json.dumps(parsed, ensure_ascii=False)
return f"{title}\n{body}".strip()
def _build_content_text(self, parsed_stdout, raw_stdout: str = "") -> str:
if isinstance(parsed_stdout, str):
return parsed_stdout.strip()
if isinstance(parsed_stdout, (dict, list, tuple)):
try:
return json.dumps(parsed_stdout, ensure_ascii=False, indent=2).strip()
except Exception:
return str(parsed_stdout).strip()
if parsed_stdout is None:
return str(raw_stdout or "").strip()
return str(parsed_stdout).strip()
def _get_by_path(self, data, path: str):
if not path:
return None

67
api/apps/document_app.py
View File

@@ -18,36 +18,38 @@ import os.path
import pathlib
import re
from pathlib import Path, PurePosixPath, PureWindowsPath
from quart import request, make_response
from quart import make_response, request
from api.apps import current_user, login_required
from api.common.check_team_permission import check_kb_team_permission
from api.constants import FILE_NAME_LEN_LIMIT, IMG_BASE64_PREFIX
from api.db import VALID_FILE_TYPES, FileType
from api.db.db_models import Task
from api.db.services import duplicate_name
from api.db.services.document_service import DocumentService, doc_upload_and_parse
from api.db.services.doc_metadata_service import DocMetadataService
from common.metadata_utils import meta_filter, convert_conditions, turn2jsonschema
from api.db.services.document_service import DocumentService, doc_upload_and_parse
from api.db.services.file2document_service import File2DocumentService
from api.db.services.file_service import FileService
from api.db.services.knowledgebase_service import KnowledgebaseService
from api.db.services.task_service import TaskService, cancel_all_task_of
from api.db.services.user_service import UserTenantService
from common.misc_utils import get_uuid, thread_pool_exec
from api.utils.api_utils import (
get_data_error_result,
get_json_result,
get_request_json,
server_error_response,
validate_request,
get_request_json,
)
from api.utils.file_utils import filename_type, thumbnail
from common.file_utils import get_project_base_directory
from common.constants import RetCode, VALID_TASK_STATUS, ParserType, TaskStatus
from api.utils.web_utils import CONTENT_TYPE_MAP, apply_safe_file_response_headers, html2pdf, is_valid_url
from deepdoc.parser.html_parser import RAGFlowHtmlParser
from rag.nlp import search, rag_tokenizer
from common import settings
from common.constants import SANDBOX_ARTIFACT_BUCKET, VALID_TASK_STATUS, ParserType, RetCode, TaskStatus
from common.file_utils import get_project_base_directory
from common.metadata_utils import convert_conditions, meta_filter, turn2jsonschema
from common.misc_utils import get_uuid, thread_pool_exec
from deepdoc.parser.html_parser import RAGFlowHtmlParser
from rag.nlp import rag_tokenizer, search
def _is_safe_download_filename(name: str) -> bool:
@@ -75,6 +77,7 @@ async def upload():
return get_json_result(data=False, message="No file part!", code=RetCode.ARGUMENT_ERROR)
file_objs = files.getlist("file")
def _close_file_objs(objs):
for obj in objs:
try:
@@ -84,6 +87,7 @@ async def upload():
obj.stream.close()
except Exception:
pass
for file_obj in file_objs:
if file_obj.filename == "":
_close_file_objs(file_objs)
@@ -239,7 +243,6 @@ async def list_docs():
kb_id = request.args.get("kb_id")
if not kb_id:
return get_json_result(data=False, message='Lack of "KB ID"', code=RetCode.ARGUMENT_ERROR)
tenants = UserTenantService.query(user_id=current_user.id)
for tenant in tenants:
if KnowledgebaseService.query(tenant_id=tenant.tenant_id, id=kb_id):
@@ -608,6 +611,7 @@ async def run():
req = await get_request_json()
uid = current_user.id
try:
def _run_sync():
for doc_id in req["doc_ids"]:
if not DocumentService.accessible(doc_id, uid):
@@ -670,6 +674,7 @@ async def rename():
req = await get_request_json()
uid = current_user.id
try:
def _rename_sync():
if not DocumentService.accessible(req["doc_id"], uid):
return get_json_result(data=False, message="No authorization.", code=RetCode.AUTHENTICATION_ERROR)
@@ -827,6 +832,44 @@ async def get_image(image_id):
return server_error_response(e)
ARTIFACT_CONTENT_TYPES = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".svg": "image/svg+xml",
".pdf": "application/pdf",
".csv": "text/csv",
".json": "application/json",
".html": "text/html",
}
@manager.route("/artifact/<filename>", methods=["GET"]) # noqa: F821
@login_required
async def get_artifact(filename):
try:
bucket = SANDBOX_ARTIFACT_BUCKET
# Validate filename: must be uuid hex + allowed extension, nothing else
basename = os.path.basename(filename)
if basename != filename or "/" in filename or "\\" in filename:
return get_data_error_result(message="Invalid filename.")
ext = os.path.splitext(basename)[1].lower()
if ext not in ARTIFACT_CONTENT_TYPES:
return get_data_error_result(message="Invalid file type.")
data = await thread_pool_exec(settings.STORAGE_IMPL.get, bucket, basename)
if not data:
return get_data_error_result(message="Artifact not found.")
content_type = ARTIFACT_CONTENT_TYPES.get(ext, "application/octet-stream")
response = await make_response(data)
safe_filename = re.sub(r"[^\w.\-]", "_", basename)
apply_safe_file_response_headers(response, content_type, ext)
if not response.headers.get("Content-Disposition"):
response.headers.set("Content-Disposition", f'inline; filename="{safe_filename}"')
return response
except Exception as e:
return server_error_response(e)
@manager.route("/upload_and_parse", methods=["POST"]) # noqa: F821
@login_required
@validate_request("conversation_id")
@@ -942,8 +985,8 @@ async def set_meta():
@manager.route("/upload_info", methods=["POST"]) # noqa: F821
async def upload_info():
files = await request.files
file = files['file'] if files and files.get("file") else None
file = files["file"] if files and files.get("file") else None
try:
return get_json_result(data=FileService.upload_info(current_user.id, file, request.args.get("url")))
except Exception as e:
return server_error_response(e)
return server_error_response(e)

9
api/db/services/llm_service.py
View File

@@ -405,7 +405,10 @@ class LLMBundle(LLM4Tenant):
async def async_chat_streamly(self, system: str, history: list, gen_conf: dict = {}, **kwargs):
total_tokens = 0
ans = ""
if self.is_tools and getattr(self.mdl, "is_tools", False) and hasattr(self.mdl, "async_chat_streamly_with_tools"):
_bundle_is_tools = self.is_tools
_mdl_is_tools = getattr(self.mdl, "is_tools", False)
_has_with_tools = hasattr(self.mdl, "async_chat_streamly_with_tools")
if _bundle_is_tools and _mdl_is_tools and _has_with_tools:
stream_fn = getattr(self.mdl, "async_chat_streamly_with_tools", None)
elif hasattr(self.mdl, "async_chat_streamly"):
stream_fn = getattr(self.mdl, "async_chat_streamly", None)
@@ -425,7 +428,7 @@ class LLMBundle(LLM4Tenant):
total_tokens = txt
break
if txt.endswith("</think>"):
if txt.endswith("</think>") and ans.endswith("</think>"):
ans = ans[: -len("</think>")]
if not self.verbose_tool_use:
@@ -468,7 +471,7 @@ class LLMBundle(LLM4Tenant):
total_tokens = txt
break
if txt.endswith("</think>"):
if txt.endswith("</think>") and ans.endswith("</think>"):
ans = ans[: -len("</think>")]
if not self.verbose_tool_use:

3
common/constants.py
View File

@@ -14,11 +14,14 @@
# limitations under the License.
#
import os
from enum import Enum, IntEnum
from strenum import StrEnum
SERVICE_CONF = "service_conf.yaml"
RAG_FLOW_SERVICE_NAME = "ragflow"
SANDBOX_ARTIFACT_BUCKET = os.environ.get("SANDBOX_ARTIFACT_BUCKET", "sandbox-artifacts")
SANDBOX_ARTIFACT_EXPIRE_DAYS = int(os.environ.get("SANDBOX_ARTIFACT_EXPIRE_DAYS", "7"))
class CustomEnum(Enum):

7
conf/llm_factories.json
View File

@@ -379,6 +379,13 @@
"rank": "950",
"url" : "https://dashscope.aliyuncs.com/compatible-mode/v1",
"llm": [
{
"llm_name": "qwen3.5-122b-a10b",
"tags": "LLM,CHAT,128k",
"max_tokens": 128000,
"model_type": "chat",
"is_tools": true
},
{
"llm_name": "Moonshot-Kimi-K2-Instruct",
"tags": "LLM,CHAT,128K",

4
docker/.env
View File

@@ -261,6 +261,10 @@ REGISTER_ENABLED=1
# SANDBOX_ENABLE_SECCOMP=false
# SANDBOX_MAX_MEMORY=256m # b, k, m, g
# SANDBOX_TIMEOUT=10s # s, m, 1m30s
# The MinIO bucket name for storing sandbox-generated artifacts (charts, files, etc.).
SANDBOX_ARTIFACT_BUCKET=sandbox-artifacts
# Number of days before sandbox artifacts are automatically deleted from storage.
SANDBOX_ARTIFACT_EXPIRE_DAYS=7
# Enable DocLing
USE_DOCLING=false

188
rag/llm/chat_model.py
View File

@@ -324,6 +324,34 @@ class Base(ABC):
hist.append({"role": "tool", "tool_call_id": tool_call.id, "content": str(tool_res)})
return hist
def _append_history_batch(self, hist, results):
"""
Append a batch of tool calls to history following the OpenAI protocol:
one assistant message containing all tool_calls, followed by one tool message per call.
results: list of (tool_call, name, args, result, error)
"""
hist.append({
"role": "assistant",
"tool_calls": [
{
"index": tc.index,
"id": tc.id,
"function": {"name": tc.function.name, "arguments": tc.function.arguments},
"type": "function",
}
for tc, _, _, _, _ in results
],
})
for tc, _, _, result, err in results:
if err:
content = str(err)
elif isinstance(result, dict):
content = json.dumps(result, ensure_ascii=False)
else:
content = str(result)
hist.append({"role": "tool", "tool_call_id": tc.id, "content": content})
return hist
def bind_tools(self, toolcall_session, tools):
if not (toolcall_session and tools):
return
@@ -360,18 +388,24 @@ class Base(ABC):
return ans, tk_count
for tool_call in response.choices[0].message.tool_calls:
logging.info(f"Response {tool_call=}")
name = tool_call.function.name
async def _exec_tool(tc):
name = tc.function.name
try:
args = json_repair.loads(tool_call.function.arguments)
tool_response = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
history = self._append_history(history, tool_call, tool_response)
ans += self._verbose_tool_use(name, args, tool_response)
args = json_repair.loads(tc.function.arguments)
if hasattr(self.toolcall_session, "tool_call_async"):
result = await self.toolcall_session.tool_call_async(name, args)
else:
result = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
return tc, name, args, result, None
except Exception as e:
logging.exception(msg=f"Wrong JSON argument format in LLM tool call response: {tool_call}")
history.append({"role": "tool", "tool_call_id": tool_call.id, "content": f"Tool call error: \n{tool_call}\nException:\n" + str(e)})
ans += self._verbose_tool_use(name, {}, str(e))
logging.exception(f"Tool call failed: {tc}")
return tc, name, {}, None, e
logging.info(f"Response tool_calls={response.choices[0].message.tool_calls}")
results = await asyncio.gather(*[_exec_tool(tc) for tc in response.choices[0].message.tool_calls])
history = self._append_history_batch(history, results)
for tc, name, args, result, err in results:
ans += self._verbose_tool_use(name, args, err if err else result)
logging.warning(f"Exceed max rounds: {self.max_rounds}")
history.append({"role": "user", "content": f"Exceed max rounds: {self.max_rounds}"})
@@ -398,9 +432,9 @@ class Base(ABC):
for attempt in range(self.max_retries + 1):
history = deepcopy(hist)
try:
for _ in range(self.max_rounds + 1):
for _round in range(self.max_rounds + 1):
reasoning_start = False
logging.info(f"{tools=}")
logging.info(f"[ToolLoop] round={_round} model={self.model_name} tools={[t['function']['name'] for t in tools]}")
response = await self.async_client.chat.completions.create(model=self.model_name, messages=history, stream=True, tools=tools, tool_choice="auto", **gen_conf)
@@ -450,22 +484,36 @@ class Base(ABC):
if finish_reason == "length":
yield self._length_stop("")
if answer:
if answer and not final_tool_calls:
logging.info(f"[ToolLoop] round={_round} completed with text response, exiting")
yield total_tokens
return
for tool_call in final_tool_calls.values():
name = tool_call.function.name
async def _exec_tool(tc):
name = tc.function.name
try:
args = json_repair.loads(tool_call.function.arguments)
yield self._verbose_tool_use(name, args, "Begin to call...")
tool_response = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
history = self._append_history(history, tool_call, tool_response)
yield self._verbose_tool_use(name, args, tool_response)
args = json_repair.loads(tc.function.arguments)
if hasattr(self.toolcall_session, "tool_call_async"):
result = await self.toolcall_session.tool_call_async(name, args)
else:
result = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
return tc, name, args, result, None
except Exception as e:
logging.exception(msg=f"Wrong JSON argument format in LLM tool call response: {tool_call}")
history.append({"role": "tool", "tool_call_id": tool_call.id, "content": f"Tool call error: \n{tool_call}\nException:\n" + str(e)})
yield self._verbose_tool_use(name, {}, str(e))
logging.exception(f"Tool call failed: {tc}")
return tc, name, {}, None, e
tcs = list(final_tool_calls.values())
logging.info(f"[ToolLoop] round={_round} executing {len(tcs)} tool(s): {[tc.function.name for tc in tcs]}")
for tc in tcs:
try:
args = json_repair.loads(tc.function.arguments)
except Exception:
args = {}
yield self._verbose_tool_use(tc.function.name, args, "Begin to call...")
results = await asyncio.gather(*[_exec_tool(tc) for tc in tcs])
history = self._append_history_batch(history, results)
for tc, name, args, result, err in results:
yield self._verbose_tool_use(name, args, err if err else result)
logging.warning(f"Exceed max rounds: {self.max_rounds}")
history.append({"role": "user", "content": f"Exceed max rounds: {self.max_rounds}"})
@@ -1419,6 +1467,34 @@ class LiteLLMBase(ABC):
hist.append({"role": "tool", "tool_call_id": tool_call.id, "content": str(tool_res)})
return hist
def _append_history_batch(self, hist, results):
"""
Append a batch of tool calls to history following the OpenAI protocol:
one assistant message containing all tool_calls, followed by one tool message per call.
results: list of (tool_call, name, args, result, error)
"""
hist.append({
"role": "assistant",
"tool_calls": [
{
"index": tc.index,
"id": tc.id,
"function": {"name": tc.function.name, "arguments": tc.function.arguments},
"type": "function",
}
for tc, _, _, _, _ in results
],
})
for tc, _, _, result, err in results:
if err:
content = str(err)
elif isinstance(result, dict):
content = json.dumps(result, ensure_ascii=False)
else:
content = str(result)
hist.append({"role": "tool", "tool_call_id": tc.id, "content": content})
return hist
def bind_tools(self, toolcall_session, tools):
if not (toolcall_session and tools):
return
@@ -1463,18 +1539,24 @@ class LiteLLMBase(ABC):
ans = self._length_stop(ans)
return ans, tk_count
for tool_call in message.tool_calls:
logging.info(f"Response {tool_call=}")
name = tool_call.function.name
async def _exec_tool(tc):
name = tc.function.name
try:
args = json_repair.loads(tool_call.function.arguments)
tool_response = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
history = self._append_history(history, tool_call, tool_response)
ans += self._verbose_tool_use(name, args, tool_response)
args = json_repair.loads(tc.function.arguments)
if hasattr(self.toolcall_session, "tool_call_async"):
result = await self.toolcall_session.tool_call_async(name, args)
else:
result = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
return tc, name, args, result, None
except Exception as e:
logging.exception(msg=f"Wrong JSON argument format in LLM tool call response: {tool_call}")
history.append({"role": "tool", "tool_call_id": tool_call.id, "content": f"Tool call error: \n{tool_call}\nException:\n" + str(e)})
ans += self._verbose_tool_use(name, {}, str(e))
logging.exception(f"Tool call failed: {tc}")
return tc, name, {}, None, e
logging.info(f"Response tool_calls={message.tool_calls}")
results = await asyncio.gather(*[_exec_tool(tc) for tc in message.tool_calls])
history = self._append_history_batch(history, results)
for tc, name, args, result, err in results:
ans += self._verbose_tool_use(name, args, err if err else result)
logging.warning(f"Exceed max rounds: {self.max_rounds}")
history.append({"role": "user", "content": f"Exceed max rounds: {self.max_rounds}"})
@@ -1503,9 +1585,9 @@ class LiteLLMBase(ABC):
for attempt in range(self.max_retries + 1):
history = deepcopy(hist)
try:
for _ in range(self.max_rounds + 1):
for _round in range(self.max_rounds + 1):
reasoning_start = False
logging.info(f"{tools=}")
logging.info(f"[ToolLoop] round={_round} model={self.model_name} tools={[t['function']['name'] for t in tools]}")
completion_args = self._construct_completion_args(history=history, stream=True, tools=True, **gen_conf)
response = await litellm.acompletion(
@@ -1560,22 +1642,36 @@ class LiteLLMBase(ABC):
if finish_reason == "length":
yield self._length_stop("")
if answer:
if answer and not final_tool_calls:
logging.info(f"[ToolLoop] round={_round} completed with text response, exiting")
yield total_tokens
return
for tool_call in final_tool_calls.values():
name = tool_call.function.name
async def _exec_tool(tc):
name = tc.function.name
try:
args = json_repair.loads(tool_call.function.arguments)
yield self._verbose_tool_use(name, args, "Begin to call...")
tool_response = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
history = self._append_history(history, tool_call, tool_response)
yield self._verbose_tool_use(name, args, tool_response)
args = json_repair.loads(tc.function.arguments)
if hasattr(self.toolcall_session, "tool_call_async"):
result = await self.toolcall_session.tool_call_async(name, args)
else:
result = await thread_pool_exec(self.toolcall_session.tool_call, name, args)
return tc, name, args, result, None
except Exception as e:
logging.exception(msg=f"Wrong JSON argument format in LLM tool call response: {tool_call}")
history.append({"role": "tool", "tool_call_id": tool_call.id, "content": f"Tool call error: \n{tool_call}\nException:\n" + str(e)})
yield self._verbose_tool_use(name, {}, str(e))
logging.exception(f"Tool call failed: {tc}")
return tc, name, {}, None, e
tcs = list(final_tool_calls.values())
logging.info(f"[ToolLoop] round={_round} executing {len(tcs)} tool(s): {[tc.function.name for tc in tcs]}")
for tc in tcs:
try:
args = json_repair.loads(tc.function.arguments)
except Exception:
args = {}
yield self._verbose_tool_use(tc.function.name, args, "Begin to call...")
results = await asyncio.gather(*[_exec_tool(tc) for tc in tcs])
history = self._append_history_batch(history, results)
for tc, name, args, result, err in results:
yield self._verbose_tool_use(name, args, err if err else result)
logging.warning(f"Exceed max rounds: {self.max_rounds}")
history.append({"role": "user", "content": f"Exceed max rounds: {self.max_rounds}"})

6
web/src/components/markdown-content/index.tsx
View File

@@ -211,7 +211,7 @@ const MarkdownContent = ({
const renderReference = useCallback(
(text: string) => {
let replacedText = reactStringReplace(text, currentReg, (match, i) => {
const replacedText = reactStringReplace(text, currentReg, (match, i) => {
const chunkIndex = getChunkIndex(match);
return (
@@ -242,9 +242,7 @@ const MarkdownContent = ({
remarkPlugins={[remarkGfm, remarkMath]}
components={
{
p: ({ children, node, ...props }: any) => (
<p {...props}>{children}</p>
),
p: ({ children, ...props }: any) => <p {...props}>{children}</p>,
'custom-typography': ({ children }: { children: string }) =>
renderReference(children),
code(props: any) {

24
web/src/components/next-markdown-content/index.module.less
View File

@@ -79,3 +79,27 @@
display: inline-block;
max-width: 40px;
}
.artifactImageWrapper {
display: block;
margin: 8px 0;
}
.artifactImage {
max-width: 100%;
max-height: 60vh;
border-radius: 8px;
border: 1px solid #e5e7eb;
display: block;
}
.artifactDownload {
display: inline-block;
margin-top: 4px;
font-size: 12px;
color: #1677ff;
text-decoration: none;
&:hover {
text-decoration: underline;
}
}

156
web/src/components/next-markdown-content/index.tsx
View File

@@ -2,8 +2,10 @@ import Image from '@/components/image';
import SvgIcon from '@/components/svg-icon';
import { IReferenceChunk, IReferenceObject } from '@/interfaces/database/chat';
import { getExtension } from '@/utils/document-util';
import { downloadFileFromBlob } from '@/utils/file-util';
import request from '@/utils/request';
import DOMPurify from 'dompurify';
import { memo, useCallback, useEffect, useMemo } from 'react';
import { memo, useCallback, useEffect, useMemo, useState } from 'react';
import Markdown from 'react-markdown';
import SyntaxHighlighter from 'react-syntax-highlighter';
import rehypeKatex from 'rehype-katex';
@@ -38,9 +40,120 @@ import {
HoverCardContent,
HoverCardTrigger,
} from '../ui/hover-card';
import message from '../ui/message';
import styles from './index.module.less';
const getChunkIndex = (match: string) => parseCitationIndex(match);
const isArtifactUrl = (url?: string) =>
Boolean(url && url.includes('/document/artifact/'));
const fetchArtifactBlob = async (url: string): Promise<Blob> => {
const response = await request(url, {
method: 'GET',
responseType: 'blob',
});
return response.data as Blob;
};
const getArtifactName = (url?: string, fallback?: string) =>
fallback || url?.split('/').pop()?.split('?')[0] || 'artifact';
function ArtifactLink({
href,
className,
children,
}: {
href: string;
className?: string;
children: React.ReactNode;
}) {
const handleClick = useCallback(
async (e: React.MouseEvent<HTMLAnchorElement>) => {
e.preventDefault();
try {
const blob = await fetchArtifactBlob(href);
const objectUrl = URL.createObjectURL(blob);
window.open(objectUrl, '_blank', 'noopener,noreferrer');
window.setTimeout(() => URL.revokeObjectURL(objectUrl), 60 * 1000);
} catch {
message.error('Failed to open artifact');
}
},
[href],
);
return (
<a href={href} className={className} onClick={handleClick}>
{children}
</a>
);
}
function ArtifactImage({
src,
alt,
downloadLabel,
}: {
src: string;
alt?: string;
downloadLabel: string;
}) {
const [imageSrc, setImageSrc] = useState('');
useEffect(() => {
let objectUrl = '';
let active = true;
const load = async () => {
try {
const blob = await fetchArtifactBlob(src);
objectUrl = URL.createObjectURL(blob);
if (active) {
setImageSrc(objectUrl);
}
} catch {
message.error('Failed to load artifact image');
}
};
load();
return () => {
active = false;
if (objectUrl) {
URL.revokeObjectURL(objectUrl);
}
};
}, [alt, src]);
const handleDownload = useCallback(async () => {
try {
const blob = await fetchArtifactBlob(src);
downloadFileFromBlob(blob, getArtifactName(src, alt));
} catch {
message.error('Failed to download artifact');
}
}, [alt, src]);
return (
<span className={styles.artifactImageWrapper}>
{imageSrc ? (
<img src={imageSrc} alt={alt || ''} className={styles.artifactImage} />
) : (
<span className={styles.artifactImage} />
)}
<button
type="button"
className={styles.artifactDownload}
onClick={handleDownload}
>
{downloadLabel}
</button>
</span>
);
}
// TODO: The display of the table is inconsistent with the display previously placed in the MessageItem.
function MarkdownContent({
reference,
@@ -213,7 +326,7 @@ function MarkdownContent({
const renderReference = useCallback(
(text: string) => {
let replacedText = reactStringReplace(text, currentReg, (match, i) => {
const replacedText = reactStringReplace(text, currentReg, (match, i) => {
const chunkIndex = getChunkIndex(match);
return (
@@ -244,11 +357,44 @@ function MarkdownContent({
remarkPlugins={[remarkGfm, remarkMath]}
components={
{
p: ({ children, node, ...props }: any) => (
<p {...props}>{children}</p>
),
p: ({ children, ...props }: any) => <p {...props}>{children}</p>,
'custom-typography': ({ children }: { children: string }) =>
renderReference(children),
a({ href, children, ...props }: any) {
if (isArtifactUrl(href)) {
return (
<ArtifactLink href={href} className={styles.artifactDownload}>
{children}
</ArtifactLink>
);
}
return (
<a href={href} {...omit(props, 'node')}>
{children}
</a>
);
},
img({ src, alt, ...props }: any) {
if (isArtifactUrl(src)) {
return (
<ArtifactImage
src={src}
alt={alt || ''}
downloadLabel={t('common.download')}
/>
);
}
return (
<span className={styles.artifactImageWrapper}>
<img
src={src}
alt={alt || ''}
className={styles.artifactImage}
{...omit(props, 'node')}
/>
</span>
);
},
code(props: any) {
const { children, className, ...rest } = props;
const restProps = omit(rest, 'node');

13
web/src/pages/agent/form/code-form/index.tsx
View File

@@ -42,6 +42,12 @@ const options = [
].map((x) => ({ value: x, label: x }));
const DynamicFieldName = 'outputs';
const CodeSystemOutputs = {
content: {
type: 'string',
value: '',
},
};
function CodeForm({ node }: INextOperatorForm) {
const formData = node?.data.form as ICodeForm;
@@ -159,7 +165,12 @@ function CodeForm({ node }: INextOperatorForm) {
)}
</FormWrapper>
<div className="p-5">
<Output list={buildOutputList(formData.outputs)}></Output>
<Output
list={buildOutputList({
...(formData?.outputs ?? {}),
...CodeSystemOutputs,
})}
></Output>
</div>
</Form>
);

21
web/src/utils/canvas-util.tsx
View File

@@ -61,13 +61,28 @@ export function buildSecondaryOutputOptions(
}));
}
function getNodeOutputs(x: BaseNode) {
const outputs = x.data.form?.outputs ?? {};
if (x.data.label !== Operator.Code) {
return outputs;
}
return {
...outputs,
content: outputs.content ?? {
type: JsonSchemaDataType.String,
value: '',
},
};
}
export function buildOutputOptions(x: BaseNode) {
return {
label: x.data.name,
value: x.id,
title: x.data.name,
options: buildSecondaryOutputOptions(
x.data.form.outputs,
getNodeOutputs(x),
x.id,
x.data.name,
<OperatorIcon name={x.data.label as Operator} />,
@@ -83,7 +98,7 @@ export function buildNodeOutputOptions({
nodeIds: string[];
}) {
const nodeWithOutputList = nodes.filter(
(x) => nodeIds.some((y) => y === x.id) && !isEmpty(x.data?.form?.outputs),
(x) => nodeIds.some((y) => y === x.id) && !isEmpty(getNodeOutputs(x)),
);
return nodeWithOutputList.map((x) => buildOutputOptions(x));
@@ -114,7 +129,7 @@ export function buildChildOutputOptions({
nodeId?: string;
}) {
const nodeWithOutputList = nodes.filter(
(x) => x.parentId === nodeId && !isEmpty(x.data?.form?.outputs),
(x) => x.parentId === nodeId && !isEmpty(getNodeOutputs(x)),
);
return nodeWithOutputList.map((x) => buildOutputOptions(x));