for trigger

Author: chaochao0822

01-tutorials/workshop/session3/E3_lark_doc/agent.py

@@ -18,7 +18,10 @@
 import json
 
 import lark_oapi as lark
-from lark_oapi.api.docx.v1 import *
+from lark_oapi.api.docx.v1 import (
+    RawContentDocumentRequest,
+    RawContentDocumentResponse,
+)
 from lark_oapi.core.model import RequestOption
 from veadk.integrations.ve_identity import (
     VeIdentityFunctionTool,
@@ -32,6 +35,7 @@
 
 short_term_memory = ShortTermMemory(backend="local")
 
+
 async def lark_document_query(document_id: str, *, access_token: str) -> str:
     """
     查询飞书文档内容
@@ -83,6 +87,7 @@ async def lark_document_query(document_id: str, *, access_token: str) -> str:
     ),
 )
 
+
 async def clean_state(args: dict[str, Any], *, tool_context: ToolContext) -> None:
     """Clean user's Oauth identity state.
 

01-tutorials/workshop/session3/E4_volc_ops/agent.py

@@ -1,4 +1,3 @@
-
 from typing import Any
 from google.adk.planners import BuiltInPlanner
 from google.adk.tools.mcp_tool.mcp_toolset import (
@@ -32,6 +31,8 @@
         timeout=30.0,
     ),
 )
+
+
 async def clean_state(args: dict[str, Any], *, tool_context: ToolContext) -> None:
     """Clean user's Oauth identity state.
 

01-tutorials/workshop/session3/E6a_mail_ast_with_guard/agent.py

@@ -23,10 +23,16 @@
 from typing import Optional
 
 from agentkit.apps import AgentkitSimpleApp
-from veadk.prompts.agent_default_prompt import DEFAULT_DESCRIPTION, DEFAULT_INSTRUCTION
 
 from .colors import print_agent_permission
-from .tools import read_inbox, read_email, classify_email, forward_email, generate_report
+from .tools import (
+    read_inbox,
+    read_email,
+    classify_email,
+    forward_email,
+    generate_report,
+)
+
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -35,91 +41,101 @@
 app_name = "simple_app"
 model_name = "deepseek-v3-250324"
 
-class BeforeModelPermissionCallback():
+
+class BeforeModelPermissionCallback:
     """模型调用前的权限检查回调"""
 
     def __init__(self, service_url, api_key):
         self.service_url = service_url
         self.api_key = api_key
 
-    def __call__(self, callback_context: CallbackContext, llm_request: LlmRequest) -> Optional[LlmResponse]:
+    def __call__(
+        self, callback_context: CallbackContext, llm_request: LlmRequest
+    ) -> Optional[LlmResponse]:
         """处理模型调用前的权限检查"""
         try:
             payload = {
-                'session_id': callback_context.session.id,
-                'llm_request': llm_request.model_dump()
+                "session_id": callback_context.session.id,
+                "llm_request": llm_request.model_dump(),
             }
 
             headers = {
                 "Authorization": f"Bearer {self.api_key}",
                 "Content-Type": "application/json"
             }
 
-            resp = requests.post(f"{self.service_url}/before_check", json=payload, headers=headers, timeout=200)
+            resp = requests.post(
+                f"{self.service_url}/before_check",
+                json=payload,
+                headers=headers,
+                timeout=200,
+            )
             resp.raise_for_status()
 
             resp_json = resp.json()
-            status = resp_json.get('status', '')
-            data = resp_json.get('response', {})
+            status = resp_json.get("status", "")
+            data = resp_json.get("response", {})
 
             # 记录权限信息
-            if 'permissions' in data:
-                print_agent_permission(data.get('permissions', []))
+            if "permissions" in data:
+                print_agent_permission(data.get("permissions", []))
 
             # 检查权限结果
-            if status != 'success':
+            if status != "success":
                 error_msg = "智能体行为异常，流程终止"
                 return LlmResponse(
                     content=Content(
                         role="model",
-                        parts=[
-                            Part(text=error_msg)
-                        ],
+                        parts=[Part(text=error_msg)],
                     )
                 )
             return None
-        except requests.exceptions.RequestException as e:
+        except requests.exceptions.RequestException:
             return None
-        except Exception as e:
+        except Exception:
             return None
 
 
-class AfterModelPermissionCallback():
+class AfterModelPermissionCallback:
     """模型调用后的权限检查回调"""
 
     def __init__(self, service_url, api_key):
         self.service_url = service_url
         self.api_key = api_key
 
-    def __call__(self, callback_context: CallbackContext, llm_response: LlmResponse) -> LlmResponse | None:
+    def __call__(
+        self, callback_context: CallbackContext, llm_response: LlmResponse
+    ) -> LlmResponse | None:
         """处理模型调用后的权限检查"""
         try:
             payload = {
-                'session_id': callback_context.session.id,
-                'llm_response': llm_response.model_dump()
+                "session_id": callback_context.session.id,
+                "llm_response": llm_response.model_dump(),
             }
             headers = {
                 "Authorization": f"Bearer {self.api_key}",
-                "Content-Type": "application/json"
+                "Content-Type": "application/json",
             }
 
-            resp = requests.post(f"{self.service_url}/check", json=payload, headers=headers, timeout=200)
+            resp = requests.post(
+                f"{self.service_url}/check", json=payload, headers=headers, timeout=200
+            )
             resp.raise_for_status()
 
             resp_json = resp.json()
-            status = resp_json.get('status', '')
-            data = resp_json.get('response', {})
+            status = resp_json.get("status", "")
+            data = resp_json.get("response", {})
 
             # 记录权限信息
             if not llm_response.custom_metadata:
                 llm_response.custom_metadata = {}
             llm_response.custom_metadata.update(data)
 
-            if 'permissions' in data:
-                print_agent_permission(data.get('permissions', []))
+            if "permissions" in data:
+                print_agent_permission(data.get("permissions", []))
 
             # 检查权限结果
-            if status != 'success':
+            if status != "success":
                 error_msg = "智能体行为异常，流程终止"
                 llm_response.content.parts = [Part(text=error_msg)]
 
@@ -133,13 +149,7 @@ def __call__(self, callback_context: CallbackContext, llm_response: LlmResponse)
             return llm_response
 
 
-tools = [
-    read_inbox,
-    read_email,
-    classify_email,
-    forward_email,
-    generate_report
-]
+tools = [read_inbox, read_email, classify_email, forward_email, generate_report]
 
 system_instruction = """
 你是一名企业级邮件助手智能体，核心职责是协助用户高效处理收件箱的各类业务事项，深度解析邮件中的业务诉求与落地需求，通过工具执行保障内外部业务沟通闭环，确保重要事务不遗漏。
