LangGraph 多 Agent 实战落地｜打造可商用的智能旅行预订系统，含前后端 + Docker 部署

引言

在前四篇博客中，我们从A2A 协议理论→LangGraph 基础开发→Handoffs 智能体移交→企业级 Supervisor 架构优化，完成了多智能体协作的全技术栈学习，打造了高可用、高扩展的企业级 Supervisor 调度架构。

但技术的最终价值在于商业落地，很多开发者在完成架构设计后，会卡在工程化落地环节：如何将多智能体架构封装为后端接口？如何开发简单易用的前端交互页面？如何实现项目的容器化部署？如何保证生产环境的稳定性和可维护性？

本文作为 LangGraph 多智能体系列的最终落地篇，将把前序的企业级 Supervisor 架构与实际业务结合，打造一套可直接商用的智能旅行预订系统，包含FastAPI 后端接口、Streamlit 前端交互、Docker 容器化部署、生产环境优化四大核心部分，实现从本地开发到线上商用的全流程落地，让你真正掌握多智能体项目的商业落地能力。

本文所有代码均基于 Python3.8 + 实现，前后端分离设计，容器化部署后可直接在服务器上线运行，适配企业级生产环境。

一、项目整体架构设计：商用级工程化落地

为保证项目的可维护性、可扩展性、高可用性，采用前后端分离 + 分层架构的设计思路，同时融入工程化最佳实践（如配置分离、日志规范、异常处理、接口文档），打造符合企业级商用标准的智能旅行预订系统。

1. 项目整体技术栈

2. 项目分层架构设计

采用五层架构设计，各层之间解耦独立，每层仅负责自身核心职责，便于后续维护和扩展：

┌─────────────────────────────────────────────────────────┐
│ 前端交互层（Streamlit）：用户操作页面、预订进度展示、结果展示 │
└───────────────────┬─────────────────────────────────────┘
│
┌───────────────────▼─────────────────────────────────────┐
│ 后端接口层（FastAPI）：提供RESTful接口、参数校验、身份认证 │
└───────────────────┬─────────────────────────────────────┘
│
┌───────────────────▼─────────────────────────────────────┐
│ 核心业务层（LangGraph）：企业级Supervisor调度、多智能体协作 │
└───────────────────┬─────────────────────────────────────┘
│
┌───────────────────▼─────────────────────────────────────┐
│ 工具层（Python）：专业预订工具、外部API对接、数据处理 │
└───────────────────┬─────────────────────────────────────┘
│
┌───────────────────▼─────────────────────────────────────┐
│ 基础设施层：配置管理、日志系统、异常处理、Docker部署 │
└─────────────────────────────────────────────────────────┘

3. 核心业务流程

用户通过前端页面提交旅行预订需求→前端调用后端 FastAPI 接口→后端调用企业级 Supervisor 调度核心→多智能体协作完成预订→结果结构化返回→前端展示预订结果和进度，完整流程无人工干预，实现自动化旅行预订。

4. 项目目录结构

遵循Python 工程化最佳实践，设计清晰的目录结构，便于团队协作和后续维护，所有核心代码按功能模块划分：

smart_travel_booking/
├── app/ # 项目核心代码目录
│ ├── api/ # 后端接口层（FastAPI）
│ │ ├── __init__.py
│ │ └── v1/ # API版本v1
│ │ ├── __init__.py
│ │ └── booking.py # 预订核心接口
│ ├── core/ # 核心业务层（多智能体架构）
│ │ ├── __init__.py
│ │ ├── supervisor.py # 企业级Supervisor调度
│ │ ├── agent_center.py # 智能体能力注册中心
│ │ ├── task_parser.py # 动态任务拆解
│ │ ├── fault_tolerant.py# 容错执行逻辑
│ │ └── result_integrate.py # 结果整合
│ ├── tools/ # 工具层
│ │ ├── __init__.py
│ │ ├── flight.py # 航班预订工具
│ │ ├── hotel.py # 酒店预订工具
│ │ ├── pickup.py # 接送机预订工具
│ │ └── scenic.py # 景点预订工具
│ ├── common/ # 公共组件层
│ │ ├── __init__.py
│ │ ├── config.py # 配置管理
│ │ ├── logger.py # 日志系统
│ │ ├── models.py # 数据模型
│ │ └── exceptions.py # 自定义异常
│ └── main.py # 项目入口文件（FastAPI启动）
├── frontend/ # 前端交互层（Streamlit）
│ ├── __init__.py
│ └── main.py # 前端入口文件
├── .env.development # 开发环境配置
├── .env.production # 生产环境配置
├── Dockerfile # 后端Dockerfile
├── Dockerfile-frontend # 前端Dockerfile
├── docker-compose.yml # Docker Compose一键部署配置
├── requirements.txt # 后端依赖
├── frontend-requirements.txt # 前端依赖
└── README.md # 项目说明文档

