基于FastAPI与LangGraph构建生产级AI智能体开发框架

1. 项目概述：一个面向生产的AI智能体开发框架

如果你正在寻找一个能快速将基于大语言模型的智能体应用投入生产的解决方案，那么fastapi-langgraph-agent-production-ready-template这个项目很可能就是你需要的“脚手架”。这个项目标题本身就包含了几个关键信息：它基于 FastAPI 和 LangGraph，目标是构建一个“生产就绪”的智能体模板。简单来说，这不是一个玩具项目或实验性代码，而是一个为真实业务场景设计的、开箱即用的开发起点。

在实际工作中，我们常常面临这样的困境：手头有一个绝佳的 AI 应用创意，比如一个能自动处理工单的客服助手，或者一个能分析财报的金融顾问。用 Jupyter Notebook 快速验证原型后，却发现要把它变成一个 7x24 小时稳定运行、能处理高并发、具备完善监控和错误处理的在线服务，中间隔着巨大的工程鸿沟。你需要考虑 API 设计、状态管理、异步处理、数据库集成、日志、测试、部署等一系列问题。这个模板正是为了解决这个“最后一公里”的问题而生的。它预设了生产环境所需的最佳实践，让开发者可以专注于智能体本身的业务逻辑，而不是重复搭建基础设施。

它适合两类人：一是希望将 AI 智能体想法快速产品化的独立开发者或初创团队；二是在企业中负责 AI 应用落地的工程师，他们需要一个标准化、可维护的代码基来加速开发流程。接下来，我将深入拆解这个模板的核心设计、关键技术选型以及如何利用它来构建一个真正可用的智能体服务。

2. 核心架构与设计哲学解析

2.1 为什么是 FastAPI + LangGraph 的组合？

这个模板的技术栈选择非常精准，直指生产级 AI 应用的核心需求。FastAPI 是一个现代、快速（高性能）的 Python Web 框架，特别适合构建 API。它的优势在于异步支持好、自动生成交互式 API 文档（Swagger UI）、数据验证通过 Pydantic 变得极其优雅。对于 AI 应用，尤其是需要与 LLM 交互的应用，异步处理至关重要，因为调用 OpenAI、Anthropic 等外部 API 通常是网络 I/O 密集型操作，异步可以避免阻塞，极大提升并发能力。

LangGraph 则是 LangChain 生态中用于构建有状态、多步骤工作流（即智能体）的库。传统的 LangChain 链式调用虽然简单，但在处理复杂、带有循环、分支和持久化状态的智能体时显得力不从心。LangGraph 引入了图（Graph）的概念，将智能体的每一步定义为一个节点（Node），节点之间的流转由边（Edge）控制，并且内置了状态管理。这使得构建像“ReAct（思考-行动）”模式、多工具协作、带有记忆的对话等复杂智能体变得清晰和可维护。

将两者结合，FastAPI 负责对外提供稳定、高效的 HTTP 接口，处理用户请求、认证、限流等 Web 层事务；而 LangGraph 则在内部负责核心的 AI 推理与决策流程。这种分层架构确保了关注点分离：Web 层专注网络和业务协议，智能体层专注 AI 逻辑。

2.2 “生产就绪”体现在哪些方面？

一个“玩具项目”和一个“生产就绪”的项目，差距往往在细节里。这个模板的“生产就绪”特性，我理解主要体现在以下几个方面：

