Remove E402 noqa comments and restore pyproject.toml configuration in data_analysis_with_datalake directory

Author: marchhao

02-use-cases/data_analysis_with_datalake/agent.py

@@ -1,35 +1,37 @@
+# 导入所有必要的模块
 import os
-import json
+import sys
 import logging
 from pathlib import Path
-
 from dotenv import load_dotenv
+
+
+# 将当前目录添加到sys.path以便本地模块导入
+sys.path.append(str(Path(__file__).resolve().parent))
 # 加载 settings.txt（dotenv 格式）
-load_dotenv(dotenv_path=str(Path(__file__).resolve().parent / "settings.txt"), override=False)
+load_dotenv(
+    dotenv_path=str(Path(__file__).resolve().parent / "settings.txt"), override=False
+)
+
+# 导入veadk和agentkit相关模块
+from veadk import Agent, Runner  # noqa: E402
+from veadk.auth.veauth.ark_veauth import get_ark_token  # noqa: E402
+from veadk.memory.short_term_memory import ShortTermMemory  # noqa: E402
+from veadk.tools.builtin_tools.video_generate import video_generate  # noqa: E402
+from agentkit.apps import AgentkitAgentServerApp  # noqa: E402
+
+# 导入本地模块
+from tools.catalog_discovery import catalog_discovery  # noqa: E402
+from tools.duckdb_sql_execution import duckdb_sql_execution  # noqa: E402
+from tools.lancedb_hybrid_execution import lancedb_hybrid_execution  # noqa: E402
+from prompts import SYSTEM_PROMPT  # noqa: E402
 
-# Import get_ark_token and set MODEL_AGENT_API_KEY environment variable
-from veadk.auth.veauth.ark_veauth import get_ark_token
 # Check if MODEL_AGENT_API_KEY environment variable exists and is not empty
 if "MODEL_AGENT_API_KEY" not in os.environ or not os.environ["MODEL_AGENT_API_KEY"]:
     os.environ["MODEL_AGENT_API_KEY"] = get_ark_token()
 # Optionally assign to a variable for easier use in the file
 MODEL_AGENT_API_KEY = os.environ["MODEL_AGENT_API_KEY"]
 
-from veadk import Agent, Runner
-from veadk.a2a.agent_card import get_agent_card
-from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
-from agentkit.apps import AgentkitA2aApp
-
-import sys
-sys.path.append(str(Path(__file__).resolve().parent))
-from tools.catalog_discovery import catalog_discovery
-from tools.duckdb_sql_execution import duckdb_sql_execution
-from tools.lancedb_hybrid_execution import lancedb_hybrid_execution
-from prompts import SYSTEM_PROMPT
-from veadk.memory.short_term_memory import ShortTermMemory
-from veadk.tools.builtin_tools.video_generate import video_generate
-from agentkit.apps import AgentkitAgentServerApp
-
 short_term_memory = ShortTermMemory(backend="local")
 
 # 设置日志
@@ -41,10 +43,17 @@
 # --- Logging Configuration ---
 logger = logging.getLogger(__name__)
 
-tools = [catalog_discovery, duckdb_sql_execution, lancedb_hybrid_execution, video_generate]
+tools = [
+    catalog_discovery,
+    duckdb_sql_execution,
+    lancedb_hybrid_execution,
+    video_generate,
+]
 
 # 创建带记忆的 Agent
