diff --git a/api/apps/canvas_app.py b/api/apps/canvas_app.py deleted file mode 100644 index b16a97c46..000000000 --- a/api/apps/canvas_app.py +++ /dev/null @@ -1,662 +0,0 @@ -# -# Copyright 2024 The InfiniFlow Authors. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -import copy -import inspect -import json -import logging -from functools import partial -from quart import request, Response, make_response -from agent.component import LLM -from api.db import CanvasCategory -from api.db.services.canvas_service import CanvasTemplateService, UserCanvasService, API4ConversationService -from api.db.services.document_service import DocumentService -from api.db.services.file_service import FileService -from api.db.services.pipeline_operation_log_service import PipelineOperationLogService -from api.db.services.task_service import queue_dataflow, CANVAS_DEBUG_DOC_ID, TaskService -from api.db.services.user_service import TenantService -from api.db.services.user_canvas_version import UserCanvasVersionService -from common.constants import RetCode -from common.misc_utils import get_uuid, thread_pool_exec -from api.utils.api_utils import ( - get_json_result, - server_error_response, - validate_request, - get_data_error_result, - get_request_json, -) -from agent.canvas import Canvas -from peewee import MySQLDatabase, PostgresqlDatabase -from api.db.db_models import APIToken, Task - -from rag.flow.pipeline import Pipeline -from rag.nlp import search -from rag.utils.redis_conn import REDIS_CONN -from common import settings -from api.apps import login_required, current_user -from api.apps.services.canvas_replica_service import CanvasReplicaService -from api.db.services.canvas_service import completion as agent_completion - - -@manager.route("/templates", methods=["GET"]) # noqa: F821 -@login_required -def templates(): - return get_json_result(data=[c.to_dict() for c in CanvasTemplateService.get_all()]) - - -@manager.route("/rm", methods=["POST"]) # noqa: F821 -@validate_request("canvas_ids") -@login_required -async def rm(): - req = await get_request_json() - for i in req["canvas_ids"]: - if not UserCanvasService.accessible(i, current_user.id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - UserCanvasService.delete_by_id(i) - return get_json_result(data=True) - - -@manager.route("/set", methods=["POST"]) # noqa: F821 -@validate_request("dsl", "title") -@login_required -async def save(): - req = await get_request_json() - try: - req["dsl"] = CanvasReplicaService.normalize_dsl(req["dsl"]) - except ValueError as e: - return get_data_error_result(message=str(e)) - cate = req.get("canvas_category", CanvasCategory.Agent) - if "id" not in req: - req["user_id"] = current_user.id - if UserCanvasService.query(user_id=current_user.id, title=req["title"].strip(), canvas_category=cate): - return get_data_error_result(message=f"{req['title'].strip()} already exists.") - req["id"] = get_uuid() - if not UserCanvasService.save(**req): - return get_data_error_result(message="Fail to save canvas.") - else: - if not UserCanvasService.accessible(req["id"], current_user.id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - UserCanvasService.update_by_id(req["id"], req) - # save version - UserCanvasVersionService.save_or_replace_latest( - user_canvas_id=req["id"], - dsl=req["dsl"], - title=UserCanvasVersionService.build_version_title(getattr(current_user, "nickname", current_user.id), req.get("title")), - ) - replica_ok = CanvasReplicaService.replace_for_set( - canvas_id=req["id"], - tenant_id=str(current_user.id), - runtime_user_id=str(current_user.id), - dsl=req["dsl"], - canvas_category=req.get("canvas_category", cate), - title=req.get("title", ""), - ) - if not replica_ok: - return get_data_error_result(message="canvas saved, but replica sync failed.") - return get_json_result(data=req) - - -@manager.route("/get/", methods=["GET"]) # noqa: F821 -@login_required -def get(canvas_id): - if not UserCanvasService.accessible(canvas_id, current_user.id): - return get_data_error_result(message="canvas not found.") - e, c = UserCanvasService.get_by_canvas_id(canvas_id) - if not e: - return get_data_error_result(message="canvas not found.") - try: - # DELETE - CanvasReplicaService.bootstrap( - canvas_id=canvas_id, - tenant_id=str(current_user.id), - runtime_user_id=str(current_user.id), - dsl=c.get("dsl"), - canvas_category=c.get("canvas_category", CanvasCategory.Agent), - title=c.get("title", ""), - ) - except ValueError as e: - return get_data_error_result(message=str(e)) - return get_json_result(data=c) - - -@manager.route("/getsse/", methods=["GET"]) # type: ignore # noqa: F821 -def getsse(canvas_id): - token = request.headers.get("Authorization").split() - if len(token) != 2: - return get_data_error_result(message="Authorization is not valid!") - token = token[1] - objs = APIToken.query(beta=token) - if not objs: - return get_data_error_result(message='Authentication error: API key is invalid!"') - tenant_id = objs[0].tenant_id - if not UserCanvasService.query(user_id=tenant_id, id=canvas_id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - e, c = UserCanvasService.get_by_id(canvas_id) - if not e or c.user_id != tenant_id: - return get_data_error_result(message="canvas not found.") - return get_json_result(data=c.to_dict()) - - -@manager.route("/completion", methods=["POST"]) # noqa: F821 -@validate_request("id") -@login_required -async def run(): - req = await get_request_json() - query = req.get("query", "") - files = req.get("files", []) - inputs = req.get("inputs", {}) - tenant_id = str(current_user.id) - runtime_user_id = req.get("user_id") or tenant_id - user_id = str(runtime_user_id) - if not await thread_pool_exec(UserCanvasService.accessible, req["id"], tenant_id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - - replica_payload = CanvasReplicaService.load_for_run( - canvas_id=req["id"], - tenant_id=tenant_id, - runtime_user_id=user_id, - ) - - if not replica_payload: - return get_data_error_result(message="canvas replica not found, please call /get/ first.") - - replica_dsl = replica_payload.get("dsl", {}) - canvas_title = replica_payload.get("title", "") - canvas_category = replica_payload.get("canvas_category", CanvasCategory.Agent) - dsl_str = json.dumps(replica_dsl, ensure_ascii=False) - - if canvas_category == CanvasCategory.DataFlow: - task_id = get_uuid() - Pipeline(dsl_str, tenant_id=tenant_id, doc_id=CANVAS_DEBUG_DOC_ID, task_id=task_id, flow_id=req["id"]) - ok, error_message = await thread_pool_exec(queue_dataflow, user_id, req["id"], task_id, CANVAS_DEBUG_DOC_ID, files[0], 0) - if not ok: - return get_data_error_result(message=error_message) - return get_json_result(data={"message_id": task_id}) - - try: - canvas = Canvas(dsl_str, tenant_id, canvas_id=req["id"]) - except Exception as e: - return server_error_response(e) - - async def sse(): - nonlocal canvas, user_id - try: - async for ans in canvas.run(query=query, files=files, user_id=user_id, inputs=inputs): - yield "data:" + json.dumps(ans, ensure_ascii=False) + "\n\n" - - commit_ok = CanvasReplicaService.commit_after_run( - canvas_id=req["id"], - tenant_id=tenant_id, - runtime_user_id=user_id, - dsl=json.loads(str(canvas)), - canvas_category=canvas_category, - title=canvas_title, - ) - if not commit_ok: - logging.error( - "Canvas runtime replica commit failed: canvas_id=%s tenant_id=%s runtime_user_id=%s", - req["id"], - tenant_id, - user_id, - ) - - except Exception as e: - logging.exception(e) - canvas.cancel_task() - yield "data:" + json.dumps({"code": 500, "message": str(e), "data": False}, ensure_ascii=False) + "\n\n" - finally: - canvas.close() - - resp = Response(sse(), mimetype="text/event-stream") - resp.headers.add_header("Cache-control", "no-cache") - resp.headers.add_header("Connection", "keep-alive") - resp.headers.add_header("X-Accel-Buffering", "no") - resp.headers.add_header("Content-Type", "text/event-stream; charset=utf-8") - # resp.call_on_close(lambda: canvas.cancel_task()) - return resp - - -@manager.route("//completion", methods=["POST"]) # noqa: F821 -@login_required -async def exp_agent_completion(canvas_id): - tenant_id = current_user.id - req = await get_request_json() - return_trace = bool(req.get("return_trace", False)) - - async def generate(): - trace_items = [] - async for answer in agent_completion(tenant_id=tenant_id, agent_id=canvas_id, **req): - if isinstance(answer, str): - try: - ans = json.loads(answer[5:]) # remove "data:" - except Exception: - continue - - event = ans.get("event") - if event == "node_finished": - if return_trace: - data = ans.get("data", {}) - trace_items.append( - { - "component_id": data.get("component_id"), - "trace": [copy.deepcopy(data)], - } - ) - ans.setdefault("data", {})["trace"] = trace_items - answer = "data:" + json.dumps(ans, ensure_ascii=False) + "\n\n" - yield answer - - if event not in ["message", "message_end"]: - continue - - yield answer - - yield "data:[DONE]\n\n" - - resp = Response(generate(), mimetype="text/event-stream") - resp.headers.add_header("Cache-control", "no-cache") - resp.headers.add_header("Connection", "keep-alive") - resp.headers.add_header("X-Accel-Buffering", "no") - resp.headers.add_header("Content-Type", "text/event-stream; charset=utf-8") - return resp - - -@manager.route("/rerun", methods=["POST"]) # noqa: F821 -@validate_request("id", "dsl", "component_id") -@login_required -async def rerun(): - req = await get_request_json() - doc = PipelineOperationLogService.get_documents_info(req["id"]) - if not doc: - return get_data_error_result(message="Document not found.") - doc = doc[0] - if 0 < doc["progress"] < 1: - return get_data_error_result(message=f"`{doc['name']}` is processing...") - - if settings.docStoreConn.index_exist(search.index_name(current_user.id), doc["kb_id"]): - settings.docStoreConn.delete({"doc_id": doc["id"]}, search.index_name(current_user.id), doc["kb_id"]) - doc["progress_msg"] = "" - doc["chunk_num"] = 0 - doc["token_num"] = 0 - DocumentService.clear_chunk_num_when_rerun(doc["id"]) - DocumentService.update_by_id(id, doc) - TaskService.filter_delete([Task.doc_id == id]) - - dsl = req["dsl"] - dsl["path"] = [req["component_id"]] - PipelineOperationLogService.update_by_id(req["id"], {"dsl": dsl}) - queue_dataflow(tenant_id=current_user.id, flow_id=req["id"], task_id=get_uuid(), doc_id=doc["id"], priority=0, rerun=True) - return get_json_result(data=True) - - -@manager.route("/cancel/", methods=["PUT"]) # noqa: F821 -@login_required -def cancel(task_id): - try: - REDIS_CONN.set(f"{task_id}-cancel", "x") - except Exception as e: - logging.exception(e) - return get_json_result(data=True) - - -@manager.route("/reset", methods=["POST"]) # noqa: F821 -@validate_request("id") -@login_required -async def reset(): - req = await get_request_json() - if not UserCanvasService.accessible(req["id"], current_user.id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - try: - e, user_canvas = UserCanvasService.get_by_id(req["id"]) - if not e: - return get_data_error_result(message="canvas not found.") - - canvas = Canvas(json.dumps(user_canvas.dsl), current_user.id, canvas_id=user_canvas.id) - canvas.reset() - req["dsl"] = json.loads(str(canvas)) - UserCanvasService.update_by_id(req["id"], {"dsl": req["dsl"]}) - return get_json_result(data=req["dsl"]) - except Exception as e: - return server_error_response(e) - - -@manager.route("/upload/", methods=["POST"]) # noqa: F821 -async def upload(canvas_id): - e, cvs = UserCanvasService.get_by_canvas_id(canvas_id) - if not e: - return get_data_error_result(message="canvas not found.") - - user_id = cvs["user_id"] - files = await request.files - file_objs = files.getlist("file") if files and files.get("file") else [] - try: - if len(file_objs) == 1: - return get_json_result(data=FileService.upload_info(user_id, file_objs[0], request.args.get("url"))) - results = [FileService.upload_info(user_id, f) for f in file_objs] - return get_json_result(data=results) - except Exception as e: - return server_error_response(e) - - -@manager.route("/input_form", methods=["GET"]) # noqa: F821 -@login_required -def input_form(): - cvs_id = request.args.get("id") - cpn_id = request.args.get("component_id") - try: - e, user_canvas = UserCanvasService.get_by_id(cvs_id) - if not e: - return get_data_error_result(message="canvas not found.") - if not UserCanvasService.query(user_id=current_user.id, id=cvs_id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - - canvas = Canvas(json.dumps(user_canvas.dsl), current_user.id, canvas_id=user_canvas.id) - return get_json_result(data=canvas.get_component_input_form(cpn_id)) - except Exception as e: - return server_error_response(e) - - -@manager.route("/debug", methods=["POST"]) # noqa: F821 -@validate_request("id", "component_id", "params") -@login_required -async def debug(): - req = await get_request_json() - if not UserCanvasService.accessible(req["id"], current_user.id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - try: - e, user_canvas = UserCanvasService.get_by_id(req["id"]) - canvas = Canvas(json.dumps(user_canvas.dsl), current_user.id, canvas_id=user_canvas.id) - canvas.reset() - canvas.message_id = get_uuid() - component = canvas.get_component(req["component_id"])["obj"] - component.reset() - - if isinstance(component, LLM): - component.set_debug_inputs(req["params"]) - component.invoke(**{k: o["value"] for k, o in req["params"].items()}) - outputs = component.output() - for k in outputs.keys(): - if isinstance(outputs[k], partial): - txt = "" - iter_obj = outputs[k]() - if inspect.isasyncgen(iter_obj): - async for c in iter_obj: - txt += c - else: - for c in iter_obj: - txt += c - outputs[k] = txt - return get_json_result(data=outputs) - except Exception as e: - return server_error_response(e) - - -@manager.route("/test_db_connect", methods=["POST"]) # noqa: F821 -@validate_request("db_type", "database", "username", "host", "port", "password") -@login_required -async def test_db_connect(): - req = await get_request_json() - try: - if req["db_type"] in ["mysql", "mariadb"]: - db = MySQLDatabase(req["database"], user=req["username"], host=req["host"], port=req["port"], password=req["password"]) - elif req["db_type"] == "oceanbase": - db = MySQLDatabase(req["database"], user=req["username"], host=req["host"], port=req["port"], password=req["password"], charset="utf8mb4") - elif req["db_type"] == "postgres": - db = PostgresqlDatabase(req["database"], user=req["username"], host=req["host"], port=req["port"], password=req["password"]) - elif req["db_type"] == "mssql": - import pyodbc - - connection_string = f"DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={req['host']},{req['port']};DATABASE={req['database']};UID={req['username']};PWD={req['password']};" - db = pyodbc.connect(connection_string) - cursor = db.cursor() - cursor.execute("SELECT 1") - cursor.close() - elif req["db_type"] == "IBM DB2": - import ibm_db - - conn_str = f"DATABASE={req['database']};HOSTNAME={req['host']};PORT={req['port']};PROTOCOL=TCPIP;UID={req['username']};PWD={req['password']};" - redacted_conn_str = f"DATABASE={req['database']};HOSTNAME={req['host']};PORT={req['port']};PROTOCOL=TCPIP;UID={req['username']};PWD=****;" - logging.info(redacted_conn_str) - conn = ibm_db.connect(conn_str, "", "") - stmt = ibm_db.exec_immediate(conn, "SELECT 1 FROM sysibm.sysdummy1") - ibm_db.fetch_assoc(stmt) - ibm_db.close(conn) - return get_json_result(data="Database Connection Successful!") - elif req["db_type"] == "trino": - - def _parse_catalog_schema(db_name: str): - if not db_name: - return None, None - if "." in db_name: - catalog_name, schema_name = db_name.split(".", 1) - elif "/" in db_name: - catalog_name, schema_name = db_name.split("/", 1) - else: - catalog_name, schema_name = db_name, "default" - return catalog_name, schema_name - - try: - import trino - import os - except Exception as e: - return server_error_response(f"Missing dependency 'trino'. Please install: pip install trino, detail: {e}") - - catalog, schema = _parse_catalog_schema(req["database"]) - if not catalog: - return server_error_response("For Trino, 'database' must be 'catalog.schema' or at least 'catalog'.") - - http_scheme = "https" if os.environ.get("TRINO_USE_TLS", "0") == "1" else "http" - - auth = None - if http_scheme == "https" and req.get("password"): - auth = trino.BasicAuthentication(req.get("username") or "ragflow", req["password"]) - - conn = trino.dbapi.connect( - host=req["host"], port=int(req["port"] or 8080), user=req["username"] or "ragflow", catalog=catalog, schema=schema or "default", http_scheme=http_scheme, auth=auth - ) - cur = conn.cursor() - cur.execute("SELECT 1") - cur.fetchall() - cur.close() - conn.close() - return get_json_result(data="Database Connection Successful!") - else: - return server_error_response("Unsupported database type.") - if req["db_type"] != "mssql": - db.connect() - db.close() - - return get_json_result(data="Database Connection Successful!") - except Exception as e: - return server_error_response(e) - - -# api get list version dsl of canvas -@manager.route("/getlistversion/", methods=["GET"]) # noqa: F821 -@login_required -def getlistversion(canvas_id): - try: - versions = sorted([c.to_dict() for c in UserCanvasVersionService.list_by_canvas_id(canvas_id)], key=lambda x: x["update_time"] * -1) - return get_json_result(data=versions) - except Exception as e: - return get_data_error_result(message=f"Error getting history files: {e}") - - -# api get version dsl of canvas -@manager.route("/getversion/", methods=["GET"]) # noqa: F821 -@login_required -def getversion(version_id): - try: - e, version = UserCanvasVersionService.get_by_id(version_id) - if version: - return get_json_result(data=version.to_dict()) - except Exception as e: - return get_json_result(data=f"Error getting history file: {e}") - - -@manager.route("/list", methods=["GET"]) # noqa: F821 -@login_required -def list_canvas(): - keywords = request.args.get("keywords", "") - page_number = int(request.args.get("page", 0)) - items_per_page = int(request.args.get("page_size", 0)) - orderby = request.args.get("orderby", "create_time") - canvas_category = request.args.get("canvas_category") - if request.args.get("desc", "true").lower() == "false": - desc = False - else: - desc = True - owner_ids = [id for id in request.args.get("owner_ids", "").strip().split(",") if id] - if not owner_ids: - tenants = TenantService.get_joined_tenants_by_user_id(current_user.id) - tenants = [m["tenant_id"] for m in tenants] - tenants.append(current_user.id) - canvas, total = UserCanvasService.get_by_tenant_ids(tenants, current_user.id, page_number, items_per_page, orderby, desc, keywords, canvas_category) - else: - tenants = owner_ids - canvas, total = UserCanvasService.get_by_tenant_ids(tenants, current_user.id, 0, 0, orderby, desc, keywords, canvas_category) - return get_json_result(data={"canvas": canvas, "total": total}) - - -@manager.route("/setting", methods=["POST"]) # noqa: F821 -@validate_request("id", "title", "permission") -@login_required -async def setting(): - req = await get_request_json() - req["user_id"] = current_user.id - - if not UserCanvasService.accessible(req["id"], current_user.id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - - e, flow = UserCanvasService.get_by_id(req["id"]) - if not e: - return get_data_error_result(message="canvas not found.") - flow = flow.to_dict() - flow["title"] = req["title"] - - for key in ["description", "permission", "avatar"]: - if value := req.get(key): - flow[key] = value - - num = UserCanvasService.update_by_id(req["id"], flow) - return get_json_result(data=num) - - -@manager.route("/trace", methods=["GET"]) # noqa: F821 -def trace(): - cvs_id = request.args.get("canvas_id") - msg_id = request.args.get("message_id") - try: - binary = REDIS_CONN.get(f"{cvs_id}-{msg_id}-logs") - if not binary: - return get_json_result(data={}) - - return get_json_result(data=json.loads(binary.encode("utf-8"))) - except Exception as e: - logging.exception(e) - - -@manager.route("//sessions", methods=["GET"]) # noqa: F821 -@login_required -def sessions(canvas_id): - tenant_id = current_user.id - if not UserCanvasService.accessible(canvas_id, tenant_id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - - user_id = request.args.get("user_id") - page_number = int(request.args.get("page", 1)) - items_per_page = int(request.args.get("page_size", 30)) - keywords = request.args.get("keywords") - from_date = request.args.get("from_date") - to_date = request.args.get("to_date") - orderby = request.args.get("orderby", "update_time") - exp_user_id = request.args.get("exp_user_id") - if request.args.get("desc") == "False" or request.args.get("desc") == "false": - desc = False - else: - desc = True - - if exp_user_id: - sess = API4ConversationService.get_names(canvas_id, exp_user_id) - return get_json_result(data={"total": len(sess), "sessions": sess}) - - # dsl defaults to True in all cases except for False and false - include_dsl = request.args.get("dsl") != "False" and request.args.get("dsl") != "false" - total, sess = API4ConversationService.get_list(canvas_id, tenant_id, page_number, items_per_page, orderby, desc, None, user_id, include_dsl, keywords, from_date, to_date, exp_user_id=exp_user_id) - try: - return get_json_result(data={"total": total, "sessions": sess}) - except Exception as e: - return server_error_response(e) - - -@manager.route("//sessions", methods=["PUT"]) # noqa: F821 -@login_required -async def set_session(canvas_id): - req = await get_request_json() - tenant_id = current_user.id - e, cvs = UserCanvasService.get_by_id(canvas_id) - assert e, "Agent not found." - if not isinstance(cvs.dsl, str): - cvs.dsl = json.dumps(cvs.dsl, ensure_ascii=False) - session_id = get_uuid() - canvas = Canvas(cvs.dsl, tenant_id, canvas_id, canvas_id=cvs.id) - canvas.reset() - conv = {"id": session_id, "name": req.get("name", ""), "dialog_id": cvs.id, "user_id": tenant_id, "exp_user_id": tenant_id, "message": [], "source": "agent", "dsl": cvs.dsl, "reference": []} - API4ConversationService.save(**conv) - return get_json_result(data=conv) - - -@manager.route("//sessions/", methods=["GET"]) # noqa: F821 -@login_required -def get_session(canvas_id, session_id): - tenant_id = current_user.id - if not UserCanvasService.accessible(canvas_id, tenant_id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - _, conv = API4ConversationService.get_by_id(session_id) - return get_json_result(data=conv.to_dict()) - - -@manager.route("//sessions/", methods=["DELETE"]) # noqa: F821 -@login_required -def del_session(canvas_id, session_id): - tenant_id = current_user.id - if not UserCanvasService.accessible(canvas_id, tenant_id): - return get_json_result(data=False, message="Only owner of canvas authorized for this operation.", code=RetCode.OPERATING_ERROR) - return get_json_result(data=API4ConversationService.delete_by_id(session_id)) - - -@manager.route("/prompts", methods=["GET"]) # noqa: F821 -@login_required -def prompts(): - from rag.prompts.generator import ANALYZE_TASK_SYSTEM, ANALYZE_TASK_USER, NEXT_STEP, REFLECT, CITATION_PROMPT_TEMPLATE - - return get_json_result( - data={ - "task_analysis": ANALYZE_TASK_SYSTEM + "\n\n" + ANALYZE_TASK_USER, - "plan_generation": NEXT_STEP, - "reflection": REFLECT, - # "context_summary": SUMMARY4MEMORY, - # "context_ranking": RANK_MEMORY, - "citation_guidelines": CITATION_PROMPT_TEMPLATE, - } - ) - - -@manager.route("/download", methods=["GET"]) # noqa: F821 -async def download(): - id = request.args.get("id") - created_by = request.args.get("created_by") - blob = FileService.get_blob(created_by, id) - return await make_response(blob) diff --git a/api/db/services/evaluation_service.py b/api/db/services/evaluation_service.py deleted file mode 100644 index d2dadef85..000000000 --- a/api/db/services/evaluation_service.py +++ /dev/null @@ -1,634 +0,0 @@ -# -# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -""" -RAG Evaluation Service - -Provides functionality for evaluating RAG system performance including: -- Dataset management -- Test case management -- Evaluation execution -- Metrics computation -- Configuration recommendations -""" - -import asyncio -import logging -import queue -import threading -from typing import List, Dict, Any, Optional, Tuple -from datetime import datetime -from timeit import default_timer as timer - -from api.db.db_models import EvaluationDataset, EvaluationCase, EvaluationRun, EvaluationResult -from api.db.services.common_service import CommonService -from api.db.services.dialog_service import DialogService -from common.misc_utils import get_uuid -from common.time_utils import current_timestamp -from common.constants import StatusEnum -from common.token_utils import num_tokens_from_string - - -class EvaluationService(CommonService): - """Service for managing RAG evaluations""" - - model = EvaluationDataset - - # ==================== Dataset Management ==================== - - @classmethod - def create_dataset(cls, name: str, description: str, kb_ids: List[str], tenant_id: str, user_id: str) -> Tuple[bool, str]: - """ - Create a new evaluation dataset. - - Args: - name: Dataset name - description: Dataset description - kb_ids: List of knowledge base IDs to evaluate against - tenant_id: Tenant ID - user_id: User ID who creates the dataset - - Returns: - (success, dataset_id or error_message) - """ - try: - timestamp = current_timestamp() - dataset_id = get_uuid() - dataset = { - "id": dataset_id, - "tenant_id": tenant_id, - "name": name, - "description": description, - "kb_ids": kb_ids, - "created_by": user_id, - "create_time": timestamp, - "update_time": timestamp, - "status": StatusEnum.VALID.value, - } - - if not EvaluationDataset.create(**dataset): - return False, "Failed to create dataset" - - return True, dataset_id - except Exception as e: - logging.error(f"Error creating evaluation dataset: {e}") - return False, str(e) - - @classmethod - def get_dataset(cls, dataset_id: str) -> Optional[Dict[str, Any]]: - """Get dataset by ID""" - try: - dataset = EvaluationDataset.get_by_id(dataset_id) - if dataset: - return dataset.to_dict() - return None - except Exception as e: - logging.error(f"Error getting dataset {dataset_id}: {e}") - return None - - @classmethod - def list_datasets(cls, tenant_id: str, user_id: str, page: int = 1, page_size: int = 20) -> Dict[str, Any]: - """List datasets for a tenant""" - try: - query = EvaluationDataset.select().where((EvaluationDataset.tenant_id == tenant_id) & (EvaluationDataset.status == StatusEnum.VALID.value)).order_by(EvaluationDataset.create_time.desc()) - - total = query.count() - datasets = query.paginate(page, page_size) - - return {"total": total, "datasets": [d.to_dict() for d in datasets]} - except Exception as e: - logging.error(f"Error listing datasets: {e}") - return {"total": 0, "datasets": []} - - @classmethod - def update_dataset(cls, dataset_id: str, **kwargs) -> bool: - """Update dataset""" - try: - kwargs["update_time"] = current_timestamp() - return EvaluationDataset.update(**kwargs).where(EvaluationDataset.id == dataset_id).execute() > 0 - except Exception as e: - logging.error(f"Error updating dataset {dataset_id}: {e}") - return False - - @classmethod - def delete_dataset(cls, dataset_id: str) -> bool: - """Soft delete dataset""" - try: - return EvaluationDataset.update(status=StatusEnum.INVALID.value, update_time=current_timestamp()).where(EvaluationDataset.id == dataset_id).execute() > 0 - except Exception as e: - logging.error(f"Error deleting dataset {dataset_id}: {e}") - return False - - # ==================== Test Case Management ==================== - - @classmethod - def add_test_case( - cls, - dataset_id: str, - question: str, - reference_answer: Optional[str] = None, - relevant_doc_ids: Optional[List[str]] = None, - relevant_chunk_ids: Optional[List[str]] = None, - metadata: Optional[Dict[str, Any]] = None, - ) -> Tuple[bool, str]: - """ - Add a test case to a dataset. - - Args: - dataset_id: Dataset ID - question: Test question - reference_answer: Optional ground truth answer - relevant_doc_ids: Optional list of relevant document IDs - relevant_chunk_ids: Optional list of relevant chunk IDs - metadata: Optional additional metadata - - Returns: - (success, case_id or error_message) - """ - try: - case_id = get_uuid() - case = { - "id": case_id, - "dataset_id": dataset_id, - "question": question, - "reference_answer": reference_answer, - "relevant_doc_ids": relevant_doc_ids, - "relevant_chunk_ids": relevant_chunk_ids, - "metadata": metadata, - "create_time": current_timestamp(), - } - - if not EvaluationCase.create(**case): - return False, "Failed to create test case" - - return True, case_id - except Exception as e: - logging.error(f"Error adding test case: {e}") - return False, str(e) - - @classmethod - def get_test_cases(cls, dataset_id: str) -> List[Dict[str, Any]]: - """Get all test cases for a dataset""" - try: - cases = EvaluationCase.select().where(EvaluationCase.dataset_id == dataset_id).order_by(EvaluationCase.create_time) - - return [c.to_dict() for c in cases] - except Exception as e: - logging.error(f"Error getting test cases for dataset {dataset_id}: {e}") - return [] - - @classmethod - def delete_test_case(cls, case_id: str) -> bool: - """Delete a test case""" - try: - return EvaluationCase.delete().where(EvaluationCase.id == case_id).execute() > 0 - except Exception as e: - logging.error(f"Error deleting test case {case_id}: {e}") - return False - - @classmethod - def import_test_cases(cls, dataset_id: str, cases: List[Dict[str, Any]]) -> Tuple[int, int]: - """ - Bulk import test cases from a list. - - Args: - dataset_id: Dataset ID - cases: List of test case dictionaries - - Returns: - (success_count, failure_count) - """ - success_count = 0 - failure_count = 0 - case_instances = [] - - if not cases: - return success_count, failure_count - - cur_timestamp = current_timestamp() - - try: - for case_data in cases: - case_id = get_uuid() - case_info = { - "id": case_id, - "dataset_id": dataset_id, - "question": case_data.get("question", ""), - "reference_answer": case_data.get("reference_answer"), - "relevant_doc_ids": case_data.get("relevant_doc_ids"), - "relevant_chunk_ids": case_data.get("relevant_chunk_ids"), - "metadata": case_data.get("metadata"), - "create_time": cur_timestamp, - } - - case_instances.append(EvaluationCase(**case_info)) - EvaluationCase.bulk_create(case_instances, batch_size=300) - success_count = len(case_instances) - failure_count = 0 - - except Exception as e: - logging.error(f"Error bulk importing test cases: {str(e)}") - failure_count = len(cases) - success_count = 0 - - return success_count, failure_count - - # ==================== Evaluation Execution ==================== - - @classmethod - def start_evaluation(cls, dataset_id: str, dialog_id: str, user_id: str, name: Optional[str] = None) -> Tuple[bool, str]: - """ - Start an evaluation run. - - Args: - dataset_id: Dataset ID - dialog_id: Dialog configuration to evaluate - user_id: User ID who starts the run - name: Optional run name - - Returns: - (success, run_id or error_message) - """ - try: - # Get dialog configuration - success, dialog = DialogService.get_by_id(dialog_id) - if not success: - return False, "Dialog not found" - - # Create evaluation run - run_id = get_uuid() - if not name: - name = f"Evaluation Run {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" - - run = { - "id": run_id, - "dataset_id": dataset_id, - "dialog_id": dialog_id, - "name": name, - "config_snapshot": dialog.to_dict(), - "metrics_summary": None, - "status": "RUNNING", - "created_by": user_id, - "create_time": current_timestamp(), - "complete_time": None, - } - - if not EvaluationRun.create(**run): - return False, "Failed to create evaluation run" - - # Execute evaluation asynchronously (in production, use task queue) - # For now, we'll execute synchronously - cls._execute_evaluation(run_id, dataset_id, dialog) - - return True, run_id - except Exception as e: - logging.error(f"Error starting evaluation: {e}") - return False, str(e) - - @classmethod - def _execute_evaluation(cls, run_id: str, dataset_id: str, dialog: Any): - """ - Execute evaluation for all test cases. - - This method runs the RAG pipeline for each test case and computes metrics. - """ - try: - # Get all test cases - test_cases = cls.get_test_cases(dataset_id) - - if not test_cases: - EvaluationRun.update(status="FAILED", complete_time=current_timestamp()).where(EvaluationRun.id == run_id).execute() - return - - # Execute each test case - results = [] - for case in test_cases: - result = cls._evaluate_single_case(run_id, case, dialog) - if result: - results.append(result) - - # Compute summary metrics - metrics_summary = cls._compute_summary_metrics(results) - - # Update run status - EvaluationRun.update(status="COMPLETED", metrics_summary=metrics_summary, complete_time=current_timestamp()).where(EvaluationRun.id == run_id).execute() - - except Exception as e: - logging.error(f"Error executing evaluation {run_id}: {e}") - EvaluationRun.update(status="FAILED", complete_time=current_timestamp()).where(EvaluationRun.id == run_id).execute() - - @classmethod - def _evaluate_single_case(cls, run_id: str, case: Dict[str, Any], dialog: Any) -> Optional[Dict[str, Any]]: - """ - Evaluate a single test case. - - Args: - run_id: Evaluation run ID - case: Test case dictionary - dialog: Dialog configuration - - Returns: - Result dictionary or None if failed - """ - try: - # Prepare messages - messages = [{"role": "user", "content": case["question"]}] - - # Execute RAG pipeline - start_time = timer() - answer = "" - retrieved_chunks = [] - - def _sync_from_async_gen(async_gen): - result_queue: queue.Queue = queue.Queue() - - def runner(): - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - async def consume(): - try: - async for item in async_gen: - result_queue.put(item) - except Exception as e: - result_queue.put(e) - finally: - result_queue.put(StopIteration) - - loop.run_until_complete(consume()) - loop.close() - - threading.Thread(target=runner, daemon=True).start() - - while True: - item = result_queue.get() - if item is StopIteration: - break - if isinstance(item, Exception): - raise item - yield item - - def chat(dialog, messages, stream=True, **kwargs): - from api.db.services.dialog_service import async_chat - - return _sync_from_async_gen(async_chat(dialog, messages, stream=stream, **kwargs)) - - for ans in chat(dialog, messages, stream=False): - if isinstance(ans, dict): - answer = ans.get("answer", "") - retrieved_chunks = ans.get("reference", {}).get("chunks", []) - break - else: - ans = {} - logging.warning( - "Evaluation case %s produced no answer from chat; token_usage will reflect empty output", - case.get("id", "unknown"), - ) - - execution_time = timer() - start_time - - # Compute metrics - metrics = cls._compute_metrics( - question=case["question"], - generated_answer=answer, - reference_answer=case.get("reference_answer"), - retrieved_chunks=retrieved_chunks, - relevant_chunk_ids=case.get("relevant_chunk_ids"), - dialog=dialog, - ) - - # Track token usage: use full prompt from async_chat when available. - # Note: Counts use tiktoken (cl100k_base), which matches OpenAI models but is an - # approximation for other providers (Anthropic, local models, etc.). Downstream - # consumers should treat these values as estimates for cost tracking. - full_prompt = ans.get("prompt", "") - if full_prompt: - prompt_tokens = num_tokens_from_string(full_prompt) - else: - logging.debug( - "Evaluation case %s: ans has no 'prompt' key; using question-only count (undercounts system + retrieved context)", - case.get("id", "unknown"), - ) - prompt_tokens = num_tokens_from_string(case.get("question", "") or "") - completion_tokens = num_tokens_from_string(answer or "") - token_usage = { - "prompt_tokens": prompt_tokens, - "completion_tokens": completion_tokens, - "total_tokens": prompt_tokens + completion_tokens, - } - - # Save result - result_id = get_uuid() - result = { - "id": result_id, - "run_id": run_id, - "case_id": case["id"], - "generated_answer": answer, - "retrieved_chunks": retrieved_chunks, - "metrics": metrics, - "execution_time": execution_time, - "token_usage": token_usage, - "create_time": current_timestamp(), - } - - EvaluationResult.create(**result) - - return result - except Exception as e: - logging.error(f"Error evaluating case {case.get('id')}: {e}") - return None - - @classmethod - def _compute_metrics( - cls, question: str, generated_answer: str, reference_answer: Optional[str], retrieved_chunks: List[Dict[str, Any]], relevant_chunk_ids: Optional[List[str]], dialog: Any - ) -> Dict[str, float]: - """ - Compute evaluation metrics for a single test case. - - Returns: - Dictionary of metric names to values - """ - metrics = {} - - # Retrieval metrics (if ground truth chunks provided) - if relevant_chunk_ids: - retrieved_ids = [c.get("chunk_id") for c in retrieved_chunks] - metrics.update(cls._compute_retrieval_metrics(retrieved_ids, relevant_chunk_ids)) - - # Generation metrics - if generated_answer: - # Basic metrics - metrics["answer_length"] = len(generated_answer) - metrics["has_answer"] = 1.0 if generated_answer.strip() else 0.0 - - # TODO: Implement advanced metrics using LLM-as-judge - # - Faithfulness (hallucination detection) - # - Answer relevance - # - Context relevance - # - Semantic similarity (if reference answer provided) - - return metrics - - @classmethod - def _compute_retrieval_metrics(cls, retrieved_ids: List[str], relevant_ids: List[str]) -> Dict[str, float]: - """ - Compute retrieval metrics. - - Args: - retrieved_ids: List of retrieved chunk IDs - relevant_ids: List of relevant chunk IDs (ground truth) - - Returns: - Dictionary of retrieval metrics - """ - if not relevant_ids: - return {} - - retrieved_set = set(retrieved_ids) - relevant_set = set(relevant_ids) - - # Precision: proportion of retrieved that are relevant - precision = len(retrieved_set & relevant_set) / len(retrieved_set) if retrieved_set else 0.0 - - # Recall: proportion of relevant that were retrieved - recall = len(retrieved_set & relevant_set) / len(relevant_set) if relevant_set else 0.0 - - # F1 score - f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0 - - # Hit rate: whether any relevant chunk was retrieved - hit_rate = 1.0 if (retrieved_set & relevant_set) else 0.0 - - # MRR (Mean Reciprocal Rank): position of first relevant chunk - mrr = 0.0 - for i, chunk_id in enumerate(retrieved_ids, 1): - if chunk_id in relevant_set: - mrr = 1.0 / i - break - - return {"precision": precision, "recall": recall, "f1_score": f1, "hit_rate": hit_rate, "mrr": mrr} - - @classmethod - def _compute_summary_metrics(cls, results: List[Dict[str, Any]]) -> Dict[str, Any]: - """ - Compute summary metrics across all test cases. - - Args: - results: List of result dictionaries - - Returns: - Summary metrics dictionary - """ - if not results: - return {} - - # Aggregate metrics - metric_sums = {} - metric_counts = {} - - for result in results: - metrics = result.get("metrics", {}) - for key, value in metrics.items(): - if isinstance(value, (int, float)): - metric_sums[key] = metric_sums.get(key, 0) + value - metric_counts[key] = metric_counts.get(key, 0) + 1 - - # Compute averages - summary = {"total_cases": len(results), "avg_execution_time": sum(r.get("execution_time", 0) for r in results) / len(results)} - - for key in metric_sums: - summary[f"avg_{key}"] = metric_sums[key] / metric_counts[key] - - return summary - - # ==================== Results & Analysis ==================== - - @classmethod - def get_run_results(cls, run_id: str) -> Dict[str, Any]: - """Get results for an evaluation run""" - try: - run = EvaluationRun.get_by_id(run_id) - if not run: - return {} - - results = EvaluationResult.select().where(EvaluationResult.run_id == run_id).order_by(EvaluationResult.create_time) - - return {"run": run.to_dict(), "results": [r.to_dict() for r in results]} - except Exception as e: - logging.error(f"Error getting run results {run_id}: {e}") - return {} - - @classmethod - def get_recommendations(cls, run_id: str) -> List[Dict[str, Any]]: - """ - Analyze evaluation results and provide configuration recommendations. - - Args: - run_id: Evaluation run ID - - Returns: - List of recommendation dictionaries - """ - try: - run = EvaluationRun.get_by_id(run_id) - if not run or not run.metrics_summary: - return [] - - metrics = run.metrics_summary - recommendations = [] - - # Low precision: retrieving irrelevant chunks - if metrics.get("avg_precision", 1.0) < 0.7: - recommendations.append( - { - "issue": "Low Precision", - "severity": "high", - "description": "System is retrieving many irrelevant chunks", - "suggestions": ["Increase similarity_threshold to filter out less relevant chunks", "Enable reranking to improve chunk ordering", "Reduce top_k to return fewer chunks"], - } - ) - - # Low recall: missing relevant chunks - if metrics.get("avg_recall", 1.0) < 0.7: - recommendations.append( - { - "issue": "Low Recall", - "severity": "high", - "description": "System is missing relevant chunks", - "suggestions": [ - "Increase top_k to retrieve more chunks", - "Lower similarity_threshold to be more inclusive", - "Enable hybrid search (keyword + semantic)", - "Check chunk size - may be too large or too small", - ], - } - ) - - # Slow response time - if metrics.get("avg_execution_time", 0) > 5.0: - recommendations.append( - { - "issue": "Slow Response Time", - "severity": "medium", - "description": f"Average response time is {metrics['avg_execution_time']:.2f}s", - "suggestions": ["Reduce top_k to retrieve fewer chunks", "Optimize embedding model selection", "Consider caching frequently asked questions"], - } - ) - - return recommendations - except Exception as e: - logging.error(f"Error generating recommendations for run {run_id}: {e}") - return [] diff --git a/test/unit_test/api/db/services/test_evaluation_service_token_usage.py b/test/unit_test/api/db/services/test_evaluation_service_token_usage.py deleted file mode 100644 index ecf17140f..000000000 --- a/test/unit_test/api/db/services/test_evaluation_service_token_usage.py +++ /dev/null @@ -1,170 +0,0 @@ -# -# Copyright 2026 The InfiniFlow Authors. All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use it except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -"""Unit tests for token_usage tracking in EvaluationService._evaluate_single_case.""" - -import pytest -from types import SimpleNamespace - -from api.db.services.evaluation_service import EvaluationService - - -@pytest.fixture -def mock_dialog(): - """Minimal dialog object for evaluation.""" - return SimpleNamespace( - kb_ids=["kb-1"], - prompt_config={"quote": True}, - llm_id="test-llm", - tenant_id="tenant-1", - ) - - -@pytest.fixture -def minimal_case(): - """Minimal test case for evaluation.""" - return { - "id": "case-1", - "question": "What is the capital of France?", - "reference_answer": None, - "relevant_chunk_ids": None, - } - - -@pytest.mark.p2 -def test_token_usage_structure_when_prompt_available(monkeypatch, mock_dialog, minimal_case): - """Verify token_usage dict has correct structure when ans contains 'prompt'.""" - captured_create = {} - - async def mock_async_chat(dialog, messages, stream, **kwargs): - # Simulate async_chat yielding one result with full prompt - yield { - "answer": "Paris is the capital of France.", - "reference": {"chunks": []}, - "prompt": "System instructions here.\n\nKnowledge: Some context.\n\nQuery: What is the capital of France?", - } - - def capture_create(**kwargs): - captured_create.update(kwargs) - - monkeypatch.setattr( - "api.db.services.dialog_service.async_chat", - mock_async_chat, - ) - monkeypatch.setattr( - "api.db.services.evaluation_service.EvaluationResult.create", - capture_create, - ) - - result = EvaluationService._evaluate_single_case("run-1", minimal_case, mock_dialog) - - assert result is not None - assert "token_usage" in captured_create - token_usage = captured_create["token_usage"] - assert "prompt_tokens" in token_usage - assert "completion_tokens" in token_usage - assert "total_tokens" in token_usage - assert token_usage["total_tokens"] == token_usage["prompt_tokens"] + token_usage["completion_tokens"] - assert token_usage["prompt_tokens"] > 0 - assert token_usage["completion_tokens"] > 0 - - -@pytest.mark.p2 -def test_token_usage_fallback_when_prompt_missing(monkeypatch, mock_dialog, minimal_case): - """Verify fallback to question-only count when ans has no 'prompt' key.""" - captured_create = {} - - async def mock_async_chat_no_prompt(dialog, messages, stream, **kwargs): - # Simulate response without 'prompt' (e.g. async_chat_solo) - yield { - "answer": "Paris.", - "reference": {"chunks": []}, - } - - def capture_create(**kwargs): - captured_create.update(kwargs) - - monkeypatch.setattr( - "api.db.services.dialog_service.async_chat", - mock_async_chat_no_prompt, - ) - monkeypatch.setattr( - "api.db.services.evaluation_service.EvaluationResult.create", - capture_create, - ) - - result = EvaluationService._evaluate_single_case("run-1", minimal_case, mock_dialog) - - assert result is not None - assert "token_usage" in captured_create - token_usage = captured_create["token_usage"] - assert "prompt_tokens" in token_usage - assert "completion_tokens" in token_usage - assert "total_tokens" in token_usage - assert token_usage["total_tokens"] == token_usage["prompt_tokens"] + token_usage["completion_tokens"] - # With fallback, prompt_tokens should reflect question only (smaller than full prompt) - assert token_usage["prompt_tokens"] >= 0 - assert token_usage["completion_tokens"] > 0 - - -@pytest.mark.p2 -def test_token_usage_no_answer_logs_warning(monkeypatch, mock_dialog, minimal_case, caplog): - """When chat yields no answers, we still record token_usage and log a warning.""" - captured_create = {} - - async def mock_async_chat_empty(dialog, messages, stream, **kwargs): - # Simulate async_chat that yields no items at all - if False: - yield {} - - def capture_create(**kwargs): - captured_create.update(kwargs) - - monkeypatch.setattr( - "api.db.services.dialog_service.async_chat", - mock_async_chat_empty, - ) - monkeypatch.setattr( - "api.db.services.evaluation_service.EvaluationResult.create", - capture_create, - ) - - with caplog.at_level("WARNING"): - result = EvaluationService._evaluate_single_case("run-1", minimal_case, mock_dialog) - - assert result is not None - token_usage = captured_create["token_usage"] - # No answer tokens in this case - assert token_usage["completion_tokens"] == 0 - assert token_usage["prompt_tokens"] >= 0 - assert token_usage["total_tokens"] == token_usage["prompt_tokens"] - assert any("produced no answer from chat" in msg for msg in caplog.messages) - - -@pytest.mark.p2 -def test_compute_summary_metrics_aggregates_metrics(): - """_compute_summary_metrics should average numeric metrics correctly.""" - results = [ - {"execution_time": 1.0, "metrics": {"precision": 0.5, "answer_length": 10}}, - {"execution_time": 3.0, "metrics": {"precision": 1.0, "answer_length": 20}}, - ] - - summary = EvaluationService._compute_summary_metrics(results) - - assert summary["total_cases"] == 2 - assert summary["avg_execution_time"] == pytest.approx(2.0) - assert summary["avg_precision"] == pytest.approx(0.75) - assert summary["avg_answer_length"] == pytest.approx(15.0)