1. 配置化管理：所有敏感信息（如 API Keys）和可变参数（如模型名称、温度）都通过环境变量或配置文件管理，遵循 Twelve-Factor App 原则，便于不同环境（开发、测试、生产）的部署。
2. 结构化日志：集成了像structlog这样的库，日志输出为 JSON 格式，包含请求 ID、时间戳、严重级别等结构化信息。这对于使用 ELK（Elasticsearch, Logstash, Kibana）或 Datadog 等日志聚合系统进行问题排查和监控至关重要。
3. 健康检查与就绪探针：提供了/health和/ready这样的端点。在 Kubernetes 或 Docker Swarm 等容器编排平台中，这些端点用于判断服务实例是否健康、是否可以接收流量，是实现高可用的基础。
4. 错误处理与优雅降级：对 LLM 调用超时、API 额度不足、网络异常等常见错误有统一的捕获和处理机制，并可能提供降级策略（例如，切换备用模型或返回缓存结果），避免单点故障导致整个服务不可用。
5. 可观测性集成：可能预留了与 OpenTelemetry 等标准的集成点，用于收集分布式追踪、指标和日志，帮助开发者洞察智能体内部每个节点的执行耗时、状态流转。
6. 测试脚手架：包含了单元测试和集成测试的示例，鼓励测试驱动开发，确保智能体逻辑的稳定性和可回归性。
7. 容器化与部署脚本：提供 Dockerfile 和可能的 docker-compose.yml 或 Kubernetes manifests，让应用可以轻松地打包成容器镜像，并在任何云环境或本地数据中心一致地运行。

3. 模板核心模块与目录结构拆解

一个优秀的模板，其目录结构本身就能传达设计思想。虽然我无法看到该模板仓库的最新代码，但基于其目标，我们可以推断并构建一个典型的核心结构。理解这个结构，是你上手和定制化的第一步。

fastapi-langgraph-agent-production-ready-template/ ├── .env.example # 环境变量示例文件 ├── .gitignore ├── Dockerfile # 容器化构建文件 ├── docker-compose.yml # 本地开发与测试编排 ├── requirements.txt # Python 依赖清单 ├── pyproject.toml # 现代Python项目配置（可选，用于管理依赖和工具） ├── app/ # 应用核心代码目录 │ ├── __init__.py │ ├── main.py # FastAPI 应用入口点 │ ├── api/ # API 路由层 │ │ ├── __init__.py │ │ ├── endpoints/ # 各个API端点 │ │ │ ├── __init__.py │ │ │ ├── agent.py # 智能体交互端点 │ │ │ └── health.py # 健康检查端点 │ │ └── dependencies.py # 依赖注入（如获取数据库会话、验证Token） │ ├── core/ # 核心配置与工具 │ │ ├── __init__.py │ │ ├── config.py # 配置加载（从环境变量） │ │ ├── logging.py # 结构化日志配置 │ │ └── security.py # 安全相关（如JWT认证） │ ├── models/ # Pydantic 数据模型 │ │ ├── __init__.py │ │ ├── request.py # 请求体模型 │ │ └── response.py # 响应体模型 │ ├── schemas/ # SQLAlchemy 数据模型（如果用到数据库） │ │ └── __init__.py │ ├── agents/ # LangGraph 智能体定义（核心！） │ │ ├── __init__.py │ │ ├── base.py # 基础智能体类，定义共享状态和工具 │ │ ├── research_agent.py # 示例：研究型智能体 │ │ ├── customer_support_agent.py # 示例：客服智能体 │ │ └── tools/ # 智能体可用的工具集 │ │ ├── __init__.py │ │ ├── web_search.py │ │ └── calculator.py │ ├── services/ # 业务逻辑层 │ │ ├── __init__.py │ │ └── agent_service.py # 封装 LangGraph 图的编译与执行 │ └── db/ # 数据库相关（可选） │ └── session.py ├── tests/ # 测试目录 │ ├── __init__.py │ ├── test_api.py │ └── test_agents.py └── scripts/ # 辅助脚本 └── start.sh

关键目录解读：

• app/agents/：这是灵魂所在。你定义的所有智能体（LangGraph 图）都放在这里。base.py可能会定义一个State类（使用TypedDict或 Pydantic），它描述了在整个图执行过程中流转的状态对象，比如messages（对话历史）、intermediate_steps（工具调用结果）等。每个具体的智能体文件则负责构建节点（函数）和边（条件判断），最终编译成一个可执行的Graph对象。
• app/api/endpoints/agent.py：这是智能体与外界交互的桥梁。它接收 HTTP 请求（通常包含用户查询和会话ID），调用agent_service.py中的方法执行图，并返回流式或非流式响应。
• app/services/agent_service.py：它充当控制器。负责从配置或工厂中获取编译好的图实例，传入初始状态，执行图，并处理执行过程中的异常。它也可能负责会话的持久化（将状态保存到数据库）。
• app/core/config.py：使用pydantic-settings库是当前最佳实践。它能自动从.env文件、环境变量中加载配置，并进行强类型验证，确保应用启动时配置就是正确的。