目录设计亮点：

二、后端开发：FastAPI 封装多智能体调度接口

后端采用FastAPI开发，作为多智能体架构与前端的中间层，核心职责是封装调度接口、做参数校验、统一异常处理、自动生成接口文档，同时融入工程化最佳实践，保证接口的高可用性和可维护性。

1. 前置准备：环境配置与公共组件开发

1.1 配置管理（app/common/config.py）

实现多环境配置加载，支持开发 / 生产环境切换，配置与代码完全分离：

import os
from dotenv import load_dotenv
from pydantic_settings import BaseSettings

# 加载环境变量：根据ENVIRONMENT选择配置文件
env = os.getenv("ENVIRONMENT", "development")
load_dotenv(f".env.{env}")

class Settings(BaseSettings):
"""项目配置类"""
# 项目基础配置
PROJECT_NAME: str = "智能旅行预订系统"
PROJECT_VERSION: str = "v1.0.0"
API_PREFIX: str = "/api/v1"
# 服务器配置
HOST: str = os.getenv("HOST", "0.0.0.0")
PORT: int = int(os.getenv("PORT", 8000))
RELOAD: bool = env == "development"
# 大模型配置
LLM_API_KEY: str = os.getenv("aliQwen-api")
LLM_BASE_URL: str = os.getenv("llm_base_url", "https://dashscope.aliyuncs.com/compatible-mode/v1")
LLM_MODEL: str = os.getenv("llm_model", "qwen-plus")
LLM_TEMPERATURE: float = float(os.getenv("llm_temperature", 0.1))
# 容错配置
RETRY_TIMES: int = int(os.getenv("RETRY_TIMES", 3))
RETRY_INTERVAL: float = float(os.getenv("RETRY_INTERVAL", 1.0))

# 实例化配置对象，全局使用
settings = Settings()

配置文件示例（.env.development）：

# 环境标识
ENVIRONMENT=development
# 服务器配置
HOST=0.0.0.0
PORT=8000
# 大模型配置
aliQwen-api=你的通义千问API_KEY
llm_base_url=https://dashscope.aliyuncs.com/compatible-mode/v1
llm_model=qwen-plus
llm_temperature=0.1
# 容错配置
RETRY_TIMES=3
RETRY_INTERVAL=1.0

1.2 日志系统（app/common/logger.py）

实现标准化日志系统，支持控制台输出 + 文件输出，按日期分割日志文件，便于生产环境日志排查：

import os
import logging
from logging.handlers import TimedRotatingFileHandler
from app.common.config import settings

# 创建日志目录
log_dir = "logs"
if not os.path.exists(log_dir):
os.makedirs(log_dir)

