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

Fix the sandbox exec image cannot show and download (#16577)

Browse Source
This commit is contained in:
Wang Qi
2026-07-02 20:49:51 +08:00
committed by GitHub
parent 4a81b9cfde
commit 93f6d647d4
4 changed files with 75 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
agent/tools/code_exec.py
View File

@@ -69,9 +69,7 @@ def select_business_output(outputs: Mapping[str, object]) -> tuple[str, object]:
business_outputs = [(name, meta) for name, meta in outputs.items() if name not in SYSTEM_OUTPUT_KEYS]
if len(business_outputs) != 1:
raise ContractError(
f"CodeExec contract must contain exactly one business output, got {len(business_outputs)}"
)
raise ContractError(f"CodeExec contract must contain exactly one business output, got {len(business_outputs)}")
_validate_business_output_name(business_outputs[0][0])
return business_outputs[0]
@@ -124,10 +122,7 @@ def _is_number(value) -> bool:
def _validate_top_level_value_domain(value) -> None:
allowed = value is None or isinstance(value, (bool, str, dict, list)) or _is_number(value)
if not allowed:
raise ContractError(
f"CodeExec unsupported top-level result type: {type(value).__name__}. "
"Allowed top-level values are String, Number, Boolean, Object, Array, or Null."
)
raise ContractError(f"CodeExec unsupported top-level result type: {type(value).__name__}. Allowed top-level values are String, Number, Boolean, Object, Array, or Null.")
def _normalize_expected_type(expected_type: str) -> str:
@@ -161,9 +156,7 @@ def _validate_expected_type(expected_type: str, value, path: str = "") -> None:
if etype.startswith("Array<") and etype.endswith(">"):
inner_type = etype[6:-1].strip()
if not isinstance(value, list):
raise ContractError(
f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {infer_actual_type(value)}"
)
raise ContractError(f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {infer_actual_type(value)}")
for index, item in enumerate(value):
child_path = f"{path}[{index}]" if path else f"[{index}]"
_validate_expected_type(inner_type, item, child_path)
@@ -184,9 +177,7 @@ def _validate_expected_type(expected_type: str, value, path: str = "") -> None:
raise ContractError(f"Unsupported expected type: {expected_type}")
if not valid:
raise ContractError(
f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {actual_type}"
)
raise ContractError(f"CodeExec contract mismatch at {path or 'value'}: expected type {etype}, got {actual_type}")
def build_code_exec_contract(outputs: Mapping[str, object], raw_result) -> dict[str, object]:
@@ -371,10 +362,7 @@ class CodeExec(ToolBase, ABC):
reload_provider()
provider_info = get_provider_info()
provider_type = provider_info.get("provider_type") or "unknown"
logging.info(
f"[CodeExec]: dispatching execution to sandbox provider '{provider_type}' "
f"(language={language}, timeout={timeout_seconds}s)"
)
logging.info(f"[CodeExec]: dispatching execution to sandbox provider '{provider_type}' (language={language}, timeout={timeout_seconds}s)")
# Execute code using the provider system
result = sandbox_execute_code(code=code, language=language, timeout=timeout_seconds, arguments=arguments)
@@ -556,7 +544,7 @@ class CodeExec(ToolBase, ABC):
settings.STORAGE_IMPL.put(SANDBOX_ARTIFACT_BUCKET, storage_name, binary)
url = f"/api/v1/documents/artifact/{storage_name}"
url = f"/api/v1/documents/artifact/{storage_name}?session_id={self._canvas.task_id}"
uploaded.append(
{
"name": name,

4
api/apps/restful_apis/agent_api.py
View File

@@ -1407,7 +1407,7 @@ async def agent_chat_completion(tenant_id, agent_id=None):
dsl_str = workflow_dsl
else:
dsl_str = json.dumps(workflow_dsl, ensure_ascii=False)
canvas = Canvas(dsl_str, str(tenant_id), canvas_id=agent_id, custom_header=custom_header)
canvas = Canvas(dsl_str, str(tenant_id), task_id=session_id, canvas_id=agent_id, custom_header=custom_header)
except Exception as exc:
return server_error_response(exc)
@@ -1528,7 +1528,7 @@ async def agent_chat_completion(tenant_id, agent_id=None):
try:
from agent.canvas import Canvas
canvas = Canvas(dsl_str, str(tenant_id), canvas_id=agent_id, custom_header=custom_header)
canvas = Canvas(dsl_str, str(tenant_id), task_id=session_id, canvas_id=agent_id, custom_header=custom_header)
canvas.clear_history()
except Exception as exc:
return server_error_response(exc)

94
api/apps/restful_apis/document_api.py
View File

@@ -20,15 +20,21 @@ import os
import re
from pathlib import Path
from quart import request, make_response,send_file
from quart import request, make_response, send_file
from peewee import OperationalError
from pydantic import ValidationError
from api.apps import AUTH_JWT, AUTH_API, AUTH_BETA, current_user, login_required
from api.constants import FILE_NAME_LEN_LIMIT, IMG_BASE64_PREFIX
from api.apps.services.document_api_service import validate_document_update_fields, map_doc_keys, \
map_doc_keys_with_run_status, update_document_name_only, update_chunk_method, update_document_status_only, \
reset_document_for_reparse
from api.apps.services.document_api_service import (
validate_document_update_fields,
map_doc_keys,
map_doc_keys_with_run_status,
update_document_name_only,
update_chunk_method,
update_document_status_only,
reset_document_for_reparse,
)
from api.db import VALID_FILE_TYPES, FileType
from api.db.db_models import API4Conversation, DB
from api.db.services import duplicate_name
@@ -41,11 +47,24 @@ from api.db.services.knowledgebase_service import KnowledgebaseService
from api.db.services.canvas_service import UserCanvasService
from api.common.check_team_permission import check_kb_team_permission
from api.db.services.task_service import TaskService, cancel_all_task_of
from api.utils.api_utils import construct_json_result, get_data_error_result, get_error_data_result, get_result, get_json_result, \
server_error_response, add_tenant_id_to_kwargs, get_request_json, get_error_argument_result, check_duplicate_ids
from api.utils.api_utils import (
construct_json_result,
get_data_error_result,
get_error_data_result,
get_result,
get_json_result,
server_error_response,
add_tenant_id_to_kwargs,
get_request_json,
get_error_argument_result,
check_duplicate_ids,
)
from api.utils.pagination_utils import validate_rest_api_page_size
from api.utils.validation_utils import (
UpdateDocumentReq, format_validation_error_message, validate_and_parse_json_request, DeleteDocumentReq,
UpdateDocumentReq,
format_validation_error_message,
validate_and_parse_json_request,
DeleteDocumentReq,
)
from common import settings
@@ -121,7 +140,7 @@ async def upload_info(tenant_id: str):
return server_error_response(e)
@manager.route("/datasets/<dataset_id>/documents/<document_id>", methods=["PATCH"]) # noqa: F821
@manager.route("/datasets/<dataset_id>/documents/<document_id>", methods=["PATCH"]) # noqa: F821
@login_required
@add_tenant_id_to_kwargs
async def update_document(tenant_id, dataset_id, document_id):
@@ -211,8 +230,7 @@ async def update_document(tenant_id, dataset_id, document_id):
return error
# "parser_id" provided but does not match with existing doc's file type
if "parser_id" in req and ((doc.type == FileType.VISUAL and req["parser_id"] != "picture")
or (re.search(r"\.(ppt|pptx|pages)$", doc.name) and req["parser_id"] != "presentation")):
if "parser_id" in req and ((doc.type == FileType.VISUAL and req["parser_id"] != "picture") or (re.search(r"\.(ppt|pptx|pages)$", doc.name) and req["parser_id"] != "presentation")):
return get_data_error_result(message="Not supported yet!")
# parser config provided (already validated in UpdateDocumentReq), update it
@@ -229,7 +247,7 @@ async def update_document(tenant_id, dataset_id, document_id):
if error := update_chunk_method(req, doc, tenant_id):
return error
if "enabled" in req: # already checked in UpdateDocumentReq - it's int if present
if "enabled" in req: # already checked in UpdateDocumentReq - it's int if present
# "enabled" flag provided, the update method will check if it's changed and then update if so
if error := update_document_status_only(int(req["enabled"]), doc, kb):
return error
@@ -604,7 +622,10 @@ async def _upload_local_documents(kb, tenant_id):
parser_config_override = None
err, files = await thread_pool_exec(
FileService.upload_document, kb, file_objs, tenant_id,
FileService.upload_document,
kb,
file_objs,
tenant_id,
parent_path=form.get("parent_path"),
parser_config_override=parser_config_override,
)
@@ -772,7 +793,7 @@ def list_docs(dataset_id, tenant_id):
return get_json_result(data={"total": total, "docs": renamed_doc_list})
def _get_docs_with_request(req, dataset_id:str):
def _get_docs_with_request(req, dataset_id: str):
"""Get documents with request parameters from a dataset.
This function extracts filtering parameters from the request and returns
@@ -839,7 +860,7 @@ def _get_docs_with_request(req, dataset_id:str):
if doc_id:
if not DocumentService.query(id=doc_id, kb_id=dataset_id):
return RetCode.DATA_ERROR, f"You don't own the document {doc_id}.", [], 0
doc_ids_filter = [doc_id] # id provided, ignore other filters
doc_ids_filter = [doc_id] # id provided, ignore other filters
if doc_name and not DocumentService.query(name=doc_name, kb_id=dataset_id):
return RetCode.DATA_ERROR, f"You don't own the document {doc_name}.", [], 0
@@ -849,8 +870,9 @@ def _get_docs_with_request(req, dataset_id:str):
if len(doc_ids) > 0:
doc_ids_filter = doc_ids
docs, total = DocumentService.get_by_kb_id(dataset_id, page, page_size, orderby, desc, keywords, run_status_converted, types, suffix,
name=doc_name, doc_ids=doc_ids_filter, return_empty_metadata=return_empty_metadata)
docs, total = DocumentService.get_by_kb_id(
dataset_id, page, page_size, orderby, desc, keywords, run_status_converted, types, suffix, name=doc_name, doc_ids=doc_ids_filter, return_empty_metadata=return_empty_metadata
)
# time range filter (0 means no bound)
create_time_from = int(q.get("create_time_from", 0))
@@ -907,6 +929,7 @@ def _parse_run_status_filter(req_args):
invalid_statuses = {status for status in converted if status not in valid_statuses}
return converted, invalid_statuses
def _parse_doc_id_filter_with_metadata(req, kb_id):
"""Parse document ID filter based on metadata conditions from the request.
@@ -976,12 +999,12 @@ def _parse_doc_id_filter_with_metadata(req, kb_id):
try:
metadata_condition = json.loads(req.get("metadata_condition", "{}"))
except json.JSONDecodeError:
msg = f'metadata_condition must be valid JSON: {req.get("metadata_condition")}.'
msg = f"metadata_condition must be valid JSON: {req.get('metadata_condition')}."
return RetCode.DATA_ERROR, msg, [], return_empty_metadata
try:
metadata = json.loads(req.get("metadata", "{}"))
except json.JSONDecodeError:
logging.error(msg=f'metadata must be valid JSON: {req.get("metadata")}.')
logging.error(msg=f"metadata must be valid JSON: {req.get('metadata')}.")
return RetCode.DATA_ERROR, "metadata must be valid JSON.", [], return_empty_metadata
if isinstance(metadata, dict) and metadata.get("empty_metadata"):
@@ -1106,9 +1129,7 @@ async def delete_documents(tenant_id, dataset_id):
dataset_doc_ids = {doc.id for doc in DocumentService.query(kb_id=dataset_id)}
invalid_ids = [doc_id for doc_id in doc_ids if doc_id not in dataset_doc_ids]
if invalid_ids:
return get_error_data_result(
message=f"These documents do not belong to dataset {dataset_id} or Document not found: {', '.join(invalid_ids)}"
)
return get_error_data_result(message=f"These documents do not belong to dataset {dataset_id} or Document not found: {', '.join(invalid_ids)}")
# make sure each id is unique
unique_doc_ids, duplicate_messages = check_duplicate_ids(doc_ids, "document")
@@ -1128,6 +1149,7 @@ async def delete_documents(tenant_id, dataset_id):
logging.exception(e)
return get_error_data_result(message="Internal server error")
@manager.route("/datasets/<dataset_id>/documents/<document_id>/metadata/config", methods=["PUT"]) # noqa: F821
@login_required
@add_tenant_id_to_kwargs
@@ -1349,17 +1371,13 @@ async def update_metadata(tenant_id, dataset_id):
kb_doc_ids = KnowledgebaseService.list_documents_by_ids([dataset_id])
invalid_ids = set(document_ids) - set(kb_doc_ids)
if invalid_ids:
return get_error_data_result(
message=f"These documents do not belong to dataset {dataset_id}: {', '.join(invalid_ids)}"
)
return get_error_data_result(message=f"These documents do not belong to dataset {dataset_id}: {', '.join(invalid_ids)}")
target_doc_ids = set(document_ids)
# Apply metadata_condition filtering if provided
if metadata_condition:
metas = DocMetadataService.get_flatted_meta_by_kbs([dataset_id])
filtered_ids = set(
meta_filter(metas, convert_conditions(metadata_condition), metadata_condition.get("logic", "and"))
)
filtered_ids = set(meta_filter(metas, convert_conditions(metadata_condition), metadata_condition.get("logic", "and")))
target_doc_ids = target_doc_ids & filtered_ids
if metadata_condition.get("conditions") and not target_doc_ids:
return get_result(data={"updated": 0, "matched_docs": 0})
@@ -1389,7 +1407,8 @@ async def ingest(tenant_id):
logging.exception("document ingest/run failed")
return server_error_response(e)
def _run_sync(user_id:str, req):
def _run_sync(user_id: str, req):
for doc_id in req["doc_ids"]:
if not DocumentService.accessible(doc_id, user_id):
return RetCode.AUTHENTICATION_ERROR, "No authorization."
@@ -1513,6 +1532,7 @@ async def parse_documents(tenant_id, dataset_id):
return get_error_data_result(message=f"Documents not found: {not_found_ids}")
try:
def _run_sync():
kb_table_num_map = {}
success_count = 0
@@ -1622,6 +1642,7 @@ async def stop_parse_documents(tenant_id, dataset_id):
return get_error_data_result(message=f"Documents not found: {not_found_ids}")
try:
def _run_sync():
success_count = 0
for doc_id in valid_doc_ids:
@@ -1787,6 +1808,18 @@ def _sandbox_artifact_accessible(filename: str, user_id: str) -> bool:
return False
@DB.connection_context()
def _sandbox_artifact_session_accessible(session_id: str, user_id: str) -> bool:
if not session_id:
return False
conv = API4Conversation.get_or_none(API4Conversation.id == session_id)
if not conv:
return False
if str(conv.user_id) != str(user_id) and str(conv.exp_user_id or "") != str(user_id):
return False
return UserCanvasService.accessible(conv.dialog_id, user_id)
@manager.route("/documents/artifact/<filename>", methods=["GET"]) # noqa: F821
@login_required
async def get_artifact(filename):
@@ -1823,7 +1856,8 @@ async def get_artifact(filename):
ext = os.path.splitext(basename)[1].lower()
if ext not in ARTIFACT_CONTENT_TYPES:
return get_data_error_result(message="Invalid file type.")
if not await thread_pool_exec(_sandbox_artifact_accessible, basename, current_user.id):
session_id = request.args.get("session_id", "")
if not await thread_pool_exec(_sandbox_artifact_accessible, basename, current_user.id) and not await thread_pool_exec(_sandbox_artifact_session_accessible, session_id, current_user.id):
return get_data_error_result(message="Artifact not found.")
data = await thread_pool_exec(settings.STORAGE_IMPL.get, bucket, basename)
if not data:
@@ -1959,6 +1993,7 @@ async def batch_update_document_status(tenant_id, dataset_id):
return get_json_result(data=result, message="Partial failure", code=RetCode.SERVER_ERROR)
return get_json_result(data=result)
@manager.route("/documents/<doc_id>/preview", methods=["GET"]) # noqa: F821
@login_required(auth_types=[AUTH_JWT, AUTH_API, AUTH_BETA])
async def get(doc_id):
@@ -2064,6 +2099,7 @@ async def download(dataset_id, document_id):
mimetype=_mimetype_for_document(doc[0]),
)
@manager.route("/documents/<document_id>", methods=["GET"]) # noqa: F821
@login_required
async def download_document(document_id):

4
api/db/services/canvas_service.py
View File

@@ -316,12 +316,12 @@ async def completion(tenant_id, agent_id, session_id=None, **kwargs):
conv.message = []
if not isinstance(conv.dsl, str):
conv.dsl = json.dumps(conv.dsl, ensure_ascii=False)
canvas = Canvas(conv.dsl, tenant_id, agent_id, canvas_id=agent_id, custom_header=custom_header)
canvas = Canvas(conv.dsl, tenant_id, task_id=session_id, canvas_id=agent_id, custom_header=custom_header)
else:
cvs, dsl = await thread_pool_exec(UserCanvasService.get_agent_dsl_with_release, agent_id, release_mode=release_mode == "true", tenant_id=tenant_id)
session_id = get_uuid()
canvas = Canvas(dsl, tenant_id, agent_id, canvas_id=cvs.id, custom_header=custom_header)
canvas = Canvas(dsl, tenant_id, task_id=session_id, canvas_id=cvs.id, custom_header=custom_header)
canvas.reset()
# Get the version title based on release_mode
version_title = await thread_pool_exec(UserCanvasVersionService.get_latest_version_title, cvs.id, release_mode=release_mode == "true")