注意：在实际使用模板时，第一件事就是仔细阅读README.md和这个目录结构。尝试运行起示例，然后沿着请求的流向（API -> Service -> Agent）走一遍代码，这是理解框架最快的方式。

4. 构建你的第一个生产级智能体：以“天气查询助手”为例

让我们脱离抽象的讲解，通过一个具体的例子——构建一个“天气查询助手”智能体，来演示如何使用这个模板。这个智能体能理解用户关于天气的询问，调用外部 API 获取实时天气，并组织成友好的回复。

4.1 步骤一：环境准备与配置

首先，克隆模板仓库并安装依赖。假设模板使用uv或poetry作为包管理工具（现代模板的趋势）。

# 克隆项目 git clone <repository-url> cd fastapi-langgraph-agent-production-ready-template # 复制环境变量示例文件并配置 cp .env.example .env # 编辑 .env 文件，填入你的 OpenAI API Key 和天气 API Key # OPENAI_API_KEY=sk-... # WEATHER_API_KEY=your_weather_api_key # MODEL_NAME=gpt-4o-mini # 安装依赖（以 poetry 为例） poetry install poetry shell

接下来，查看app/core/config.py，了解配置是如何定义的。你可能会看到类似下面的代码：

from pydantic_settings import BaseSettings from pydantic import Field class Settings(BaseSettings): # LLM 配置 openai_api_key: str = Field(..., validation_alias="OPENAI_API_KEY") model_name: str = Field("gpt-4o-mini", validation_alias="MODEL_NAME") # 工具配置 weather_api_key: str = Field(..., validation_alias="WEATHER_API_KEY") weather_api_base_url: str = Field("https://api.weatherapi.com/v1", validation_alias="WEATHER_API_BASE_URL") # 应用配置 app_name: str = "LangGraph Agent Service" log_level: str = "INFO" class Config: env_file = ".env" extra = "ignore" # 忽略未定义的额外环境变量 settings = Settings()

这种配置方式非常清晰和安全，所有秘密都来自环境变量，代码中不出现硬编码。

4.2 步骤二：定义智能体状态与工具

在app/agents/目录下，我们创建天气助手相关的文件。首先，在tools/目录下创建weather.py。

app/agents/tools/weather.py:

import requests from typing import TypedDict from app.core.config import settings from langchain_core.tools import tool # 定义工具输入参数的 Schema class WeatherQueryInput(TypedDict): location: str # 城市名，如 "Beijing" days: int # 预报天数 @tool(args_schema=WeatherQueryInput) def get_current_weather(location: str, days: int = 1) -> str: """ 获取指定城市当前和未来的天气情况。 Args: location: 城市名称，例如 'London' 或 '北京'。 days: 需要预报的天数，默认为1（今天）。 Returns: 格式化的天气信息字符串。 """ try: # 调用外部天气API，这里以 weatherapi.com 为例 url = f"{settings.weather_api_base_url}/forecast.json" params = { "key": settings.weather_api_key, "q": location, "days": days, "aqi": "no", "alerts": "no" } response = requests.get(url, params=params, timeout=10.0) response.raise_for_status() # 如果状态码不是200，抛出HTTPError data = response.json() current = data['current'] forecast_day = data['forecast']['forecastday'][0]['day'] result = ( f"{location}的天气：\n" f"- 当前温度：{current['temp_c']}°C (体感 {current['feelslike_c']}°C)\n" f"- 状况：{current['condition']['text']}\n" f"- 湿度：{current['humidity']}%\n" f"- 风速：{current['wind_kph']} km/h\n" f"- 今日预报：最高 {forecast_day['maxtemp_c']}°C，最低 {forecast_day['mintemp_c']}°C" ) return result except requests.exceptions.RequestException as e: return f"获取天气信息失败：网络或API错误 - {str(e)}" except KeyError as e: return f"解析天气API响应失败：缺少关键字段 - {str(e)}"

