{ "cells": [ { "cell_type": "code", "execution_count": 25, "id": "173faf58", "metadata": {}, "outputs": [], "source": [ "import os\n", "from langchain.chat_models import init_chat_model\n", "from langchain.tools import tool\n", "from dotenv import load_dotenv\n", "\n", "load_dotenv()\n", "\n", "\n", "# 使用 DeepSeek 模型，通过阿里云百炼平台接入\n", "# init_chat_model 会自动适配 OpenAI 兼容接口\n", "model = init_chat_model(\n", " model=\"deepseek-v4-flash\",\n", " model_provider=\"openai\",\n", " base_url=\"https://api.deepseek.com\",\n", " api_key=os.getenv('OPENAI_API_KEY'),\n", ")\n", "\n", "# ---- 模拟数据 ----\n", "# 实际项目中，这些数据来自数据库或 API\n", "EXPORT_RIGHTS = {\n", " \"zhangsan\": {\"role\": \"finance\", \"region\": \"all\"},\n", " \"lisi\": {\"role\": \"operation\", \"region\": \"east\"},\n", "}\n", "\n", "\n", "@tool\n", "def check_export_permission(username: str) -> dict:\n", " \"\"\"查询用户是否有报表导出权限。\n", " 参数 username 为员工账号（英文名）。\n", " \"\"\"\n", " user_info = EXPORT_RIGHTS.get(username)\n", " if not user_info:\n", " return {\"username\": username, \"can_export\": False, \"reason\": \"用户不存在\"}\n", " return {\n", " \"username\": username,\n", " \"role\": user_info[\"role\"],\n", " \"region\": user_info[\"region\"],\n", " \"can_export\": user_info[\"role\"] in (\"finance\", \"ops_manager\"),\n", " }\n", "\n", "\n", "@tool\n", "def create_export_task(\n", " report_name: str, file_format: str, estimated_rows: int, reason: str\n", ") -> dict:\n", " \"\"\"创建报表导出任务。\n", " report_name: 报表名称\n", " file_format: 导出格式 (xlsx / csv)\n", " estimated_rows: 预计导出行数\n", " reason: 导出原因\n", " \"\"\"\n", " return {\n", " \"task_id\": \"EXPORT-20260706-001\",\n", " \"report_name\": report_name,\n", " \"file_format\": file_format,\n", " \"estimated_rows\": estimated_rows,\n", " \"reason\": reason,\n", " \"status\": \"queued\",\n", " }\n", "\n", "\n", "tools = [check_export_permission, create_export_task]" ] }, { "cell_type": "code", "execution_count": 26, "id": "73293168", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[审计] 准备调用模型，当前上下文消息数：1\n", "[审计] 开始执行工具：check_export_permission\n", "[审计] 工具执行完毕：check_export_permission\n", "[审计] 准备调用模型，当前上下文消息数：3\n", "抱歉，经查询，您的账号 **lisi**（运营角色）目前**没有报表导出权限**，因此无法为您创建导出任务。\n", "\n", "建议您联系上级或管理员开通导出权限，如有需要，我可再为您尝试创建导出任务。\n" ] } ], "source": [ "from langchain.agents import create_agent\n", "from langchain.agents.middleware import before_model, wrap_tool_call, AgentState\n", "\n", "\n", "@before_model\n", "def log_before_model(state: AgentState, runtime):\n", " \"\"\"模型调用前执行：看一眼当前上下文里有多少条消息。\"\"\"\n", " # state[\"messages\"] 是 Agent 当前累积的全部对话历史\n", " # runtime 携带运行时环境信息（后面会详细讲）\n", " print(f\"[审计] 准备调用模型，当前上下文消息数：{len(state['messages'])}\")\n", " return None # 返回 None 表示不做任何修改\n", "\n", "\n", "@wrap_tool_call\n", "def log_tool_call(request, handler):\n", " \"\"\"工具调用前后各打一条日志，并记录耗时。\"\"\"\n", " tool_name = request.tool_call[\"name\"]\n", " print(f\"[审计] 开始执行工具：{tool_name}\")\n", "\n", " # handler(request) 是\"真正执行工具\"的入口\n", " # 不调用它，工具就不会执行\n", " result = handler(request)\n", "\n", " print(f\"[审计] 工具执行完毕：{tool_name}\")\n", " return result\n", "\n", "\n", "# 组装 Agent，把 Middleware 列表传进去\n", "agent = create_agent(\n", " model=model,\n", " tools=tools,\n", " middleware=[\n", " log_before_model, # 排在前面：先记录状态\n", " log_tool_call, # 排在后面：包裹工具调用\n", " ],\n", " system_prompt=(\n", " \"你是报表导出平台的智能助手。\"\n", " \"用户询问导出相关问题前，先调用 check_export_permission 确认权限。\"\n", " \"只有在用户明确提出导出需求时，才调用 create_export_task 创建任务。\"\n", " ),\n", ")\n", "\n", "# 跑一次看看\n", "response = agent.invoke({\n", " \"messages\": [\n", " {\n", " \"role\": \"user\",\n", " \"content\": \"我是 lisi，需要导出本月 east 区域的订单报表，大约 5000 行，xlsx 格式。\",\n", " }\n", " ]\n", "})\n", "\n", "print(response[\"messages\"][-1].content)" ] }, { "cell_type": "code", "execution_count": 27, "id": "a2b4a168", "metadata": {}, "outputs": [], "source": [ "from typing import TypedDict\n", "from typing_extensions import NotRequired\n", "from langchain.agents.middleware import AgentState, before_model, after_model\n", "\n", "\n", "class ExportAgentState(AgentState):\n", " \"\"\"扩展默认状态，增加模型调用计数和大任务标记。\"\"\"\n", " # AgentState 已经自带 messages 字段，这里只加新字段\n", " # NotRequired 表示可以不传，Middleware 内部自己维护\n", " model_call_count: NotRequired[int]\n", " blocked_requests: NotRequired[int] # 被拦截的请求次数\n", "\n", "\n", "@before_model(state_schema=ExportAgentState)\n", "def audit_before_model(state: ExportAgentState, runtime):\n", " \"\"\"每次调模型前，看一眼当前统计。\"\"\"\n", " count = state.get(\"model_call_count\", 0)\n", " blocked = state.get(\"blocked_requests\", 0)\n", " print(f\"[统计] 第 {count + 1} 次调模型 | 已拦截 {blocked} 次\")\n", " return None\n", "\n", "\n", "@after_model(state_schema=ExportAgentState)\n", "def update_stats(state: ExportAgentState, runtime):\n", " \"\"\"模型返回后，把计数器 +1。\"\"\"\n", " # after_model 返回 dict 可以直接更新 AgentState\n", " return {\"model_call_count\": state.get(\"model_call_count\", 0) + 1}" ] }, { "cell_type": "code", "execution_count": 28, "id": "f8f3f0ba", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "[统计] 第 1 次调模型 | 已拦截 0 次\n", "[上下文] request_id=req-20260706-001 user_role=finance tenant=t-1234\n" ] } ], "source": [ "from langgraph.runtime import Runtime\n", "\n", "\n", "class RunContext(TypedDict):\n", " \"\"\"定义 runtime.context 的结构。\"\"\"\n", " request_id: str # 用于日志追踪\n", " user_role: str # 当前用户的角色\n", " tenant_id: str # 租户标识（多租户场景）\n", "\n", "\n", "@before_model(state_schema=ExportAgentState)\n", "def inject_context(state: ExportAgentState, runtime: Runtime[RunContext]):\n", " \"\"\"从 runtime.context 读取业务信息，用于日志关联。\"\"\"\n", " ctx = runtime.context or {}\n", " print(\n", " f\"[上下文] request_id={ctx.get('request_id')} \"\n", " f\"user_role={ctx.get('user_role')} \"\n", " f\"tenant={ctx.get('tenant_id')}\"\n", " )\n", " return None\n", "\n", "\n", "# 创建 Agent 时声明 context 结构\n", "agent = create_agent(\n", " model=model,\n", " tools=tools,\n", " middleware=[audit_before_model, inject_context, update_stats],\n", " state_schema=ExportAgentState, # 声明自定义状态\n", " context_schema=RunContext, # 声明上下文结构\n", " system_prompt=\"你是报表导出平台的智能助手。\",\n", ")\n", "\n", "# 调用时传入 context\n", "result = agent.invoke(\n", " {\"messages\": [{\"role\": \"user\", \"content\": \"导出本月的订单报表\"}]},\n", " context={\n", " \"request_id\": \"req-20260706-001\",\n", " \"user_role\": \"finance\",\n", " \"tenant_id\": \"t-1234\",\n", " },\n", ")" ] }, { "cell_type": "code", "execution_count": 9, "id": "cc329b2b", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "content='帮我查 task_001 的导出进度，持续查到完成为止。' additional_kwargs={} response_metadata={} id='25155520-2d96-46b6-a9fe-91c35b45e2be'\n", "content='好的，我来查询 task_001 的导出进度。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 105, 'prompt_tokens': 324, 'total_tokens': 429, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 44, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 68}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '0a00c018-8ce2-40c1-9992-f36eb1e5f022', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019f4b76-f297-7a32-8f80-0eb6e77c963a-0' tool_calls=[{'name': 'check_export_progress', 'args': {'task_id': 'task_001'}, 'id': 'call_00_55MBKV1ikA8ad8rOVdVt3155', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 324, 'output_tokens': 105, 'total_tokens': 429, 'input_token_details': {'cache_read': 256}, 'output_token_details': {'reasoning': 44}}\n", "content='第 1 次查询：任务仍在处理中，请稍候。' name='check_export_progress' id='b6c8dcdf-2aa2-4fe3-acc0-578ad22ee443' tool_call_id='call_00_55MBKV1ikA8ad8rOVdVt3155'\n", "content='任务还在处理中，我继续查询。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 75, 'prompt_tokens': 456, 'total_tokens': 531, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 17, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 384}, 'prompt_cache_hit_tokens': 384, 'prompt_cache_miss_tokens': 72}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': 'e6caac75-4ef9-4e30-9361-91f84524afc8', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019f4b76-f9ef-75e0-8c28-cca2787aac80-0' tool_calls=[{'name': 'check_export_progress', 'args': {'task_id': 'task_001'}, 'id': 'call_00_igmkFCeqOtSwKLz5VtmW6480', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 456, 'output_tokens': 75, 'total_tokens': 531, 'input_token_details': {'cache_read': 384}, 'output_token_details': {'reasoning': 17}}\n", "content='第 2 次查询：任务仍在处理中，请稍候。' name='check_export_progress' id='b69a2967-256b-4dac-8948-c5c886ff067b' tool_call_id='call_00_igmkFCeqOtSwKLz5VtmW6480'\n", "content='' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 57, 'prompt_tokens': 558, 'total_tokens': 615, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 7, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 512}, 'prompt_cache_hit_tokens': 512, 'prompt_cache_miss_tokens': 46}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': 'dc57c2ab-7c18-4d74-ade6-0030ce7ce152', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019f4b76-fe0e-7860-a4aa-ef77c500a8bf-0' tool_calls=[{'name': 'check_export_progress', 'args': {'task_id': 'task_001'}, 'id': 'call_00_yyOXS7ED8ur9gytRrzX64854', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 558, 'output_tokens': 57, 'total_tokens': 615, 'input_token_details': {'cache_read': 512}, 'output_token_details': {'reasoning': 7}}\n", "content='第 3 次查询：任务仍在处理中，请稍候。' name='check_export_progress' id='9858f2f8-6ee9-4632-bc8c-663dec13e776' tool_call_id='call_00_yyOXS7ED8ur9gytRrzX64854'\n", "content='Model call limits exceeded: run limit (3/3)' additional_kwargs={} response_metadata={} id='87be7ee0-c941-4559-8833-537356a25d91' tool_calls=[] invalid_tool_calls=[]\n" ] } ], "source": [ "from langchain.agents.middleware import ModelCallLimitMiddleware\n", "\n", "\n", "# 模拟一个\"前 4 次返回处理中，第 5 次返回完成\"的进度查询工具\n", "export_progress = {\"task_001\": 0}\n", "\n", "\n", "@tool\n", "def check_export_progress(task_id: str) -> str:\n", " \"\"\"查询导出任务的进度。\"\"\"\n", " export_progress[task_id] = export_progress.get(task_id, 0) + 1\n", " attempt = export_progress[task_id]\n", " if attempt < 5:\n", " return f\"第 {attempt} 次查询：任务仍在处理中，请稍候。\"\n", " return f\"第 {attempt} 次查询：导出完成，文件已生成。\"\n", "\n", "\n", "# 不加限制：Agent 会一直查到第 5 次\n", "# 加上 ModelCallLimitMiddleware(run_limit=3)：最多调 3 次模型就强制结束\n", "agent = create_agent(\n", " model=model,\n", " tools=[check_export_progress],\n", " middleware=[\n", " ModelCallLimitMiddleware(\n", " run_limit=3, # 单次 invoke 最多调 3 次模型\n", " exit_behavior=\"end\", # 到达上限后尝试优雅结束（生成总结）\n", " ),\n", " ],\n", " system_prompt=(\n", " \"你是导出任务进度查询助手。\"\n", " \"当用户查询任务进度时，调用 check_export_progress。\"\n", " \"如果任务还在处理中，继续查询直到完成。\"\n", " ),\n", ")\n", "\n", "result = agent.invoke({\n", " \"messages\": [{\"role\": \"user\", \"content\": \"帮我查 task_001 的导出进度，持续查到完成为止。\"}]\n", "})\n", "# 输出类似：\"Model call limits exceeded: run limit (3/3)\"\n", "# Agent 被强制刹车，不会无限循环\n", "for message in result['messages']:\n", " print(message)" ] }, { "cell_type": "code", "execution_count": 10, "id": "ea688334", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Agent 已暂停，等待审批...\n" ] } ], "source": [ "from typing import Literal\n", "from pydantic import BaseModel, Field\n", "from langchain.agents.middleware import HumanInTheLoopMiddleware\n", "from langgraph.checkpoint.memory import InMemorySaver\n", "\n", "\n", "# 第一步：用 Pydantic 定义工具的输入结构\n", "# 结构化的参数让模型生成更准确，也方便人工审批时查看\n", "class ExportTaskInput(BaseModel):\n", " \"\"\"导出任务的参数结构。模型会按照这个 schema 生成参数。\"\"\"\n", " report_name: str = Field(description=\"报表名称，如 'east_region_orders_202607'\")\n", " file_format: Literal[\"xlsx\", \"csv\"] = Field(description=\"导出格式\")\n", " estimated_rows: int = Field(description=\"预计导出行数\")\n", " reason: str = Field(description=\"导出原因，用于审计\")\n", " priority: Literal[\"low\", \"normal\", \"high\"] = Field(\n", " default=\"normal\", description=\"优先级\"\n", " )\n", "\n", "\n", "# 第二步：安全工具不加 args_schema（自动执行）\n", "@tool\n", "def check_export_permission(username: str) -> dict:\n", " \"\"\"查询用户导出权限。安全操作，无需审批。\"\"\"\n", " return {\"username\": username, \"can_export\": True, \"region\": \"all\"}\n", "\n", "\n", "# 第三步：危险工具加上 args_schema，配合 HITL 拦截\n", "@tool(args_schema=ExportTaskInput)\n", "def create_export_task(\n", " report_name: str,\n", " file_format: str,\n", " estimated_rows: int,\n", " reason: str,\n", " priority: str = \"normal\",\n", ") -> dict:\n", " \"\"\"创建报表导出任务。这是一个高风险操作，需要人工审批。\"\"\"\n", " print(\n", " f\"[导出] 创建任务：{report_name} | {file_format} | \"\n", " f\"{estimated_rows} 行 | 优先级 {priority}\"\n", " )\n", " return {\n", " \"task_id\": \"EXPORT-20260706-002\",\n", " \"report_name\": report_name,\n", " \"file_format\": file_format,\n", " \"estimated_rows\": estimated_rows,\n", " \"status\": \"queued\",\n", " }\n", "\n", "\n", "# 第四步：创建带 HITL 的 Agent\n", "checkpointer = InMemorySaver() # HITL 必需：暂停后需要从这里恢复状态\n", "\n", "agent = create_agent(\n", " model=model,\n", " tools=[check_export_permission, create_export_task],\n", " middleware=[\n", " HumanInTheLoopMiddleware(\n", " interrupt_on={\n", " # 安全工具：不中断，自动执行\n", " \"check_export_permission\": False,\n", " # 危险工具：中断，提供三种审批选项\n", " \"create_export_task\": {\n", " \"allowed_decisions\": [\"approve\", \"edit\", \"reject\"],\n", " },\n", " },\n", " ),\n", " ],\n", " checkpointer=checkpointer,\n", " system_prompt=(\n", " \"你是报表导出平台的智能助手。\"\n", " \"用户查询权限时，直接调用 check_export_permission。\"\n", " \"只在用户明确要求创建导出任务时调用 create_export_task。\"\n", " ),\n", ")\n", "\n", "# 第五步：第一次 invoke——会被 HITL 拦截\n", "config = {\"configurable\": {\"thread_id\": \"export-001\"}}\n", "result = agent.invoke(\n", " {\n", " \"messages\": [\n", " {\n", " \"role\": \"user\",\n", " \"content\": \"我是 zhangsan，导出 east 区本月订单报表，xlsx，约 5000 行，月度对账用。\",\n", " }\n", " ]\n", " },\n", " config=config,\n", ")\n", "\n", "# result 里会包含中断信息，UI 层可以据此展示审批界面\n", "print(\"Agent 已暂停，等待审批...\")" ] }, { "cell_type": "code", "execution_count": 12, "id": "a2e8a224", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "{'messages': [HumanMessage(content='我是 zhangsan，导出 east 区本月订单报表，xlsx，约 5000 行，月度对账用。', additional_kwargs={}, response_metadata={}, id='fe4ee1b0-996b-4264-bf00-f2883fec0411'),\n", " AIMessage(content='好的，我先帮您查询导出权限。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 87, 'prompt_tokens': 529, 'total_tokens': 616, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 30, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 529}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '4133fb41-c27c-4ef3-b9c6-24df1f20883a', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019f4b7f-d312-74c3-aa9f-8f2401c210ec-0', tool_calls=[{'name': 'check_export_permission', 'args': {'username': 'zhangsan'}, 'id': 'call_00_yHROD8Mj8Or6bXeFkVNX8535', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 529, 'output_tokens': 87, 'total_tokens': 616, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 30}}),\n", " ToolMessage(content='{\"username\": \"zhangsan\", \"can_export\": true, \"region\": \"all\"}', name='check_export_permission', id='d4492593-d9fb-47e6-aab8-99e4d366b271', tool_call_id='call_00_yHROD8Mj8Or6bXeFkVNX8535'),\n", " AIMessage(content='✅ 权限查询通过！**zhangsan** 拥有全部区域的导出权限，可以导出 east 区数据。\\n\\n在创建任务前，想跟您确认一下报表名称中的月份信息：当前是 **2026年7月**，报表名称是否确定为 `east_region_orders_202607`？还是您希望我按其他名称来创建？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 542, 'prompt_tokens': 649, 'total_tokens': 1191, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 464, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 512}, 'prompt_cache_hit_tokens': 512, 'prompt_cache_miss_tokens': 137}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '039b6556-e491-4c04-abc0-563187a2783e', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019f4b7f-d7e2-7011-b44f-2717fe00c27c-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 649, 'output_tokens': 542, 'total_tokens': 1191, 'input_token_details': {'cache_read': 512}, 'output_token_details': {'reasoning': 464}})]}" ] }, "execution_count": 12, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from langgraph.types import Command\n", "\n", "# 审批通过：工具以原始参数执行\n", "agent.invoke(\n", " Command(resume={\"decisions\": [{\"type\": \"approve\"}]}),\n", " config=config,\n", ")\n", "\n", "# 编辑后通过：修改参数再执行（比如把 estimated_rows 从 50000 改成 5000）\n", "agent.invoke(\n", " Command(resume={\n", " \"decisions\": [\n", " {\n", " \"type\": \"edit\",\n", " \"edited_action\": {\n", " \"name\": \"create_export_task\",\n", " \"args\": {\n", " \"report_name\": \"east_region_orders_202607\",\n", " \"file_format\": \"xlsx\",\n", " \"estimated_rows\": 5000, # 人工修正\n", " \"reason\": \"月度对账\",\n", " \"priority\": \"high\",\n", " },\n", " },\n", " }\n", " ]\n", " }),\n", " config=config,\n", ")\n", "\n", "# 拒绝：工具不执行，拒绝原因会反馈给模型\n", "agent.invoke(\n", " Command(resume={\n", " \"decisions\": [\n", " {\n", " \"type\": \"reject\",\n", " \"message\": \"审批被驳回：本月对账已由系统自动完成，无需手动导出。\",\n", " }\n", " ]\n", " }),\n", " config=config,\n", ")" ] }, { "cell_type": "code", "execution_count": 18, "id": "19531c59", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "{'messages': [HumanMessage(content='我的手机号 [REDACTED_PHONE_NUMBER]，邮箱 [REDACTED_EMAIL]，', additional_kwargs={}, response_metadata={}, id='b51da9fe-4d2a-4e82-9330-8edf39f9be90'), AIMessage(content='您提供的联系方式已记录。请问您遇到了什么报表导出问题？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 54, 'prompt_tokens': 43, 'total_tokens': 97, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 39, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 43}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '2d0a7858-c9aa-4f38-9c17-40096c81dfac', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019f4bdf-8ab5-7da0-866a-016abc9cc391-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 43, 'output_tokens': 54, 'total_tokens': 97, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 39}})]}\n" ] } ], "source": [ "from langchain.agents.middleware import PIIMiddleware\n", "\n", "# 中国大陆手机号：1 开头，第二位 3-9，共 11 位\n", "PHONE_PATTERN = r\"(? dict:\n", " \"\"\"查询用户的历史导出记录。\"\"\"\n", " QUERY_ATTEMPTS[user_id] = QUERY_ATTEMPTS.get(user_id, 0) + 1\n", " attempt = QUERY_ATTEMPTS[user_id]\n", "\n", " if attempt < 3:\n", " # 前两次模拟网络超时\n", " raise TimeoutError(f\"查询接口超时（第 {attempt} 次）\")\n", " # 第三次成功\n", " return {\n", " \"user_id\": user_id,\n", " \"records\": [\n", " {\"task_id\": \"EXP-001\", \"report\": \"月度订单\", \"time\": \"2026-07-01\"},\n", " {\"task_id\": \"EXP-002\", \"report\": \"客户分析\", \"time\": \"2026-07-05\"},\n", " ],\n", " }\n", "\n", "\n", "agent = create_agent(\n", " model=model,\n", " tools=[fetch_export_history],\n", " middleware=[\n", " ToolRetryMiddleware(\n", " tools=[\"fetch_export_history\"], # 只对这个工具启用重试\n", " max_retries=2, # 额外重试 2 次（共 3 次机会）\n", " retry_on=(TimeoutError,), # 只在超时时重试，权限错误不重试\n", " initial_delay=0.2, # 第一次重试前等 0.2 秒\n", " max_delay=1.0, # 重试间隔上限 1 秒（指数退避）\n", " on_failure=\"continue\", # 重试耗尽后让模型继续（不抛异常）\n", " ),\n", " ],\n", " system_prompt=\"你是报表导出平台的查询助手。根据查询结果回答用户。\",\n", ")\n", "\n", "result = agent.invoke({\n", " \"messages\": [\n", " {\n", " \"role\": \"user\",\n", " \"content\": '查询一下用户 user123 的历史导出记录',\n", " }\n", " ]\n", " })\n", "\n", "for message in result['messages']:\n", " print(message)" ] }, { "cell_type": "code", "execution_count": 40, "id": "f17b6de3", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "content='查询一下用户 user123 是否有导出权限' additional_kwargs={} response_metadata={} id='4c6f7cf0-e75b-4c71-9ef2-25e61064d862'\n", "content='好的，我来查询一下用户 user123 的导出权限。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 81, 'prompt_tokens': 430, 'total_tokens': 511, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 21, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 384}, 'prompt_cache_hit_tokens': 384, 'prompt_cache_miss_tokens': 46}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '905cb3b6-090b-42a7-b172-ffc2551a0434', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019f4f88-de3a-7583-b2a6-c3fe6fe9486f-0' tool_calls=[{'name': 'check_export_permission', 'args': {'username': 'user123'}, 'id': 'call_00_QMBCT9d3OTuz2u7nVEaQ7776', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 430, 'output_tokens': 81, 'total_tokens': 511, 'input_token_details': {'cache_read': 384}, 'output_token_details': {'reasoning': 21}}\n", "content='{\"username\": \"user123\", \"can_export\": false, \"reason\": \"用户不存在\"}' name='check_export_permission' id='0f91daac-b0d8-483b-9ff3-7859c7d8c009' tool_call_id='call_00_QMBCT9d3OTuz2u7nVEaQ7776'\n", "content='查询结果如下：\\n\\n- **用户名**：user123\\n- **是否有导出权限**：❌ 无权限\\n- **原因**：**用户不存在**\\n\\n系统中未找到名为 `user123` 的用户，因此无法为其分配导出权限。建议您：\\n\\n1. 确认用户名是否输入正确（例如是否使用英文名账号）。\\n2. 若该用户是新员工，可能需要先联系管理员将账号添加到系统中。\\n3. 如有其他需要，请提供正确的用户名，我可以再次查询。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 129, 'prompt_tokens': 544, 'total_tokens': 673, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 23, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 384}, 'prompt_cache_hit_tokens': 384, 'prompt_cache_miss_tokens': 160}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '3c7692f2-ea12-476e-8376-dcb6ca64fb9d', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019f4f88-e2c2-7791-834e-1d57fa2d271c-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 544, 'output_tokens': 129, 'total_tokens': 673, 'input_token_details': {'cache_read': 384}, 'output_token_details': {'reasoning': 23}}\n" ] } ], "source": [ "from langchain.agents.middleware import ModelRetryMiddleware, ModelFallbackMiddleware\n", "\n", "# 本地模型：Ollama + Qwen3，成本低但在高负载下可能不可用\n", "local_model = init_chat_model(model_provider=\"qwen3.5:0.8b\")\n", "\n", "# 云端模型：DeepSeek，稳定但要花钱\n", "cloud_model = init_chat_model(\n", " model=\"deepseek-v4-flash\",\n", " model_provider=\"openai\",\n", " base_url=\"https://api.deepseek.com\",\n", " api_key=os.getenv('OPENAI_API_KEY'),\n", ")\n", "\n", "agent = create_agent(\n", " model=local_model, # 默认用本地模型\n", " tools=tools,\n", " middleware=[\n", " # 第一层：同一个模型失败后重试\n", " ModelRetryMiddleware(\n", " max_retries=3, # 最多重试 3 次\n", " retry_on=(Exception,), # 任何异常都重试\n", " on_failure=\"continue\",\n", " ),\n", " # 第二层：重试也失败了，切备用模型\n", " ModelFallbackMiddleware(\n", " cloud_model, # 备用的云端模型\n", " ),\n", " ],\n", " system_prompt=\"你是报表导出平台的智能助手。\",\n", ")\n", "\n", "result = agent.invoke({\n", " \"messages\": [\n", " {\n", " \"role\": \"user\",\n", " \"content\": '查询一下用户 user123 是否有导出权限',\n", " }\n", " ]\n", " })\n", "\n", "for message in result['messages']:\n", " print(message)" ] }, { "cell_type": "code", "execution_count": 41, "id": "aa818783", "metadata": {}, "outputs": [], "source": [ "import time\n", "from langchain.agents.middleware import (\n", " before_model, after_model, wrap_model_call, wrap_tool_call,\n", " AgentState,\n", ")\n", "from langchain_core.messages import ToolMessage\n", "\n", "\n", "# ---- Middleware 1：模型调用前记录审计信息 ----\n", "@before_model\n", "def audit_before_model(state: AgentState, runtime):\n", " \"\"\"在模型调用前，记录当前上下文规模和来源用户。\"\"\"\n", " ctx = runtime.context or {}\n", " print(\n", " f\"[审计] 模型调用 #{state.get('call_count', 0) + 1} | \"\n", " f\"消息数 {len(state['messages'])} | \"\n", " f\"用户 {ctx.get('username', 'unknown')}\"\n", " )\n", " return None\n", "\n", "\n", "# ---- Middleware 2：模型返回后更新调用计数 ----\n", "@after_model\n", "def update_call_counter(state: AgentState, runtime):\n", " \"\"\"模型返回后，调用计数 +1。\"\"\"\n", " return {\"call_count\": state.get(\"call_count\", 0) + 1}\n", "\n", "\n", "# ---- Middleware 3：包裹模型调用，实现本地重试 + 云端降级 ----\n", "local_failures = 0\n", "MAX_LOCAL_FAILURES = 3\n", "\n", "\n", "@wrap_model_call\n", "def retry_local_then_cloud(request, handler):\n", " \"\"\"先用本地 Ollama 模型，失败 N 次后切到云端 DeepSeek。\"\"\"\n", " global local_failures\n", "\n", " try:\n", " result = handler(request) # 尝试用本地模型\n", " local_failures = 0 # 成功后重置计数器\n", " return result\n", " except Exception as e:\n", " local_failures += 1\n", " if local_failures < MAX_LOCAL_FAILURES:\n", " print(f\"[降级] 本地模型失败（{local_failures}/{MAX_LOCAL_FAILURES}），重试中...\")\n", " raise # 重新抛出，让 ModelRetryMiddleware 处理\n", " print(f\"[降级] 本地模型不可用，切换到云端 DeepSeek\")\n", " # request.override 临时换模型，不影响 Agent 默认配置\n", " return handler(request.override(model=cloud_model))\n", "\n", "\n", "# ---- Middleware 4：包裹工具调用，实现业务规则拦截 ----\n", "@wrap_tool_call\n", "def guard_export_tool(request, handler):\n", " \"\"\"在 create_export_task 执行前，做业务规则校验。\"\"\"\n", " tool_name = request.tool_call[\"name\"]\n", "\n", " if tool_name == \"create_export_task\":\n", " args = request.tool_call.get(\"args\", {})\n", " estimated_rows = args.get(\"estimated_rows\", 0)\n", "\n", " # 规则 1：超过 100 万行直接拒绝\n", " if estimated_rows > 1_000_000:\n", " return ToolMessage(\n", " content=(\n", " f\"导出被拦截：预计行数 {estimated_rows} 超过系统上限 1000000。\"\n", " \"建议：缩小时间范围或分批次导出。\"\n", " ),\n", " tool_call_id=request.tool_call[\"id\"],\n", " )\n", "\n", " # 规则 2：超过 10 万行发出警告但放行（打印日志供运维关注）\n", " if estimated_rows > 100_000:\n", " print(\n", " f\"[警告] 大导出任务：{args.get('report_name')} | \"\n", " f\"{estimated_rows} 行 | 用户可能需要等待较长时间\"\n", " )\n", "\n", " # 规则 3：记录工具调用耗时\n", " start = time.time()\n", " result = handler(request)\n", " elapsed = time.time() - start\n", " print(f\"[性能] {tool_name} 执行耗时 {elapsed:.2f}s\")\n", "\n", " return result\n", "\n", " # 非目标工具直接放行\n", " return handler(request)" ] }, { "cell_type": "code", "execution_count": null, "id": "b8d71698", "metadata": {}, "outputs": [], "source": [ "from langchain.agents.middleware import AgentMiddleware\n", "\n", "\n", "class ExportGovernanceMiddleware(AgentMiddleware):\n", " \"\"\"报表导出治理中间件：按角色限制导出规模。\n", "\n", " 用法：\n", " ExportGovernanceMiddleware(\n", " role_limits={\"finance\": 500_000, \"operation\": 100_000, \"default\": 50_000}\n", " )\n", " \"\"\"\n", "\n", " def __init__(self, role_limits: dict = None):\n", " super().__init__()\n", " self.role_limits = role_limits or {\"default\": 100_000}\n", "\n", " def before_model(self, state, runtime):\n", " \"\"\"每次模型调用前，注入当前用户的权限信息到日志。\"\"\"\n", " ctx = runtime.context or {}\n", " role = ctx.get(\"user_role\", \"default\")\n", " limit = self.role_limits.get(role, self.role_limits[\"default\"])\n", " print(f\"[治理] 用户角色 {role}，导出上限 {limit} 行\")\n", " return None\n", "\n", " def wrap_tool_call(self, request, handler):\n", " \"\"\"在创建导出任务前，检查是否超过该角色的导出上限。\"\"\"\n", " if request.tool_call[\"name\"] != \"create_export_task\":\n", " return handler(request)\n", "\n", " ctx = request.runtime.context or {}\n", " role = ctx.get(\"user_role\", \"default\")\n", " limit = self.role_limits.get(role, self.role_limits[\"default\"])\n", " args = request.tool_call.get(\"args\", {})\n", " estimated_rows = args.get(\"estimated_rows\", 0)\n", "\n", " if estimated_rows > limit:\n", " return ToolMessage(\n", " content=(\n", " f\"导出被拦截：预计 {estimated_rows} 行，\"\n", " f\"超过 {role} 角色的上限 {limit} 行。\"\n", " \"请联系上级审批或缩小导出范围。\"\n", " ),\n", " tool_call_id=request.tool_call[\"id\"],\n", " )\n", "\n", " return handler(request)\n", "\n", "\n", "# 使用\n", "agent = create_agent(\n", " model=model,\n", " tools=tools,\n", " middleware=[\n", " ExportGovernanceMiddleware(\n", " role_limits={\n", " \"finance\": 500_000,\n", " \"operation\": 100_000,\n", " \"default\": 50_000,\n", " }\n", " ),\n", " ],\n", " context_schema=RunContext,\n", " system_prompt=\"你是报表导出平台的智能助手。\",\n", ")" ] }, { "cell_type": "code", "execution_count": 42, "id": "c14bec54", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "# 报表系统开发周报（2023-10-23 ~ 2023-10-27）\n", "\n", "## 一、本周完成事项（按优先级排列）\n", "\n", "1. **导出接口性能优化** \n", " - 异步分页流式导出改造完成，支持 10 万行数据集导出，耗时由平均 35s 降至 8s。 \n", " - 内存占用从峰值 600MB 降至 120MB，OOM 风险消除。 \n", " - 导出重试、进度回调机制已上线，超时重试次数调整为 3 次。\n", "\n", "2. **权限模块 RBAC 重构** \n", " - 完成用户-角色-权限三级模型数据迁移（影响 42 个接口，156 个权限点）。 \n", " - 权限校验逻辑从逐 SQL 过滤改为缓存 + 注解拦截，接口平均响应时间下降 40%。 \n", " - 新增权限测试覆盖率为 92%，回滚脚本已通过预发布验证。\n", "\n", "3. **元数据缓存预热** \n", " - 报表字段元数据从 Redis 冷启动改为启动时批量加载，首次查询耗时从 2.8s 降至 0.9s。 \n", "\n", "## 二、关键进展（数据支撑）\n", "\n", "| 指标 | 优化前 | 优化后 | 提升幅度 |\n", "|------|--------|--------|----------|\n", "| 导出 10 万行耗时 | 35s | 8s | **77%** |\n", "| 导出峰值内存 | 600MB | 120MB | **80%** |\n", "| 权限验证接口 P99 | 180ms | 68ms | **62%** |\n", "| 首次报表加载元数据 | 2.8s | 0.9s | **68%** |\n", "\n", "## 三、遇到的问题（附解决思路）\n", "\n", "1. **导出断连问题**：部分长时间导出因 Nginx 超时断开连接。 \n", " → 解决方案：前端改为轮询导出进度，后端异步写入临时文件，最后统一推送下载链接。目前已修复，监控无新增断连。\n", "\n", "2. **权限缓存击穿**：热点角色（如管理员）在缓存失效瞬间导致数据库瞬时高负载。 \n", " → 解决方案：采用互斥锁（分布式锁）控制缓存刷新，并预设永不过期的备份缓存降级。已通过压测验证，未再出现击穿。\n", "\n", "3. **元数据变更同步延迟**：报表字段新增后需等待 5 分钟才能被缓存感知。 \n", " → 解决方案：集成 Canal 监听元数据表 binlog，实现秒级失效缓存。已上线，延迟 < 2s。\n", "\n", "## 四、下周计划\n", "\n", "1. **导出接口灰度发布**：全量切流至新导出服务，观测一周稳定性，同时关闭旧导出路由。 \n", "2. **权限模块灰度验证**：逐步将 20% 用户流量引流至新鉴权中间件，持续监控慢查询和异常率。 \n", "3. **报表订阅推送优化**：基于新权限模型，修正订阅推送时的角色过滤逻辑，预计影响 8 个订阅任务。 \n", "4. **技术债务清理**：移除权限旧表冗余索引，回收已废弃的 3 个导出临时目录。\n", "\n", "---\n", "\n", "**备注**：本周无线上事故，待办清单已清零，下周四前完成灰度发布的所有评审。\n", "---\n", "# 线上事故复盘报告：报表导出服务宕机\n", "\n", "**报告人**：SRE 团队 \n", "**报告日期**：2025-03-03 \n", "**受众**：研发管理层 \n", "\n", "---\n", "\n", "## 1. 事故概述\n", "\n", "| 项目 | 内容 |\n", "|------|------|\n", "| **事故时间** | 2025年3月1日（周六）14:00 – 15:30（UTC+8） |\n", "| **影响范围** | 报表导出服务完全不可用，所有通过 Web 前端及 API 发起的导出请求均返回超时/502 错误。影响用户约 500+，涉及销售、运营等内部团队及相关外部客户。 |\n", "| **持续时间** | 1小时30分钟（从首次告警至服务完全恢复） |\n", "\n", "---\n", "\n", "## 2. 根因分析\n", "\n", "### 2.1 直接原因\n", "\n", "数据库连接池被长时间运行的报表导出查询耗尽。 \n", "- 导出服务使用 HikariCP 连接池，最大连接数配置为 20。 \n", "- 14:00 前后，多个并发导出请求触发了未优化的 SQL 查询（涉及多表 JOIN、全表扫描、无索引过滤），每笔查询平均耗时 45 秒。 \n", "- 短时间内所有连接均被占用，新请求无法获取连接，导致连接池满、请求排队超时，最终服务进程假死。\n", "\n", "### 2.2 间接原因\n", "\n", "| 原因类别 | 具体描述 |\n", "|----------|----------|\n", "| **代码与 SQL 质量** | 导出 SQL 未遵循最佳实践，缺少 WHERE 条件中的索引列，且未限制扫描行数；部分查询未使用分页/流式处理。 |\n", "| **配置合理性** | 连接池最大连接数（20）未根据历史高峰流量进行评估；缺乏针对长查询的超时退出机制（connection timeout / validation timeout 均为默认值）。 |\n", "| **监控与告警缺失** | 未对连接池使用率、慢查询（>10s）进行实时监控与告警；导出服务自身的超时熔断机制未启用。 |\n", "| **变更管理** | 上周五（2月28日）开发团队上线了新的报表导出功能，未做 SQL Review 和压测，未通知 SRE 同步更新监控规则。 |\n", "\n", "> **说明**：本次事故并非单一因素导致，而是代码质量、配置、监控、流程四方面弱点在特定流量并发下的集中体现。SRE 对配置与监控的被动维护负有管理责任，开发团队对 SQL 与变更流程负有改进责任。\n", "\n", "---\n", "\n", "## 3. 处理过程（时间线）\n", "\n", "| 时间 (UTC+8) | 事件 |\n", "|--------------|------|\n", "| 14:00 | 监控系统触发“API 5xx 错误率 > 5%”告警，值班 SRE 响应。 |\n", "| 14:02 | 初步排查：确认问题集中在 `/export/report` 接口，日志显示大量 `HikariPool-1 - Connection is not available, request timed out after 30000ms`。 |\n", "| 14:05 | 定位根因为数据库连接池耗尽，连接全部被占用且无法释放。 |\n", "| 14:08 | 临时应急：通过数据库后台手动 `KILL` 长时间运行的查询（共 12 个），释放 8 个连接，服务部分恢复（约 30% 请求成功）。 |\n", "| 14:12 | 发现新的慢查询仍在不断产生，决定重启导出服务容器以重置连接池状态。 |\n", "| 14:15 | 重启后连接池重置为 20，但新的慢查询再次迅速占满。 |\n", "| 14:20 | 紧急调整连接池最大连接数至 50，同时设置 `connectionTimeout=5000ms` 和 `maxLifetime=600000ms`，并发布配置热更新。服务逐步恢复至 80% 可用。 |\n", "| 14:25 | 联系开发团队协助分析慢查询 SQL。 |\n", "| 14:40 | 开发团队定位到上周五新增的报表 SQL 缺失索引，临时添加索引并优化 WHERE 条件。 |\n", "| 14:50 | 推送 SQL 优化补丁，通过灰度发布验证。 |\n", "| 15:00 | 全部优化生效，连接池使用率降至 30% 以下，导出服务完全恢复。 |\n", "| 15:15 | 确认无遗留异常，关闭告警。 |\n", "| 15:30 | 输出初步事故报告，通知相关方。 |\n", "\n", "---\n", "\n", "## 4. 改进措施\n", "\n", "### 4.1 短期措施（1周内完成）\n", "\n", "| 编号 | 措施 | 负责人 | 优先级 |\n", "|------|------|--------|--------|\n", "| S-1 | 增加导出服务连接池最大连接数至 50，并设置合理的超时参数（connectionTimeout=5s, validationTimeout=3s） | SRE | P0 |\n", "| S-2 | 对现有数据库执行慢查询分析，为查询时间 > 10s 的语句添加必要索引 | DBA + 开发 | P0 |\n", "| S-3 | 在 SRE 监控平台增加连接池使用率（当前连接数/最大连接数）、慢查询数量、API 超时率告警，阈值设为 70% | SRE | P0 |\n", "| S-4 | 对导出接口增加熔断/限流机制（基于 Sentinel 或 Hystrix），当错误率 > 10% 时自动降级并返回友好提示 | 开发 | P1 |\n", "\n", "### 4.2 长期措施（1个月内完成）\n", "\n", "| 编号 | 措施 | 负责人 | 预期效果 |\n", "|------|------|--------|----------|\n", "| L-1 | 所有报表导出 SQL 必须在预发布环境通过慢查询扫描（阈值 1s），纳入 CI 门禁 | 开发 + QA | 杜绝未优化 SQL 进入生产 |\n", "| L-2 | 建立变更协同流程：涉及数据库查询、连接池、外部依赖的变更须提前知会 SRE 并同步进行压测 | 开发 + SRE | 避免信息不对称导致的容量和监控盲区 |\n", "| L-3 | 将导出服务改造为异步队列模式：请求入 RabbitMQ / Kafka，worker 消费后回调通知用户下载 | 开发 + SRE | 隔离长查询对连接池的冲击，提升整体吞吐 |\n", "| L-4 | 引入数据库连接池自动扩容机制（如 HikariCP 动态调整 + Kubernetes HPA），并根据历史高峰配置基准值 | SRE | 应对突发流量，降低人工干预 |\n", "\n", "---\n", "\n", "## 5. 总结\n", "\n", "本次事故的根本原因是**代码质量、配置、监控、流程四重缺失在特定流量下的共振**。SRE 团队作为服务可用性的最终防线，未能在变更前主动识别连接池风险、未配置针对性的监控告警，负有直接管理责任。开发团队在上线前未进行 SQL Review 和压测，变更信息未同步，同样承担重要责任。\n", "\n", "后续我们将通过“短平快”的配置优化和“体系化”的异步改造及流程管控，从根源上降低此类事故再次发生的概率。同时，SRE 将定期组织跨团队的容量评估与混沌工程演练，持续提升系统韧性。\n", "\n", "---\n", "\n", "**附录**：慢查询 SQL 清单及优化方案（略） \n", "**责任人签字**：SRE 负责人 / 后端开发负责人\n", "---\n" ] } ], "source": [ "from langchain.agents.middleware import ModelRequest, dynamic_prompt\n", "\n", "\n", "def latest_user_text(request: ModelRequest) -> str:\n", " \"\"\"从消息列表中提取最近一条用户消息的文本。\"\"\"\n", " # 倒序遍历，找到最新的 human 类型消息\n", " for message in reversed(request.messages):\n", " if message.type == \"human\":\n", " return str(message.content)\n", " return \"\"\n", "\n", "\n", "@dynamic_prompt\n", "def adaptive_prompt(request: ModelRequest) -> str:\n", " \"\"\"根据用户请求的内容，动态选择 System Prompt 风格。\"\"\"\n", " user_text = latest_user_text(request)\n", "\n", " # 从 runtime.context 读取受众信息（由外部系统传入）\n", " ctx = request.runtime.context or {}\n", " audience = ctx.get(\"audience\", \"内部团队\")\n", "\n", " # 分支 1：事故复盘 → 严谨结构化\n", " if \"事故\" in user_text or \"复盘\" in user_text:\n", " return (\n", " \"你是一名资深 SRE 工程师，正在撰写线上事故复盘报告。\\n\"\n", " f\"目标受众：{audience}。\\n\"\n", " \"请按以下结构组织内容：\\n\"\n", " \"1. 事故概述（时间、影响范围、持续时间）\\n\"\n", " \"2. 根因分析（直接原因、间接原因）\\n\"\n", " \"3. 处理过程（时间线）\\n\"\n", " \"4. 改进措施（短期 + 长期）\\n\"\n", " \"语气：客观、严谨；不甩锅，不带情绪。\"\n", " )\n", "\n", " # 分支 2：周报 → 简洁干练\n", " if \"周报\" in user_text:\n", " return (\n", " \"你是一名技术团队的 PM，正在编写项目周报。\\n\"\n", " f\"目标受众：{audience}。\\n\"\n", " \"请按以下结构组织内容：\\n\"\n", " \"1. 本周完成事项（按优先级排列）\\n\"\n", " \"2. 关键进展（数据支撑）\\n\"\n", " \"3. 遇到的问题（附解决思路）\\n\"\n", " \"4. 下周计划\\n\"\n", " \"要求：简洁、有数据、不写流水账。\"\n", " )\n", "\n", " # 分支 3：默认日常回复\n", " return (\n", " \"你是报表导出平台的智能助手。\\n\"\n", " f\"目标受众：{audience}。\\n\"\n", " \"请用简洁清晰的语言回答用户问题，必要时给出操作指引。\"\n", " )\n", "\n", "\n", "agent = create_agent(\n", " model=model,\n", " middleware=[adaptive_prompt],\n", " context_schema=RunContext,\n", ")\n", "\n", "# 两个不同场景的请求，会得到完全不同的 System Prompt\n", "questions = [\n", " \"帮我写一份本周的报表系统开发周报，涵盖导出接口优化和权限模块改造\",\n", " \"帮我对上周六报表导出服务宕机的事故做一份复盘报告\",\n", "]\n", "\n", "for question in questions:\n", " result = agent.invoke(\n", " {\"messages\": [{\"role\": \"user\", \"content\": question}]},\n", " context={\"audience\": \"研发管理层\"},\n", " )\n", " print(result[\"messages\"][-1].content)\n", " print(\"---\")" ] } ], "metadata": { "kernelspec": { "display_name": "05_langgraph (3.11.x)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.15" } }, "nbformat": 4, "nbformat_minor": 5 }