# 定义日志格式
log_format = logging.Formatter(
"%(asctime)s – %(name)s – %(levelname)s – %(module)s:%(lineno)d – %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)

def get_logger(name: str = __name__) -> logging.Logger:
"""
获取标准化日志对象
Args:
name: 日志名称
Returns:
logging.Logger: 日志对象
"""
logger = logging.getLogger(name)
logger.setLevel(logging.INFO)
logger.propagate = False

# 控制台处理器
console_handler = logging.StreamHandler()
console_handler.setFormatter(log_format)
console_handler.setLevel(logging.INFO)

# 文件处理器（按日期分割，保留7天）
file_handler = TimedRotatingFileHandler(
filename=os.path.join(log_dir, "smart_travel.log"),
when="D",
interval=1,
backupCount=7,
encoding="utf-8"
)
file_handler.setFormatter(log_format)
file_handler.setLevel(logging.INFO)

# 添加处理器
if not logger.handlers:
logger.addHandler(console_handler)
logger.addHandler(file_handler)

return logger

# 实例化全局日志对象
logger = get_logger("smart_travel")

1.3 数据模型（app/common/models.py）

复用前序博客的 Pydantic 数据模型，统一前后端数据交互格式：

from pydantic import BaseModel, Field
from typing import List, Dict, Optional, Callable

# 子任务数据模型
class SubTask(BaseModel):
task_name: str = Field(description="子任务名称")
task_desc: str = Field(description="子任务详情")
priority: int = Field(description="执行优先级")
required_ability: str = Field(description="所需智能体能力")

# 智能体能力模型
class AgentAbility(BaseModel):
agent_name: str = Field(description="智能体名称")
agent_ability: str = Field(description="核心能力")
agent_tools: List[str] = Field(description="绑定工具")
agent_obj: Optional[Callable] = Field(description="智能体对象", default=None)

# 预订请求模型（前端入参）
class BookingRequest(BaseModel):
user_request: str = Field(description="用户旅行预订需求", min_length=5, max_length=500)

# 预订响应模型（后端出参）
class BookingResponse(BaseModel):
code: int = Field(description="响应码，200成功，500失败")
msg: str = Field(description="响应信息")
data: Optional[Dict] = Field(description="结构化预订结果", default=None)

# 全局统一响应模型
class GlobalResponse(BaseModel):
code: int = Field(200, description="业务响应码")
msg: str = Field("success", description="业务描述")
data: Optional[any] = Field(default=None, description="业务数据")

1.4 自定义异常（app/common/exceptions.py）

实现项目专属的自定义异常，便于统一异常处理和错误排查：

class SmartTravelException(Exception):
"""智能旅行预订系统基础异常"""
def __init__(self, code: int = 500, msg: str = "系统内部错误"):
self.code = code
self.msg = msg
super().__init__(self.msg)

class TaskParseException(SmartTravelException):
"""任务解析异常"""
def __init__(self, msg: str = "任务解析失败"):
super().__init__(code=400, msg=msg)

class AgentMatchException(SmartTravelException):
"""智能体匹配异常"""
def __init__(self, msg: str = "智能体匹配失败"):
super().__init__(code=400, msg=msg)

class TaskExecuteException(SmartTravelException):
"""任务执行异常"""
def __init__(self, msg: str = "任务执行失败"):
super().__init__(code=500, msg=msg)

class ResultIntegrateException(SmartTravelException):
"""结果整合异常"""
def __init__(self, msg: str = "结果整合失败"):
super().__init__(code=500, msg=msg)

2. 核心业务层移植与适配

将前序博客的企业级 Supervisor 架构（任务拆解、能力注册、容错执行、结果整合）移植到app/core/目录，并基于后端配置和日志系统做适配，替换全局变量为配置对象，替换打印为标准化日志，保证与后端工程化融合。

核心适配点：

3. 后端接口开发（app/api/v1/booking.py）

开发旅行预订核心接口，基于 FastAPI 实现参数校验、接口封装、异常捕获、统一响应，同时支持同步调用，适配前端交互：

from fastapi import APIRouter, Body, HTTPException
from app.common.models import BookingRequest, GlobalResponse
from app.common.logger import logger
from app.common.exceptions import SmartTravelException
from app.core.supervisor import supervisor_dispatch

# 创建API路由
router = APIRouter(prefix="/booking", tags=["旅行预订"])

@router.post("/dispatch", response_model=GlobalResponse, summary="旅行预订核心调度接口")
async def booking_dispatch(booking_request: BookingRequest = Body(…, description="用户预订需求")):
"""
旅行预订核心调度接口：接收用户需求，调用多智能体架构完成预订，返回结构化结果
– **user_request**: 用户旅行预订需求（如：订北京到上海的航班+上海外滩酒店）
"""
try:
logger.info(f"接收预订请求：{booking_request.user_request}")
# 调用企业级Supervisor调度核心
dispatch_result = supervisor_dispatch(booking_request.user_request)
logger.info(f"预订请求执行完成，结果：{dispatch_result[:100]}…")
# 统一响应格式返回
return GlobalResponse(
code=200,
msg="预订调度执行成功",
data=dispatch_result
)
except SmartTravelException as e:
# 捕获项目自定义异常
logger.error(f"预订调度失败：{e.msg}", exc_info=True)
return GlobalResponse(
code=e.code,
msg=e.msg,
data=None
)
except Exception as e:
# 捕获全局未知异常
logger.error(f"预订调度未知失败：{str(e)}", exc_info=True)
return GlobalResponse(
code=500,
msg=f"系统内部错误：{str(e)}",
data=None
)

4. 后端入口文件（app/main.py）

整合路由、配置、日志，实现 FastAPI 服务一键启动，自动生成接口文档（Swagger/Redoc）：

from fastapi import FastAPI
from app.api.v1.booking import router as booking_router
from app.common.config import settings
from app.common.logger import logger

# 初始化FastAPI应用
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.PROJECT_VERSION,
docs_url="/docs", # Swagger文档地址
redoc_url="/redoc" # Redoc文档地址
)