实操心得：工具函数是智能体与真实世界交互的抓手。这里有几个关键点：1)严格的输入输出定义：使用TypedDict或 Pydantic 明确参数，这有助于 LangChain/LangGraph 进行准确的工具调用解析。2)全面的错误处理：外部 API 调用可能因网络、权限、格式等问题失败，必须用try-except包裹，并返回对智能体友好的错误信息，而不是直接抛出异常导致图执行中断。3)超时设置：为网络请求设置超时（如timeout=10.0），避免一个缓慢的 API 拖垮整个智能体响应。

接下来，定义智能体的状态。在app/agents/base.py或新建的app/agents/weather_agent.py中，我们定义状态。

app/agents/weather_agent.py(部分):

from typing import TypedDict, List, Annotated from typing_extensions import TypedDict import operator from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage from app.core.config import settings from app.agents.tools.weather import get_current_weather # 1. 定义状态结构 class AgentState(TypedDict): # 消息历史，LangGraph 的 `add_messages` 操作符会自动处理列表的追加 messages: Annotated[List[BaseMessage], add_messages] # 可以添加其他状态，如用户ID、会话ID等，用于持久化 session_id: str # 2. 初始化LLM，并绑定工具 llm = ChatOpenAI( api_key=settings.openai_api_key, model=settings.model_name, temperature=0 # 生产环境通常降低随机性 ) llm_with_tools = llm.bind_tools([get_current_weather])

4.3 步骤三：构建 LangGraph 图

继续在weather_agent.py中构建图。我们将实现一个经典的“ReAct”风格循环：模型思考 -> 决定是否调用工具 -> 执行工具 -> 将结果返回给模型。

# 3. 定义图节点（函数） def call_model(state: AgentState): """ 节点：调用大语言模型。 模型会根据对话历史和绑定的工具，决定下一步是回复用户还是调用工具。 """ messages = state['messages'] response = llm_with_tools.invoke(messages) # 返回的结果是一个 AIMessage，如果决定调用工具，其 `tool_calls` 属性会包含信息 return {"messages": [response]} def call_tool(state: AgentState): """ 节点：执行工具调用。 处理模型返回的多个工具调用请求，并行执行，并收集结果。 """ messages = state['messages'] last_message = messages[-1] # 最新的消息应该是模型的 AIMessage tool_messages = [] if hasattr(last_message, 'tool_calls') and last_message.tool_calls: for tool_call in last_message.tool_calls: tool_name = tool_call['name'] tool_args = tool_call['args'] # 根据工具名分发执行 if tool_name == "get_current_weather": result = get_current_weather.invoke(tool_args) else: result = f"未知工具：{tool_name}" # 构造 ToolMessage，这是 LangGraph 约定的格式，用于将结果反馈给模型 tool_messages.append(ToolMessage(content=result, tool_call_id=tool_call['id'])) return {"messages": tool_messages} # 4. 定义路由逻辑（边） def should_continue(state: AgentState) -> str: """ 路由函数：根据最新消息决定下一步是调用工具还是结束。 """ messages = state['messages'] last_message = messages[-1] # 如果模型的最新消息包含工具调用，则路由到 `call_tool` 节点 if hasattr(last_message, 'tool_calls') and last_message.tool_calls: return "call_tool" # 否则，结束流程 return END # 5. 组装图 def create_weather_agent_graph(): workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("agent", call_model) # “agent”节点负责思考 workflow.add_node("action", call_tool) # “action”节点负责执行 # 设置入口点 workflow.set_entry_point("agent") # 添加条件边 workflow.add_conditional_edges( "agent", should_continue, # 根据这个函数的返回值决定路由 { "call_tool": "action", # 如果需要调用工具，去“action”节点 END: END # 否则结束 } ) # 从“action”节点执行完工具后，无条件回到“agent”节点继续思考 workflow.add_edge("action", "agent") # 编译图 return workflow.compile() # 创建图实例 weather_agent_graph = create_weather_agent_graph()

