1. 这不是又一个“AI Agent速成班”，而是一份能让你在真实项目里写得出、跑得通、扛得住压的实战手记
你点开这个标题，大概率不是为了听“Agent是智能体”“LangChain是编排框架”这种教科书定义。你真正想问的是：我昨天刚用LangChain搭了个天气查询Agent，结果用户一并发查5个城市，服务直接502；我照着教程把LangGraph流程图画得比UML还漂亮，可一接入真实API就卡在工具调用返回空；更别说看到“MCP协议”四个字，连搜都搜不出几篇像样的中文文档——它到底是个软件层的通信规范，还是硬件设备间的物理握手协议？为什么Playwright和Chrome DevTools都扯上MCP，但实现方式天差地别？这些困惑，我在过去18个月里，带着团队在三个落地项目中反复踩坑、推倒重来、再重构，才真正理清楚。
这门课之所以敢称“B站唯一讲明白的”，不是因为它讲得多全，而是它只讲必须讲透的那部分：MCP不是玄学概念，它是Agent与外部世界建立可信连接的“数字握手协议”；A2A（Agent-to-Agent）不是未来愿景，而是你现在就能用FastAPI+LangGraph搭出的跨服务协作链路；Agent开发的核心难点，从来不在模型调用，而在状态一致性、工具调用容错、上下文生命周期管理这三个被90%教程跳过的硬骨头。我不会带你从零装Python环境，但会告诉你为什么LangChain的RunnableParallel在高并发下会 silently drop 某些分支结果；不会逐行解释LangGraph源码，但会手把手教你用StateGraph的add_edge()绕过默认的conditional edge陷阱，让重试逻辑真正生效。如果你的目标是下周就要在公司内部系统里集成一个能自动处理报销单的Agent，而不是做一个能回答“今天天气如何”的Demo，那么接下来的内容，就是你该花时间读完的。
2. 核心设计思路拆解：为什么必须把MCP、A2A、LangGraph三者拧在一起讲？
2.1 不讲MCP，所有Agent都是“空中楼阁”
很多教程把Agent讲成“大模型+提示词+几个工具”，这就像教人盖房子只讲砖块和水泥，却不说地基怎么打。MCP（Model Control Protocol）就是那个被长期忽视的地基。它不是某个具体库，而是一套标准化的双向通信契约，定义了Agent如何向外部系统（浏览器、数据库、ERP、甚至硬件传感器）发送指令，以及外部系统如何结构化地返回结果、错误和状态变更。它的核心价值，在于解决三个致命问题：
工具调用的语义鸿沟：你让Agent调用“查询订单”，LangChain默认生成的tool_call可能是
{"name": "get_order", "args": {"order_id": "123"}}，但真实ERP接口可能要求{"method": "GET", "path": "/api/v2/orders/123", "headers": {...}}。MCP强制约定统一的action、input、output_schema字段，让Agent输出可预测，外部系统响应可解析。状态同步的不可靠性：传统HTTP调用是无状态的。Agent执行“提交审批”后，无法知道审批流是否真的走到下一环节。MCP通过
wss://长连接支持event推送，当审批系统触发approval_status_changed事件时，Agent能实时收到{"event": "approval_status_changed", "data": {"id": "123", "status": "approved"}}，无需轮询。安全边界的模糊地带：一个Agent同时调用财务系统和HR系统，权限如何隔离？MCP的
capability声明机制，要求每个工具调用前必须携带required_capabilities: ["finance:read", "hr:write"]，由MCP Server做RBAC校验，避免Agent越权。
提示：
wss://api.xiaozhi.me/mcp/?token=eyjhbgcioijfuzi1niisinr5cci6ikpxvcj9.eyj...这类URL不是随便写的。token是JWT，其中payload必须包含iss（签发方）、aud（受众，即目标MCP Server地址）、exp（过期时间），且aud需与Server配置的allowed_audiences严格匹配。我见过太多团队因token中aud写成https://api.xiaozhi.me而实际Server配置为wss://api.xiaozhi.me/mcp，导致401错误却查不出原因。
2.2 A2A不是“多个Agent聊天”，而是服务网格的AI原生形态
“A2A”这个词被过度浪漫化了。它不是让两个Agent互相说“你好吗”，而是构建可组合、可编排、可监控的服务协作网络。关键在于理解：A2A通信必须基于MCP，否则就是裸HTTP调用，无法保证可靠性、可观测性和安全性。
我们曾在一个政务系统中实践A2A：一个“材料预审Agent”负责检查用户上传的身份证、营业执照扫描件是否清晰、完整；一个“政策匹配Agent”根据企业类型、注册地、行业，从知识库中检索适用的补贴政策；一个“申报生成Agent”将预审结果和匹配政策合成标准申报表。三者不是独立运行，而是通过MCP Server串联：
- 用户提交材料 → 预审Agent通过MCP调用OCR服务 → 返回结构化结果
- 预审成功 → 预审Agent通过MCP向政策匹配Agent发送
{ "action": "match_policy", "input": { "biz_type": "tech_startup", "region": "shanghai" } } - 政策匹配Agent处理完毕 → 通过同一MCP Server向申报生成Agent推送
{ "event": "policy_matched", "data": { "policy_id": "SH2024-001", "amount": "50000" } }
这个链路的价值在于：故障隔离（OCR服务宕机不影响政策匹配）、弹性伸缩（申报生成Agent可水平扩容应对申报高峰）、统一追踪（所有MCP调用日志带trace_id，可在Kibana中关联查看全链路）。如果不用MCP，而是让预审Agent直接HTTP调用政策匹配Agent的API，一旦后者超时，前者就得自己实现重试、熔断、降级——这本该是服务网格做的事，不该由AI逻辑承担。
2.3 LangGraph不是“画流程图”，而是状态机驱动的Agent操作系统
LangGraph常被误解为“可视化编排工具”。其实它的本质是基于State的有限状态机（FSM）引擎。StateGraph中的每个节点（node）不是函数，而是状态转换器（state transformer）；每条边（edge）不是箭头，而是状态谓词（state predicate）。这决定了它的设计哲学：一切行为必须围绕state展开。
我们重构一个客服Agent时，发现旧版LangChain Chain在处理“用户反复修改订单地址”时逻辑混乱。改用LangGraph后，state定义为：
class OrderState(TypedDict): user_input: str # 当前用户输入 order_id: str # 订单ID current_address: str # 当前地址（来自DB） proposed_address: str # 用户提议的新地址 address_confirmed: bool # 用户是否确认新地址 needs_validation: bool # 是否需要调用地址验证API节点validate_address只做一件事：调用API验证proposed_address，更新state["needs_validation"] = False，并根据结果设置state["address_confirmed"]。边的条件不再是if "address" in user_input，而是lambda state: state["needs_validation"]。这样，无论用户第几次说“改成北京朝阳区”，只要needs_validation为True，就必然走验证流程。状态机天然解决了“上下文漂移”问题——这是纯Prompt工程永远无法根治的顽疾。
3. 核心细节与实操要点：从协议到代码，每一步都踩过坑
3.1 MCP Server搭建：选型、配置与安全加固
MCP Server是整个架构的中枢，它必须同时扮演协议网关、权限代理、事件总线三重角色。我们对比了三种主流方案：
| 方案 | 优势 | 劣势 | 适用场景 |
|---|---|---|---|
| 自研FastAPI+WebSockets | 完全可控，可深度集成公司SSO和审计日志 | 开发维护成本高，需自行实现心跳、重连、消息去重 | 大型企业，对安全审计有强要求 |
| MCP官方Reference Server (Node.js) | 协议兼容性最好，社区更新快 | TypeScript生态，Python团队需额外学习成本 | 快速验证MCP可行性，POC阶段 |
| LangChain MCP Adapter + 自定义Backend | 复用LangChain生态，工具注册简单 | 仅支持HTTP/WebSocket双模式，不支持纯WebSocket长连接 | 中小型项目，已有LangChain基础 |
我们最终选择自研FastAPI方案，核心配置如下：
# mcp_server.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from jose import JWTError, jwt import json import uuid from typing import Dict, List, Any app = FastAPI() security = HTTPBearer() # MCP Server核心配置 MCP_CONFIG = { "allowed_origins": ["https://your-frontend.com"], "max_connections_per_token": 5, "session_timeout_seconds": 1800, # 30分钟 "allowed_capabilities": { "finance": ["read", "write"], "hr": ["read"], "ocr": ["process"] } } async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): try: payload = jwt.decode( credentials.credentials, key="YOUR_SECRET_KEY", algorithms=["HS256"], options={"verify_aud": True, "require_exp": True} ) # 关键校验：audience必须精确匹配Server地址 if payload.get("aud") != "wss://api.xiaozhi.me/mcp": raise JWTError("Invalid audience") return payload except JWTError as e: raise HTTPException(status_code=401, detail=f"Token verification failed: {e}") @app.websocket("/mcp") async def mcp_websocket_endpoint( websocket: WebSocket, token_payload: dict = Depends(verify_token) ): await websocket.accept() client_id = str(uuid.uuid4()) # 建立会话，绑定token_payload中的capabilities session_caps = token_payload.get("capabilities", []) # 此处应存入Redis，key为client_id，value为session_caps # 并设置过期时间 try: while True: data = await websocket.receive_text() try: mcp_msg = json.loads(data) # 1. 校验MCP消息结构 if not all(k in mcp_msg for k in ["action", "input", "id"]): await websocket.send_text(json.dumps({"error": "Missing required fields"})) continue # 2. 校验capability权限 required_caps = mcp_msg.get("required_capabilities", []) if not set(required_caps).issubset(set(session_caps)): await websocket.send_text(json.dumps({ "error": f"Insufficient capabilities. Required: {required_caps}, Granted: {session_caps}" })) continue # 3. 路由到对应backend（如OCR、ERP） result = await route_to_backend(mcp_msg) await websocket.send_text(json.dumps(result)) except json.JSONDecodeError: await websocket.send_text(json.dumps({"error": "Invalid JSON"})) except Exception as e: await websocket.send_text(json.dumps({"error": str(e)})) except WebSocketDisconnect: # 清理会话 pass注意：MCP Server的
/mcp端点必须支持WebSocket子协议协商。客户端连接时需指定subprotocol="mcp.v1"，Server需在accept()时校验并返回相同协议。我们曾因前端未设置protocols参数，导致Chrome报错WebSocket is closed before the connection is established，排查了两天才发现是子协议不匹配。
3.2 LangGraph状态设计：如何让Agent记住“它正在做的事”
LangGraph的状态（State）不是简单的dict，而是有明确Schema的、可版本化的数据契约。我们定义BaseState时，坚持三个原则：
- 所有字段必须有默认值或明确的None含义：避免
KeyError，也避免state.get("xxx", None)的模糊判断。 - 敏感字段必须加密存储：如用户token、支付密钥，绝不以明文存入state。
- 状态变更必须原子化：一个node只能修改state中与其职责相关的字段，禁止“顺手”改其他字段。
以电商Agent为例，OrderState的TypedDict定义：
from typing import TypedDict, Optional, List, Dict, Any from datetime import datetime class Address(TypedDict): street: str city: str province: str postal_code: str class OrderItem(TypedDict): sku: str quantity: int price: float class OrderState(TypedDict): # 不变字段（初始化时设定） user_id: str session_id: str created_at: datetime # 可变字段（业务流程中更新） current_step: str # "address_selection", "payment_method", "confirm_order" order_items: List[OrderItem] shipping_address: Optional[Address] billing_address: Optional[Address] payment_method: Optional[str] # "alipay", "wechat", "credit_card" payment_status: str # "pending", "success", "failed" last_user_input: str # 最后一次用户输入，用于fallback error_count: int # 连续错误次数，用于触发人工介入 # 系统字段（LangGraph内部使用） __metadata__: Dict[str, Any] # 存放trace_id, span_id等关键技巧：用__metadata__字段承载LangGraph的内部状态，而非污染业务字段。例如，当Agent需要重试某个失败的支付操作时，不是在payment_status里写"retry_1"，而是：
def retry_payment(state: OrderState) -> OrderState: # 从metadata中读取重试次数 retry_count = state["__metadata__"].get("retry_count", 0) + 1 state["__metadata__"]["retry_count"] = retry_count # 业务状态只反映事实 state["payment_status"] = "pending" return state3.3 A2A通信实现：如何让Agent之间“说同一种语言”
A2A通信的核心是消息路由与能力发现。我们不依赖DNS或Service Mesh，而是用MCP Server内置的Registry机制：
- Agent注册：每个Agent启动时，向MCP Server发送
register消息，声明其capabilities和supported_actions。 - 能力发现：当Agent A需要调用Agent B的功能时，先向MCP Server查询
list_agents?capability=finance:write，获取可用Agent列表。 - 路由转发：MCP Server根据
action字段，将消息路由到注册了该action的Agent。
实际代码中，Agent A的调用逻辑：
import websockets import json import asyncio class A2AClient: def __init__(self, mcp_url: str, token: str): self.mcp_url = mcp_url self.token = token async def call_agent(self, target_agent_id: str, action: str, input_data: dict) -> dict: # 1. 建立WebSocket连接（复用已存在的连接池） async with websockets.connect( self.mcp_url, extra_headers={"Authorization": f"Bearer {self.token}"} ) as ws: # 2. 发送标准MCP调用消息 call_id = str(uuid.uuid4()) mcp_call = { "id": call_id, "action": action, "input": input_data, "target": target_agent_id, # 指定目标Agent ID "required_capabilities": ["finance:write"] # 权限声明 } await ws.send(json.dumps(mcp_call)) # 3. 等待响应（带超时） try: response = await asyncio.wait_for(ws.recv(), timeout=30.0) return json.loads(response) except asyncio.TimeoutError: raise TimeoutError(f"A2A call to {target_agent_id} timed out") # 在LangGraph node中使用 async def invoke_finance_agent(state: OrderState) -> OrderState: client = A2AClient("wss://api.xiaozhi.me/mcp", "your-jwt-token") try: result = await client.call_agent( target_agent_id="finance-processor-v2", action="process_refund", input_data={ "order_id": state["order_id"], "amount": state["refund_amount"] } ) state["refund_status"] = result.get("status", "unknown") state["refund_id"] = result.get("refund_id") except Exception as e: state["refund_status"] = "failed" state["error_message"] = str(e) return state实操心得：A2A调用必须实现幂等性。我们要求所有Finance相关的action（如
process_refund）必须接受idempotency_key参数，并在后端用RedisSETNX去重。否则，网络重试会导致同一笔退款被处理两次。这个细节，95%的教程都不会提，但线上事故的根源往往就在这里。
4. 完整实操流程：从零搭建一个“智能报销单处理Agent”
4.1 环境准备与依赖安装
我们采用Docker Compose统一管理所有组件，确保开发、测试、生产环境一致。docker-compose.yml核心片段：
version: '3.8' services: # MCP Server mcp-server: build: ./mcp-server ports: - "8001:8000" environment: - SECRET_KEY=your-super-secret-key-change-in-prod - DATABASE_URL=postgresql://user:pass@db:5432/mcp # LangGraph Agent (报销处理) expense-agent: build: ./expense-agent depends_on: - mcp-server - redis environment: - MCP_URL=ws://mcp-server:8000/mcp - MCP_TOKEN=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9... - REDIS_URL=redis://redis:6379/0 # Redis (用于状态存储和幂等性) redis: image: redis:7-alpine command: redis-server --save 60 1 --loglevel warning ports: - "6379:6379" # PostgreSQL (存储报销单、审批记录) db: image: postgres:15 environment: - POSTGRES_DB=expense_db - POSTGRES_USER=user - POSTGRES_PASSWORD=pass volumes: - ./pgdata:/var/lib/postgresql/data关键依赖版本（requirements.txt）：
langchain==0.1.18 langgraph==0.1.14 langchain-community==0.0.34 fastapi==0.111.0 uvicorn==0.29.0 websockets==12.0 python-jose[cryptography]==3.3.0 redis==4.6.0 psycopg2-binary==2.9.7 playwright==1.42.0 # 用于浏览器自动化注意：LangGraph 0.1.14 是目前最稳定的版本。0.2.x系列引入了
checkpointer抽象，但文档极不完善，我们在生产环境升级后遭遇了StateSnapshot序列化失败的问题，最终回滚。务必锁定版本。
4.2 MCP工具注册：让Agent能真正“操控”浏览器和ERP
报销Agent的核心能力是：1）自动登录公司OA系统；2）上传发票图片；3）填写报销单；4）提交审批。这些能力通过MCP工具暴露：
# tools/mcp_tools.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Optional, Dict, Any import asyncio import json class LoginOAInput(BaseModel): username: str = Field(..., description="OA系统用户名") password: str = Field(..., description="OA系统密码") class LoginOATool(BaseTool): name = "login_oa_system" description = "Login to company OA system using provided credentials" args_schema = LoginOAInput def _run(self, username: str, password: str) -> str: # 此处调用Playwright自动化脚本 # 但注意：Playwright必须在Docker容器中启用--no-sandbox # 否则会报错：Failed to read the 'localStorage' property return json.dumps({"status": "success", "session_id": "oa_sess_abc123"}) async def _arun(self, username: str, password: str) -> str: # 异步版本，避免阻塞Event Loop loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, self._run, username, password) return result # 注册为MCP工具 def register_mcp_tools(): from langchain_mcp import MCPClient client = MCPClient( server_url="ws://mcp-server:8000/mcp", token="your-jwt-token" ) # 注册工具，声明capability client.register_tool( tool=LoginOATool(), capability="oa:login", # 此capability需在MCP Server的allowed_capabilities中配置 description="Login to OA system" )Playwright配置关键点（playwright_config.py）：
from playwright.async_api import async_playwright async def get_browser_context(): pw = await async_playwright().start() # 必须添加--no-sandbox，否则Docker内无法启动 browser = await pw.chromium.launch( headless=True, args=["--no-sandbox", "--disable-setuid-sandbox"] ) context = await browser.new_context() # 设置全局超时，避免页面加载卡死 context.set_default_timeout(30000) return context, browser4.3 LangGraph工作流编排：处理一张报销单的完整状态机
报销单处理是一个典型的多步骤、多状态、需人工干预的流程。我们设计了7个状态节点：
from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage from typing import List, Dict, Any # 定义State class ExpenseState(TypedDict): user_id: str receipt_images: List[str] # 图片base64或URL parsed_data: Dict[str, Any] # OCR解析结果 oa_session_id: Optional[str] approval_status: str # "pending", "approved", "rejected", "manual_review" manual_review_reason: Optional[str] error_count: int last_action: str # 节点函数 def parse_receipts(state: ExpenseState) -> ExpenseState: # 调用OCR服务（通过MCP） ocr_result = call_mcp_tool("ocr_process", {"images": state["receipt_images"]}) state["parsed_data"] = ocr_result.get("data", {}) state["last_action"] = "parse_receipts" return state def login_oa(state: ExpenseState) -> ExpenseState: # 调用OA登录工具 login_result = call_mcp_tool("login_oa_system", { "username": "user_" + state["user_id"], "password": "auto_gen_pass" }) state["oa_session_id"] = login_result.get("session_id") state["last_action"] = "login_oa" return state def fill_form(state: ExpenseState) -> ExpenseState: # 填写报销单表单 form_data = { "amount": state["parsed_data"].get("total_amount", 0), "category": state["parsed_data"].get("category", "other"), "description": state["parsed_data"].get("description", "") } submit_result = call_mcp_tool("oa_fill_form", {"data": form_data}) state["last_action"] = "fill_form" return state def submit_approval(state: ExpenseState) -> ExpenseState: # 提交审批 submit_result = call_mcp_tool("oa_submit_approval", {"session_id": state["oa_session_id"]}) state["approval_status"] = submit_result.get("status", "pending") state["last_action"] = "submit_approval" return state def check_approval_status(state: ExpenseState) -> ExpenseState: # 轮询审批状态（实际应改为事件驱动，此处简化） status = call_mcp_tool("oa_check_status", {"session_id": state["oa_session_id"]}) state["approval_status"] = status.get("status", "pending") state["last_action"] = "check_approval_status" return state def escalate_to_human(state: ExpenseState) -> ExpenseState: # 触发人工审核 if state["error_count"] > 2: state["approval_status"] = "manual_review" state["manual_review_reason"] = "OCR parsing failed repeatedly" state["last_action"] = "escalate_to_human" return state def end_workflow(state: ExpenseState) -> ExpenseState: # 工作流结束 state["last_action"] = "end_workflow" return state # 构建StateGraph workflow = StateGraph(ExpenseState) # 添加节点 workflow.add_node("parse_receipts", parse_receipts) workflow.add_node("login_oa", login_oa) workflow.add_node("fill_form", fill_form) workflow.add_node("submit_approval", submit_approval) workflow.add_node("check_approval_status", check_approval_status) workflow.add_node("escalate_to_human", escalate_to_human) workflow.add_node("end", end_workflow) # 定义边（条件转移） workflow.add_edge("parse_receipts", "login_oa") workflow.add_edge("login_oa", "fill_form") workflow.add_edge("fill_form", "submit_approval") workflow.add_edge("submit_approval", "check_approval_status") # 条件边：根据approval_status决定下一步 def should_continue(state: ExpenseState) -> str: if state["approval_status"] == "pending": return "check_approval_status" # 继续轮询 elif state["approval_status"] in ["approved", "rejected"]: return "end" # 结束 else: return "escalate_to_human" # 其他状态转人工 workflow.add_conditional_edges( "check_approval_status", should_continue, { "check_approval_status": "check_approval_status", # 循环 "end": "end", "escalate_to_human": "escalate_to_human" } ) workflow.add_edge("escalate_to_human", "end") # 设置入口点 workflow.set_entry_point("parse_receipts") # 编译图 app = workflow.compile()4.4 高并发与稳定性保障：让Agent扛住1000QPS
报销系统上线首周，我们遭遇了峰值1200QPS的冲击，导致MCP Server内存暴涨、LangGraph Worker大量超时。根本原因在于：状态存储未分离、工具调用未限流、错误未降级。解决方案：
- 状态存储外置Redis：LangGraph默认将state存在内存，我们改用
RedisSaver：
from langgraph.checkpoint.redis import RedisSaver from redis import Redis redis_client = Redis(host="redis", port=6379, db=0) checkpointer = RedisSaver(redis=redis_client) app = workflow.compile(checkpointer=checkpointer)- 工具调用熔断与降级：为所有MCP工具调用增加
tenacity重试：
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception_type((ConnectionError, TimeoutError)) ) def call_mcp_tool(tool_name: str, input_data: dict) -> dict: # 实际调用逻辑 pass- 关键路径异步化：OCR解析耗时最长，我们将其设为异步任务，主流程不等待：
async def parse_receipts_async(state: ExpenseState) -> ExpenseState: # 启动后台OCR任务 task_id = await start_ocr_task(state["receipt_images"]) state["ocr_task_id"] = task_id state["approval_status"] = "ocr_processing" # 进入异步等待状态 return state # 新增节点：等待OCR完成 async def wait_for_ocr(state: ExpenseState) -> ExpenseState: result = await poll_ocr_result(state["ocr_task_id"]) state["parsed_data"] = result state["approval_status"] = "ready_to_login" return state5. 常见问题与排查技巧实录：那些只有踩过才懂的坑
5.1 MCP连接问题排查速查表
| 现象 | 可能原因 | 排查命令/方法 | 解决方案 |
|---|---|---|---|
| WebSocket连接立即关闭 | token中aud与Server配置不匹配 | jwt.io解码token，核对aud字段 | 修改token生成逻辑，确保aud精确等于MCP Server的wss://地址 |
401 Unauthorized | JWT签名密钥不一致 | openssl rsautl -verify -in token.sig -inkey public.key | 确认Server和Client使用同一SECRET_KEY，且Server未启用RSA而Client用HS256 |
403 Forbidden | required_capabilities未在token中声明 | 查看token payload的capabilities数组 | 在生成token时，显式添加"capabilities": ["oa:login", "ocr:process"] |
| 消息发送成功但无响应 | WebSocket未设置subprotocol | 浏览器控制台Network标签页，查看WebSocket帧 | Client连接时添加protocols: ['mcp.v1']，Server在accept()时指定相同协议 |
| 长连接频繁断开 | TCP Keepalive未启用 | netstat -an | grep :8001查看ESTABLISHED连接数 | 在FastAPI WebSocket中，设置ping_interval=20, ping_timeout=20 |
5.2 LangGraph状态丢失问题诊断
LangGraph状态丢失是最高频问题，根源几乎都在checkpointer配置：
- 现象：Agent执行到一半，重启后从头开始，或
state中字段为空。 - 根因：
RedisSaver未正确初始化，或thread_id未传入。 - 诊断：在
app.invoke()前，打印config：
config = {"configurable": {"thread_id": "123"}} print("Config:", config) # 确保thread_id存在且唯一 result = app.invoke(input_state, config=config)- 修复：确保每次调用都传入
configurable，且thread_id是业务唯一标识（如user_id + timestamp）。
5.3 Playwright在Docker中启动失败
- 现象：
playwright.chromium.launch()抛出Error: Failed to launch browser。 - 根因：Docker容器缺少沙箱所需依赖，或未禁用sandbox。
- 终极解决方案（
Dockerfile）：
FROM mcr.microsoft.com/playwright:v1.42.0-focal # 安装缺失依赖 RUN apt-get update && apt-get install -y \ libnss3 \ libatk1.0-0 \ libatk-bridge2.0-0 \ libc6 \ libcairo2 \ libcups2 \ libdbus-1-3 \ libexpat1 \ libfontconfig1 \ libgcc1 \ libglib2.0-0 \ libgtk-3-0 \ libnspr4 \ libpango-1.0-0 \ libpangocairo-1.0-0 \ libstdc++6 \ libx11-6 \ libx11-xcb1 \ libxcb1 \ libxcomposite1 \ libxcursor1 \ libxdamage1 \ libxext6 \ libxfixes3 \ libxi6 \ libxrandr2 \ libxrender1 \ libxss1 \ libxtst6 \ ca-certificates \ fonts-liberation \ libappindicator1 \ libasound2 \ libatk-bridge2.0-0 \ libatspi2.0-0 \ libcairo-gobject2 \ libcups2 \ libdrm2 \ libgbm1 \ libglib2.0-0 \ libgtk-3-0 \ libkrb5-3 \ libnspr4 \ libnss3 \ libpango-1.0-0 \ libpangocairo-1.0-0 \ libpulse0 \ libx11-xcb1 \ libxcomposite1 \ libxcursor1 \ libxdamage1 \ libxfixes3 \ libxi6 \ libxrandr2 \ libxrender1 \ libxss1 \ libxtst6 \ wget \ xdg-utils \ && rm -rf /var/lib/apt/lists/* # 关键：设置环境变量 ENV PULSE_SERVER=/dev/null ENV DISPLAY=:99 ENV NO_SANDBOX=1 # 启动Xvfb虚拟显示（Playwright需要） CMD ["sh", "-c", "Xvfb :99 -screen 0 1024x768x24 & exec python app.py"]5.4 A2A调用超时与重试策略
- 问题：A