# 注册API路由
app.include_router(booking_router, prefix=settings.API_PREFIX)

# 根路径健康检查
@app.get("/", summary="服务健康检查")
async def health_check():
return {"code": 200, "msg": f"{settings.PROJECT_NAME}服务运行正常", "version": settings.PROJECT_VERSION}

# 启动事件
@app.on_event("startup")
async def startup_event():
logger.info(f"【服务启动】{settings.PROJECT_NAME} v{settings.PROJECT_VERSION}")
logger.info(f"【接口文档】http://{settings.HOST}:{settings.PORT}/docs")

# 主函数启动
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main:app",
host=settings.HOST,
port=settings.PORT,
reload=settings.RELOAD
)

三、前端开发：Streamlit 快速实现交互页面

前端采用Streamlit开发，无需前端框架基础，纯 Python 编写即可实现轻量、美观的 Web 交互页面，核心实现用户需求输入、预订进度展示、结构化结果渲染，适配普通用户操作习惯。

1. 前端核心代码（frontend/main.py）

import streamlit as st
import requests
import json
import time

# 页面基础配置
st.set_page_config(
page_title="智能旅行预订系统",
page_icon="✈️",
layout="wide"
)

# 后端接口地址（Docker部署时使用服务名，本地使用localhost）
API_URL = st.secrets.get("API_URL", "http://backend:8000/api/v1/booking/dispatch")

# 页面标题
st.title("✈️ 智能旅行预订系统")
st.divider()

# 左侧：用户输入
with st.sidebar:
st.header("📝 预订需求输入")
user_request = st.text_area(
label="请输入你的旅行预订需求（支持多任务、自定义顺序）",
placeholder="例如：先订成都春熙路希尔顿酒店（5月1日入住），再订北京大兴到成都天府的航班，最后订大熊猫基地景点门票",
height=200,
key="user_request"
)
submit_btn = st.button("🚀 提交预订请求", type="primary", use_container_width=True)

# 右侧：结果展示
st.header("📊 预订结果展示")
result_container = st.empty()
log_container = st.expander("🔍 执行日志详情", expanded=False)

# 提交请求逻辑
if submit_btn and user_request:
with st.spinner("正在处理你的预订请求，请稍候…"):
try:
# 调用后端接口
response = requests.post(
url=API_URL,
json={"user_request": user_request},
timeout=60
)
response.raise_for_status()
res_data = response.json()