这个图定义了一个简单的循环：agent-> （判断）->action->agent-> ... 直到模型不再调用工具，最终走向END。

4.4 步骤四：集成到 FastAPI 服务

现在我们需要将这个图暴露为 API。首先在app/services/agent_service.py中创建一个服务层。

app/services/agent_service.py:

import asyncio from typing import AsyncIterator, Dict, Any from langchain_core.messages import HumanMessage from app.agents.weather_agent import weather_agent_graph, AgentState import logging logger = logging.getLogger(__name__) class AgentService: def __init__(self): # 可以在这里初始化不同的图，或者使用工厂模式 self.graph = weather_agent_graph async def run_agent(self, query: str, session_id: str, stream: bool = False) -> AsyncIterator[Dict[str, Any]]: """ 运行智能体。 Args: query: 用户输入。 session_id: 会话标识，可用于恢复历史。 stream: 是否启用流式输出。 Yields: 流式响应中的每个块（chunk）。 """ # 1. 构造初始状态 initial_state: AgentState = { "messages": [HumanMessage(content=query)], "session_id": session_id } # 2. 非流式执行（简单场景） if not stream: try: final_state = await asyncio.to_thread(self.graph.invoke, initial_state) # 提取最终模型回复（最后一条非工具消息） final_messages = final_state.get("messages", []) for msg in reversed(final_messages): if msg.type == "ai": # 找到最后一条 AI 消息 yield {"type": "final", "content": msg.content} break except Exception as e: logger.error(f"Agent execution failed: {e}", exc_info=True) yield {"type": "error", "content": "智能体处理请求时发生内部错误。"} # 3. 流式执行（高级场景，展示中间思考过程） else: try: # LangGraph 的 astream 方法支持流式输出 async for event in self.graph.astream(initial_state): # event 可能包含节点名、状态更新等信息 # 这里简化处理，只输出模型生成的内容块 if "agent" in event and "messages" in event["agent"]: new_msg = event["agent"]["messages"][-1] if hasattr(new_msg, 'content'): yield {"type": "chunk", "content": new_msg.content} elif "action" in event: # 可以在这里选择是否将工具执行结果也流式返回给前端 pass except Exception as e: logger.error(f"Agent streaming failed: {e}", exc_info=True) yield {"type": "error", "content": "流式处理中断。"}

然后，在app/api/endpoints/agent.py中创建对应的 API 端点。

app/api/endpoints/agent.py:

from fastapi import APIRouter, HTTPException, Depends from fastapi.responses import StreamingResponse from pydantic import BaseModel from app.services.agent_service import AgentService import uuid import json router = APIRouter(prefix="/api/v1/agent", tags=["agent"]) agent_service = AgentService() # 依赖注入更好，这里简化 class AgentRequest(BaseModel): query: str session_id: str | None = None # 如果为空，则创建新会话 stream: bool = False @router.post("/weather/chat") async def chat_with_weather_agent(request: AgentRequest): """ 与天气助手智能体对话。 """ session_id = request.session_id or str(uuid.uuid4()) # 非流式响应 if not request.stream: response_chunks = [] async for chunk in agent_service.run_agent(request.query, session_id, stream=False): if chunk["type"] == "final": return { "session_id": session_id, "response": chunk["content"], "status": "success" } elif chunk["type"] == "error": raise HTTPException(status_code=500, detail=chunk["content"]) raise HTTPException(status_code=500, detail="No response generated.") # 流式响应 (Server-Sent Events) async def event_generator(): try: async for chunk in agent_service.run_agent(request.query, session_id, stream=True): if chunk["type"] == "chunk": # 格式化为 SSE 格式: data: {json}\n\n yield f"data: {json.dumps({'content': chunk['content']})}\n\n" elif chunk["type"] == "error": yield f"data: {json.dumps({'error': chunk['content']})}\n\n" break except Exception as e: yield f"data: {json.dumps({'error': 'Stream generation failed.'})}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" # 对 Nginx 代理有用 } )