@@ -181,13 +191,19 @@ def __call__(self, callback_context: CallbackContext, llm_response: LlmResponse)
 )
 
 # 设置权限围栏
-adaptive_permission_service_url = "http://sd4i3neu6omp034ocgsm0.apigateway-cn-beijing.volceapi.com"
+adaptive_permission_service_url = (
+    "http://sd4i3neu6omp034ocgsm0.apigateway-cn-beijing.volceapi.com"
+)
 adaptive_permission_api_key = os.getenv("ADAPTIVE_PERMISSION_SERVICE_KEY")
 
 if adaptive_permission_api_key:
     logger.info("权限围栏已开启")
-    agent.before_model_callback=BeforeModelPermissionCallback(adaptive_permission_service_url, adaptive_permission_api_key)
-    agent.after_model_callback=AfterModelPermissionCallback(adaptive_permission_service_url, adaptive_permission_api_key)
+    agent.before_model_callback=BeforeModelPermissionCallback(
+        adaptive_permission_service_url, adaptive_permission_api_key
+    )
+    agent.after_model_callback=AfterModelPermissionCallback(
+        adaptive_permission_service_url, adaptive_permission_api_key
+    )
 else:
     logger.warning("权限围栏未开启")
 
@@ -207,7 +223,9 @@ async def run(payload: dict, headers: dict) -> str:
     logger.info(
         f"Running agent with prompt: {prompt}, user_id: {user_id}, session_id: {session_id}"
     )
-    response = await runner.run(messages=prompt, user_id=user_id, session_id=session_id)  # 请勿修改此行，不要使用sse模式
+    response = await runner.run(
+        messages=prompt, user_id=user_id, session_id=session_id
+    )  # 请勿修改此行，不要使用sse模式
 
     logger.info(f"Run response: {response}")
 

01-tutorials/workshop/session3/E6a_mail_ast_with_guard/colors.py

@@ -1,13 +1,13 @@
 class Colors:
-    HEADER = '\033[37m'
-    BLUE = '\033[94m'
-    CYAN = '\033[96m'
-    GREEN = '\033[92m'
-    YELLOW = '\033[93m'
-    RED = '\033[91m'
-    ENDC = '\033[0m'
-    BOLD = '\033[1m'
-    UNDERLINE = '\033[4m'
+    HEADER = "\033[37m"
+    BLUE = "\033[94m"
+    CYAN = "\033[96m"
+    GREEN = "\033[92m"
+    YELLOW = "\033[93m"
+    RED = "\033[91m"
+    ENDC = "\033[0m"
+    BOLD = "\033[1m"
+    UNDERLINE = "\033[4m"
 
 def print_agent_reflect(text: str):
     """打印智能体反思"""