# 渲染结果
with result_container:
if res_data["code"] == 200:
st.success("✅ 预订请求执行完成！")
# 解析结构化结果
result = json.loads(res_data["data"])
# 展示汇总信息
st.info(f"**执行状态**：{result['status']}")
st.markdown(f"**结果汇总**：{result['summary']}")
st.divider()
# 展示各任务执行结果
st.subheader("📋 各任务执行详情")
for task_name, task_res in result["execute_result"].items():
st.write(f"**{task_name}**：{task_res}")
# 日志详情
with log_container:
st.subheader("🔍 任务执行日志")
for log in result["task_progress"]:
st.write(f"[{log['timestamp']}] {log['task_name']} | {log['agent_name']} | {log['status']}：{log['detail']}")
else:
st.error(f"❌ 预订请求执行失败：{res_data['msg']}")
except requests.exceptions.RequestException as e:
st.error(f"❌ 接口调用失败：{str(e)}")
except Exception as e:
st.error(f"❌ 数据解析失败：{str(e)}")
elif submit_btn and not user_request:
st.warning("⚠️ 请输入你的旅行预订需求！")

# 页面底部说明
st.divider()
st.caption("💡 支持功能：航班预订/酒店预订/接送机预订/景点预订，可自定义执行顺序，自动容错重试")

四、容器化部署：Docker+Docker Compose 一键上线

采用Docker+Docker Compose实现项目容器化部署，解决环境依赖问题，实现一键启动 / 停止，适配服务器、云主机等任意部署环境，符合企业级商用标准。

1. 后端 Dockerfile（Dockerfile）

# 基础镜像
FROM python:3.8-slim

# 设置工作目录
WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install –no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 复制项目代码
COPY . .

# 设置环境变量
ENV ENVIRONMENT=production
ENV PYTHONUNBUFFERED=1

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["python", "app/main.py"]

2. 前端 Dockerfile（Dockerfile-frontend）

# 基础镜像
FROM python:3.8-slim

# 设置工作目录
WORKDIR /app/frontend

# 安装依赖
COPY frontend-requirements.txt .
RUN pip install –no-cache-dir -r frontend-requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 复制前端代码
COPY frontend/ .

# 暴露端口
EXPOSE 8501

# 启动命令
CMD ["streamlit", "run", "main.py", "–server.port=8501", "–server.address=0.0.0.0"]

3. 依赖文件

后端依赖（requirements.txt）

fastapi>=0.104.1
uvicorn>=0.24.0
langchain>=0.1.0
langgraph>=0.0.24
pydantic-settings>=2.1.0
python-dotenv>=1.0.0
requests>=2.31.0
pydantic>=2.4.2

前端依赖（frontend-requirements.txt）

streamlit>=1.28.2
requests>=2.31.0

4. Docker Compose 配置（docker-compose.yml）

实现后端、前端服务一键编排，自动创建网络，服务间可通过服务名通信：

version: '3.8'

services:
# 后端服务
backend:
build:
context: .
dockerfile: Dockerfile
container_name: smart-travel-backend
restart: always
environment:
– ENVIRONMENT=production
– LLM_API_KEY=你的通义千问API_KEY
ports:
– "8000:8000"
networks:
– smart-travel-network

# 前端服务
frontend:
build:
context: .
dockerfile: Dockerfile-frontend
container_name: smart-travel-frontend
restart: always
ports:
– "8501:8501"
depends_on:
– backend
networks:
– smart-travel-network

# 自定义网络
networks:
smart-travel-network:
driver: bridge

5. 一键部署命令

将项目代码上传至服务器后，执行以下命令即可一键启动服务：

# 构建并启动容器（后台运行）
docker-compose up -d –build

# 查看服务状态
docker-compose ps

# 查看日志
docker-compose logs -f backend
docker-compose logs -f frontend

# 停止服务
docker-compose down

部署完成后，访问以下地址即可使用系统：

• 前端交互页面：http://服务器IP:8501
• 后端接口文档：http://服务器IP:8000/docs

五、生产环境优化与商用建议

1. 生产环境核心优化点

2. 商用落地建议

六、系列总结：LangGraph 多智能体从入门到商用全链路

本系列从A2A 协议基础到LangGraph 核心开发，再到企业级架构优化，最终实现商用系统落地，完成了 LangGraph 多智能体协作的全链路学习与实践，核心知识点总结如下：

核心思想：多智能体协作的本质是 “专业分工 + 标准化协作”，让每个智能体聚焦自身专业能力，通过统一的协作机制实现信息互通，最终完成复杂任务，这也是 A2A 协议的核心内涵。