-model_name = os.getenv("MODEL_AGENT_NAME", "doubao-seed-1-6-251015")  # 默认使用更主流的豆包模型
+model_name = os.getenv(
+    "MODEL_AGENT_NAME", "doubao-seed-1-6-251015"
+)  # 默认使用更主流的豆包模型
 root_agent = Agent(
     description="基于LanceDB的数据检索Agent，支持结构化和向量查询。典型问题包括：1.你有哪些数据？2.给我一些样例数据？3.Ang Lee 评分超过7分的有哪些电影？4.Ang Lee 评分超过7分的电影中，有哪个电影海报中含有动物？5.Life of Pi 的电影海报，变成视频",
     instruction=SYSTEM_PROMPT,
@@ -71,8 +80,9 @@
 #     )
 
 agent_server_app = AgentkitAgentServerApp(
-    agent=root_agent, short_term_memory=short_term_memory,  
+    agent=root_agent,
+    short_term_memory=short_term_memory,
 )
 
 if __name__ == "__main__":
-    agent_server_app.run(host="0.0.0.0", port=8000)
+    agent_server_app.run(host="0.0.0.0", port=8000)

02-use-cases/data_analysis_with_datalake/client.py

@@ -1,7 +1,6 @@
 import requests
 import httpx
 import random
-import json
 
 from google.adk.cli.adk_web_server import CreateSessionRequest, RunAgentRequest
 from google.genai.types import Content, Part
@@ -12,36 +11,36 @@
     # Step 0: setup running configs
     app_name = "data_analysis_with_code"
     user_id = "agentkit_user"
-    session_id = f"agentkit_sample_session"
+    session_id = "agentkit_sample_session"
     base_url = ""
     api_key = ""
-    
-   
-    
-    task_num = 1   
-    
+
+    task_num = 1
 
     # Step 1: create a session
     def create_session():
         create_session_request = CreateSessionRequest(
-            session_id = session_id + f"_{random.randint(1, 9999)}",
+            session_id=session_id + f"_{random.randint(1, 9999)}",
         )
 
         response = requests.post(
             url=f"{base_url}/apps/{app_name}/users/{user_id}/sessions/{create_session_request.session_id}",
             headers={"Authorization": f"Bearer {api_key}"},
         )
-  
+
         print(f"[create session] Response from server: {response.json()}")
-    
+
         return create_session_request.session_id
 
     # Step 2: run agent with SSE
     run_agent_request = RunAgentRequest(
         app_name=app_name,
         user_id=user_id,
         session_id=create_session(),
-        new_message=Content(parts=[Part(text="Ang Lee的电影评分超过7分，有哪些电影海报包含动物")], role="user"),
+        new_message=Content(
+            parts=[Part(text="Ang Lee的电影评分超过7分，有哪些电影海报包含动物")],
+            role="user",
+        ),
         stream=True,
     )
 
@@ -50,18 +49,28 @@ def create_session():
     # 3. Handle streaming events
     async def send_request(message: str):
         run_agent_request = RunAgentRequest(
-        app_name=app_name,
-        user_id=user_id,
-        session_id=create_session(),
-        new_message=Content(parts=[Part(text=message)], role="user"),
-        stream=True,
+            app_name=app_name,
+            user_id=user_id,
+            session_id=create_session(),
+            new_message=Content(parts=[Part(text=message)], role="user"),
+            stream=True,
         )
 
-        with httpx.stream("POST", f"{base_url}/run_sse", json=run_agent_request.model_dump(exclude_none=True), timeout=120, headers={"Authorization": f"Bearer {api_key}"}) as r:
+        with httpx.stream(
+            "POST",
+            f"{base_url}/run_sse",
+            json=run_agent_request.model_dump(exclude_none=True),
+            timeout=120,
+            headers={"Authorization": f"Bearer {api_key}"},
+        ) as r:
             for line in r.iter_lines():
                 print(line)
-                
+
     async def send_request_parallel():
-        tasks = [send_request("Ang Lee的电影评分超过7分，有哪些电影海报包含动物") for _ in range(task_num)]
+        tasks = [
+            send_request("Ang Lee的电影评分超过7分，有哪些电影海报包含动物")
+            for _ in range(task_num)
+        ]
         await asyncio.gather(*tasks)
-    asyncio.run(send_request_parallel())
+
+    asyncio.run(send_request_parallel())

02-use-cases/data_analysis_with_datalake/prompts.py

@@ -1,6 +1,4 @@
-import os
-
-SYSTEM_PROMPT = '''
+SYSTEM_PROMPT = """
 ```你是一个火山引擎上基于 LanceDB + DuckDB + Doubao Vision 构建的数据检索专家，擅长依据用户自然语言问题，从 IMDB 数据集精准检索电影信息，以及进行多模态内容生成。
 你的核心任务是根据用户自然语言问题，从 IMDB 数据集检索电影信息，或进行多模态内容生成。
 
@@ -111,4 +109,4 @@
 - 语言表达专业、清晰，对每个步骤的描述准确明了。
 - 若使用工具，需明确写出工具名称及具体参数。
 ```
-'''
+"""

02-use-cases/data_analysis_with_datalake/tools/catalog_discovery.py

@@ -1,24 +1,27 @@
-import os
 import json
 
 from rich.console import Console
 
 # Import the LanceDBManager singleton
 from .lancedb_manager import lancedb_manager
+
 # Import utility functions
 from .utils import get_text_embedding as get_embedding
 
 console = Console()
 
+
 def catalog_discovery(query_intent: str) -> str:
     """Search metadata using vector similarity based on the user's intent keywords."""
     console.print(f"[catalog_discovery] Inputs: query_intent={query_intent!r}")
 
     if not query_intent:
-        return json.dumps({
-            "status": "error",
-            "error": "Query intent is empty. Please provide a keyword to search."
-        })
+        return json.dumps(
+            {
+                "status": "error",
+                "error": "Query intent is empty. Please provide a keyword to search.",
+            }
+        )
 
     tbl, error_msg = lancedb_manager.get_metadata_table()
     if error_msg:
@@ -31,20 +34,24 @@ def catalog_discovery(query_intent: str) -> str:
             return json.dumps({"error": emb_err})
 
         # 调用Lance进行检索
-        results_df = tbl.search(query_vector, vector_column_name="vector").limit(10).to_pandas()
+        results_df = (
+            tbl.search(query_vector, vector_column_name="vector").limit(10).to_pandas()
+        )
         records = results_df.to_dict("records")
 
         # Remove the vector column from the records before returning to the agent
         for record in records:
             record.pop("vector", None)
 
         console.print(f"✅ 检索到 {len(records)} 条相关元数据")
-        return json.dumps({
-            "status": "ok",
-            "records": records,
-            "meta": {"row_count": len(records)},
-            "echo": {"query_intent": query_intent}
-        })
+        return json.dumps(
+            {
+                "status": "ok",
+                "records": records,
+                "meta": {"row_count": len(records)},
+                "echo": {"query_intent": query_intent},
+            }
+        )
     except Exception as e:
         error_msg = f"❌ 检索失败: {e}"
         console.print(f"[red]{error_msg}[/red]")

02-use-cases/data_analysis_with_datalake/tools/duckdb_sql_execution.py

@@ -1,4 +1,3 @@
-import os
 import json
 
 from rich.console import Console
@@ -8,21 +7,24 @@
 
 console = Console()
 
+
 def duckdb_sql_execution(sql: str, user_question: str = "") -> str:
     """Execute structured SQL via DuckDB on Lance table contents.
 
     Expect sql to be a direct SQL string:
     "SELECT ..."
     """
-    console.print(f"[duckdb_sql_execution] Inputs: sql={sql!r}, user_question={user_question!r}")
+    console.print(
+        f"[duckdb_sql_execution] Inputs: sql={sql!r}, user_question={user_question!r}"
+    )
     if not sql or not isinstance(sql, str):
         return json.dumps({"error": "SQL 字符串缺失或类型错误"}, ensure_ascii=False)
 
     # Open the table using the LanceDBManager
     tbl, err = lancedb_manager.open_table()
     if err:
         return json.dumps({"error": err}, ensure_ascii=False)
-    
+
     view_name = "imdb_top_1000"
 
     # Register Arrow/Pandas to DuckDB
@@ -56,6 +58,6 @@ def duckdb_sql_execution(sql: str, user_question: str = "") -> str:
         "meta": {
             "row_count": len(records),
             "table": view_name,
-        }
+        },
     }