@@ -28,9 +28,9 @@ def print_agent_permission(texts: list):
 
 def print_header(text: str):
     """打印标题"""
-    print(f"\n{Colors.HEADER}{Colors.BOLD}{'='*60}{Colors.ENDC}")
+    print(f"\n{Colors.HEADER}{Colors.BOLD}{'=' * 60}{Colors.ENDC}")
     print(f"{Colors.HEADER}{Colors.BOLD}{text.center(60)}{Colors.ENDC}")
-    print(f"{Colors.HEADER}{Colors.BOLD}{'='*60}{Colors.ENDC}\n")
+    print(f"{Colors.HEADER}{Colors.BOLD}{'=' * 60}{Colors.ENDC}\n")
 
 def print_user_request(texts: list):
     """打印用户请求"""

01-tutorials/workshop/session3/E6a_mail_ast_with_guard/tools.py

@@ -4,8 +4,15 @@
 
 # Mock数据结构
 class Email:
-    def __init__(self, id: str, sender: str, subject: str, body: str,
-                 received_date: str, priority: str = "normal"):
+    def __init__(
+        self,
+        id: str,
+        sender: str,
+        subject: str,
+        body: str,
+        received_date: str,
+        priority: str = "normal"
+    ):
         self.id = id
         self.sender = sender
         self.subject = subject
@@ -20,16 +27,16 @@ def to_dict(self):
             "subject": self.subject,
             "body": self.body,
             "received_date": self.received_date,
-            "priority": self.priority
+            "priority": self.priority,
         }
 
 # 从JSON文件加载邮件数据
 # 获取当前文件的目录
 current_dir = os.path.dirname(os.path.abspath(__file__))
 # 构建emails.json的绝对路径
-email_file_path = os.path.join(current_dir, 'emails.json')
+email_file_path = os.path.join(current_dir, "emails.json")
 
-with open(email_file_path, 'r', encoding='utf-8') as f:
+with open(email_file_path, "r", encoding="utf-8") as f:
     email_data = json.load(f)
 
 def read_inbox(mailbox: str, unread_only: bool):
@@ -42,17 +49,19 @@ def read_inbox(mailbox: str, unread_only: bool):
     # 根据mailbox过滤邮件
     owner_emails = [Email(**email) for email in email_data.get(mailbox, [])]
 
-    emails = [{
-        "id": email.id,
-        "sender": email.sender,
-        "subject": email.subject
-    } for email in owner_emails]
+    emails = [
+        {
+            "id": email.id,
+            "sender": email.sender,
+            "subject": email.subject
+        } for email in owner_emails
+    ]
 
     result = {
         "success": True,
         "count": len(emails),
         "emails": emails,
-        "message": f"成功读取 {len(emails)} 封邮件基本信息"
+        "message": f"成功读取 {len(emails)} 封邮件基本信息",
     }
 
     return result
@@ -87,7 +96,7 @@ def classify_email(email_text: str, keywords: str):
     result = {
         "success": True,
         "classification": classification,
-        "message": f"成功分类邮件为 {classification} 优先级"
+        "message": f"成功分类邮件为 {classification} 优先级",
     }
 
     return result
@@ -132,13 +141,13 @@ def generate_report(total: int, forwarded: int, receipient: str):
     # 生成报告
     report = f"""
 ===== 邮件处理执行报告 =====
-生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
+生成时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
 
 📊 处理统计:
   - 总邮件数: {total} 封
   - 转发邮件数: {forwarded} 封
   - 目标邮箱: {receipient}
-  - 执行状态: {'✅ 成功' if success else '❌ 失败'}
+  - 执行状态: {"✅ 成功" if success else "❌ 失败"}
 """
 
     result = {
@@ -148,19 +157,19 @@ def generate_report(total: int, forwarded: int, receipient: str):
             "total_emails": total,
             "forwarded_count": forwarded,
             "target_email": receipient,
-            "execution_success": success
+            "execution_success": success,
         }
     }
 
     return result
 
 class Colors:
-    HEADER = '\033[95m'
-    BLUE = '\033[94m'
-    CYAN = '\033[96m'
-    GREEN = '\033[92m'
-    YELLOW = '\033[93m'
-    RED = '\033[91m'
-    ENDC = '\033[0m'
-    BOLD = '\033[1m'
-    UNDERLINE = '\033[4m'
+    HEADER = "\033[37m"
+    BLUE = "\033[94m"
+    CYAN = "\033[96m"
+    GREEN = "\033[92m"
+    YELLOW = "\033[93m"
+    RED = "\033[91m"
+    ENDC = "\033[0m"
+    BOLD = "\033[1m"
+    UNDERLINE = "\033[4m"