手把手教你用 FastAPI + LangGraph搭建 AI 工作流
Large Language Models (LLMs) 擅長推理，但現實世界的應用往往需要有狀態、多步驟的工作流。這就是 LangGraph 的用武之地——它讓你可以通過由 LLM 驅動的節點圖來構建智能工作流。
但如果你想把這些工作流暴露為 APIs，讓其他應用（或用戶）可以調用呢？這時候 FastAPI 就派上用場了——一個輕量級、高性能的 Python Web 框架。
在這篇指南中，你將學習如何將 LangGraph 工作流封裝在 FastAPI 中，變成一個生產就緒的 endpoint。
為什么選擇 LangGraph + FastAPI？
?LangGraph：創建多步驟、有狀態的 LLM 工作流（例如，多智能體推理、數據處理）。
?FastAPI：輕松將這些工作流暴露為REST APIs，以便與 Web 應用、微服務或自動化流水線集成。
?結合兩者：構建可從任何地方訪問的可擴展 AI 智能體。
1. 項目設置
創建一個新項目文件夾并安裝依賴：
mkdir langgraph_fastapi_demo && cd langgraph_fastapi_demo
python -m venv .venv
source .venv/bin/activate # 在 Windows 上：.venv\Scripts\activate
pip install fastapi uvicorn langgraph langchain-openai python-dotenv創建一個 ??.env?? 文件來存儲你的 API 密鑰：
OPENAI_API_KEY=你的_openai_密鑰_在此2. 構建一個簡單的 LangGraph 工作流
讓我們構建一個簡單的 LangGraph，它接收用戶的問題并返回 AI 生成的答案。
# workflow.py
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import os
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(model="gpt-4o") # 可以切換到 gpt-4o-mini 以降低成本
# 定義狀態
defanswer_question(state: dict) -> dict:
user_input = state["user_input"]
response = llm.invoke([HumanMessage(cnotallow=user_input)])
return {"answer": response.content}
# 構建圖
workflow = StateGraph(dict)
workflow.add_node("answer", answer_question)
workflow.add_edge(START, "answer")
workflow.add_edge("answer", END)
graph = workflow.compile()這個圖：
? 接收user_input
? 將其發送到GPT-4o
? 返回 AI 生成的響應
3. 讓它生產就緒
在向全世界開放之前，讓我們為真實用例加固它。
錯誤處理與重試
LLM APIs 可能會失敗或超時。用 try/except 包裝調用：
from tenacity import retry, wait_exponential, stop_after_attempt
@retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(3))
def safe_invoke_llm(message):
return llm.invoke([HumanMessage(cnotallow=message)])
def answer_question(state: dict) -> dict:
user_input = state["user_input"]
try:
response = safe_invoke_llm(user_input)
return {"answer": response.content}
except Exception as e:
return {"answer": f"錯誤：{str(e)}"}輸入驗證
我們不想讓別人發送巨大的數據負載。添加 Pydantic 約束：
from pydantic import BaseModel, constr
class RequestData(BaseModel):
user_input: constr(min_length=1, max_length=500) # 限制輸入大小日志記錄
添加日志以提高可見性：
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def answer_question(state: dict) -> dict:
logger.info(f"收到輸入：{state['user_input']}")
response = safe_invoke_llm(state['user_input'])
logger.info("已生成 LLM 響應")
return {"answer": response.content}4. 使用 FastAPI 暴露工作流
現在，讓我們將這個工作流封裝在 FastAPI 中。
# main.py
from fastapi import FastAPI
from workflow import graph, RequestData
app = FastAPI()
@app.post("/run")
async def run_workflow(data: RequestData):
result = graph.invoke({"user_input": data.user_input})
return {"result": result["answer"]}運行服務器：
uvicorn main:app --reload5. 測試 API
你可以使用 curl 測試：
curl -X POST "http://127.0.0.1:8000/run" \
-H "Content-Type: application/json" \
-d '{"user_input":"什么是 LangGraph？"}'或者在瀏覽器中打開 ??http://127.0.0.1:8000/docs?? —— FastAPI 會自動為你生成 Swagger UI！
這個交互式 UI 讓你直接在瀏覽器中測試你的 endpoint。
6. 擴展與部署
為生產環境做準備的幾個步驟：
?異步執行：FastAPI是異步原生的。對于多個 LLM 調用，讓函數變成異步的。
?工作進程：使用多進程運行以實現并發：
uvicorn main:app --workers 4?Docker 化：
FROM python:3.11-slim
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]?認證：使用 API 密鑰或JWT tokens來保護endpoints（第二部分即將推出）。
7. 架構概覽
以下是整體連接方式：
POST /runClientFastAPILangGraphOpenAI_APIResponse這個簡單的架構讓你可以將任何 LangGraph 變成一個 API。
8. 結論
通過幾個簡單的步驟，我們：
? 構建了一個LangGraph工作流
? 使用FastAPI將其暴露為REST API
? 添加了生產就緒的功能（驗證、重試、日志）
? 為可擴展的 AI 微服務奠定了基礎
這個設置可以支持從聊天機器人到文檔處理器再到 AI SaaS 產品的各種應用。
本文轉載自??AI大模型觀察站??，作者：AI大模型觀察站