至此，一个具备完整流程的天气查询助手智能体后端就构建完成了。启动服务后，你可以通过POST /api/v1/agent/weather/chat与之交互。

5. 生产环境部署与运维要点

将开发好的服务部署到生产环境，模板提供的 Dockerfile 和配置是起点，但还需要考虑更多。

5.1 容器化与编排

模板的 Dockerfile 通常是一个多阶段构建，确保最终镜像尽可能小。

# 第一阶段：构建依赖 FROM python:3.11-slim as builder WORKDIR /app COPY requirements.txt . RUN pip install --user --no-cache-dir -r requirements.txt # 第二阶段：运行环境 FROM python:3.11-slim WORKDIR /app # 从构建阶段复制已安装的包 COPY --from=builder /root/.local /root/.local # 复制应用代码 COPY ./app ./app COPY .env . # 注意：生产环境通常通过 secrets 管理，而非复制 .env 文件 # 确保 PATH 包含用户安装目录 ENV PATH=/root/.local/bin:$PATH # 声明端口 EXPOSE 8000 # 使用 gunicorn 或 uvicorn 作为生产服务器 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

关键调整：

• .env文件：生产环境绝对不能将包含密钥的.env文件打入镜像或存入代码仓库。应使用 Docker Secrets、Kubernetes Secrets 或云服务商的密钥管理服务（如 AWS Secrets Manager）。
• 工作进程数：--workers参数需要根据容器可用的 CPU 核心数进行调整。通常设置为(2 * CPU cores) + 1。对于 CPU 密集型的 LLM 应用（如果进行本地推理），需要谨慎测试。
• 基础镜像：考虑使用更安全的镜像，如python:3.11-slim-bullseye，并定期更新以修补安全漏洞。

使用docker-compose.prod.yml或 Kubernetes Deployment 来编排服务、数据库（如用于存储会话历史的 Redis/PostgreSQL）和监控组件。

5.2 监控、日志与可观测性

生产就绪的核心是可观测性。模板可能集成了基础日志，但你需要强化它。

1. 结构化日志：确保所有日志都通过structlog或json-logger输出为 JSON。在 Kubernetes 中，使用 Fluentd 或 Filebeat 收集日志并发送到 Elasticsearch。
2. 指标（Metrics）：使用 Prometheus 客户端库（如prometheus-fastapi-instrumentator）暴露关键指标：
   • http_request_duration_seconds：API 延迟。
   • http_requests_total：请求总量和状态码。
   • agent_execution_duration_seconds：智能体单次运行耗时。
   • tool_call_count：各工具被调用的次数（用于成本分析和异常检测）。
3. 分布式追踪：集成 OpenTelemetry，追踪一个用户请求从进入 FastAPI，到 LangGraph 图中各个节点的执行路径。这对于调试复杂的、多步骤的智能体流程至关重要，能清晰看到时间消耗在模型调用还是工具调用上。
4. 健康检查：模板提供的/health端点应该进行深度检查，比如测试是否能连接到 LLM API、数据库等关键外部依赖。

5.3 性能优化与成本控制

AI 应用的成本和性能是生产环境的生命线。

1. 缓存：对于频繁出现的、结果不变的查询（如“北京今天的天气”），引入缓存层（Redis）。可以在 API 网关层或服务层的工具调用前加入缓存逻辑。
2. 速率限制：在 FastAPI 层或 API 网关（如 Nginx, Kong）对用户或 API Key 进行速率限制，防止滥用。
3. LLM 调用优化：
   • 批处理：如果支持，将多个独立请求合并为一个批处理请求发送给 LLM API。
   • 上下文管理：智能清理对话历史，避免无限增长的上下文消耗大量 Token。可以总结历史对话或只保留最近 N 轮。
   • 模型降级：在非关键路径或简单查询上，使用更便宜、更快的模型（如gpt-3.5-turbo而非gpt-4）。