-    return json.dumps(result, ensure_ascii=False)
+    return json.dumps(result, ensure_ascii=False)

02-use-cases/data_analysis_with_datalake/tools/lancedb_hybrid_execution.py

@@ -1,20 +1,24 @@
-import os
 import json
 from typing import Optional
 
 from rich.console import Console
 import pandas as pd
 
-console = Console()
-
 # Import the LanceDBManager singleton
 from .lancedb_manager import lancedb_manager
+
 # Import utility functions
 from .utils import get_multimodal_text_vector as _get_text_vector
 
+console = Console()
+
 
-def lancedb_hybrid_execution(query_text: str, filters: str = "", select: Optional[list] = None, limit: int = 10) -> str:
-    console.print(f"[lancedb_hybrid_execution] Inputs: query_text={query_text!r}, filters={filters!r}")
+def lancedb_hybrid_execution(
+    query_text: str, filters: str = "", select: Optional[list] = None, limit: int = 10
+) -> str:
+    console.print(
+        f"[lancedb_hybrid_execution] Inputs: query_text={query_text!r}, filters={filters!r}"
+    )
 
     # open table
     tbl, err = lancedb_manager.open_table()
@@ -50,13 +54,19 @@ def lancedb_hybrid_execution(query_text: str, filters: str = "", select: Optiona
             df_norm.columns = header
             records_obj = df_norm.to_dict(orient="records")
         except Exception:
-            records_obj = [{header[i]: row[i] for i in range(len(header))} for row in df.values.tolist()]
+            records_obj = [
+                {header[i]: row[i] for i in range(len(header))}
+                for row in df.values.tolist()
+            ]
         records = df.values.tolist()
-        return json.dumps({
-            "status": "ok",
-            "data": [header] + records,
-            "records": records_obj,
-            "meta": {"row_count": len(records)}
-        }, ensure_ascii=False)
+        return json.dumps(
+            {
+                "status": "ok",
+                "data": [header] + records,
+                "records": records_obj,
+                "meta": {"row_count": len(records)},
+            },
+            ensure_ascii=False,
+        )
     except Exception as e:
-        return json.dumps({"error": f"混合检索失败: {e}"}, ensure_ascii=False)
+        return json.dumps({"error": f"混合检索失败: {e}"}, ensure_ascii=False)