4. 异步与并发：确保你的工具函数和 LLM 调用客户端都是异步的（使用httpx.AsyncClient,openai.AsyncOpenAI），充分利用 FastAPI 的异步优势处理高并发。

6. 常见问题排查与调试技巧

在实际开发和运维中，你肯定会遇到各种问题。以下是一些常见场景和排查思路。

6.1 智能体陷入循环或行为异常

症状：智能体不停地调用同一个工具，或者回复内容与预期不符。排查：

1. 检查状态（State）：在call_model和call_tool节点函数中打印state。确认传递给模型的消息历史是否正确，工具执行结果是否被正确格式化为ToolMessage并追加到了状态中。
2. 审查工具描述：LLM 根据工具的函数名和 Docstring 来决定是否以及如何调用。确保你的工具描述清晰、准确。过于模糊的描述可能导致误调用。
3. 调整提示词（Prompt）：模型的系统提示词（System Prompt）对行为有决定性影响。你可能需要在调用llm_with_tools.invoke之前，在messages列表的开头插入一条SystemMessage，明确指令，例如：“你是一个专业的天气助手，只回答与天气相关的问题。如果用户询问其他话题，请礼貌地告知你的职责范围。”
4. 使用langgraph的调试视图：LangGraph 提供了可视化图执行过程的方法。你可以将图的执行记录导出，帮助理解控制流。

6.2 API 响应慢或超时

症状：前端请求长时间等待或收到 504 Gateway Timeout。排查：

1. 定位瓶颈：使用分布式追踪或详细的计时日志，记录每个工具调用和 LLM 调用的耗时。瓶颈通常在外部 API（天气 API、LLM API）。
2. 设置超时：为所有外部 HTTP 请求（requests.get,openai.chat.completions.create）设置合理的超时和重试策略。在 FastAPI 层面，也可以使用timeout中间件。
3. 检查并发量：监控服务的并发连接数和请求队列。如果使用同步的 HTTP 客户端，在高并发下会迅速耗尽资源。务必换用异步客户端。
4. 数据库连接池：如果使用了数据库，检查连接池配置。连接泄漏或池大小不足会导致请求排队。

6.3 流式输出中断或不完整

症状：SSE 流提前关闭，或者前端收不到完整的回复。排查：

1. 网络代理与缓冲：检查 Nginx 或云负载均衡器的配置。确保它们没有缓冲代理响应（设置proxy_buffering off;和X-Accel-Buffering: no头部）。
2. 服务端异常：确保event_generator函数被完整的try-except包裹，任何异常都要以正确的 SSE 格式输出错误信息，而不是抛出未捕获的异常导致连接中断。
3. 心跳机制：对于长时间的流式响应（如生成长篇报告），可以定期发送冒号开头的注释行（:keepalive\n\n）作为心跳，防止代理或客户端因超时断开连接。

6.4 部署后配置不生效

症状：在 Kubernetes 或 Docker Swarm 中更新了环境变量，但服务仍然使用旧值。排查：

1. 配置热重载：像pydantic-settings这样的库通常在应用启动时加载配置。更改环境变量后，需要重启 Pod 或容器。考虑实现一个配置热重载机制（如监听 ConfigMap 变化），或者使用专门配置中心。
2. Secret 挂载：确认 Secrets 是否已正确挂载到容器内的文件路径或环境变量。进入容器内部，执行env | grep YOUR_KEY或cat /path/to/secret/file进行验证。
3. 依赖服务发现：如果智能体需要调用集群内其他服务（如内部知识库），确保使用 Kubernetes Service DNS 名称（如http://internal-knowledge-base:8080）而非 localhost 或硬编码 IP。

这个模板提供了一个强大的起点，但真正的“生产就绪”是一个持续的过程，需要你根据具体的业务负载、可靠性要求和运维能力，不断地打磨和加固每一个环节。从清晰的架构设计开始，重视配置、日志、监控这些“非功能性需求”，你就能构建出不仅能用，而且好用、耐用的 AI 智能体服务。