🎯 前言：所有代码，最终只为「上线」

到此，我们21篇核心+番外全部技术能力已经闭环：工作流编排、多智能体、工具工程化、RAG知识库、Prompt质控、批量并发、长效记忆、人格固化。

但本地能跑 ≠ 线上可用。

95%的AI开发者卡在最后一步：代码写完，不会部署。本地运行正常、服务器报错、环境冲突、依赖缺失、端口不通、重启失效、无法对接前端、无法长期稳定运行。

本篇作为全套2026 LangGraph教程真正终章，不讲新复杂原理，只讲企业标准上线流水线。手把手完成：项目标准化封装、Docker容器化、云端服务器部署、接口调试、前后端联调、守护进程保活、线上稳定性优化。

读完本篇，你的AI Agent将彻底告别「本地Demo」，真正成为7×24小时在线、可商用、可访问、可迭代的线上产品。

一、为什么必须做容器化部署？

1.1 原生直接部署的致命痛点

• 本地环境、服务器环境不一致，出现「本地能跑线上崩」

• 依赖版本混乱、Python版本冲突、库版本打架

• 程序退出即停止，服务器断开连接项目就挂

• 无法快速迁移、无法版本回滚、无法批量扩容

• 多项目环境互相污染，运维极其混乱

1.2 Docker容器化的核心价值（企业标配）

• 一次打包，随处运行：环境完全隔离，本地、服务器、云平台运行效果完全一致

• 环境纯净：独立依赖、独立运行空间，零环境冲突

• 高可用保活：支持自动重启、后台常驻、异常自愈

• 极速运维：一键部署、一键更新、一键回滚、一键迁移

二、本篇落地完整上线流程

全程复刻互联网企业AI项目上线标准流水线：

1. 项目工程化整理：规范目录、剥离本地配置、统一入口

2. 依赖清单固化：生成requirements.txt，锁定版本

3. Dockerfile编写：工业级容器构建脚本

4. 容器镜像构建：打包完整项目环境

5. 云端服务器部署：远程拉取、构建、启动容器

6. 端口放行与跨域配置：解决访问拦截、跨域报错

7. 后台常驻保活：7×24小时不宕机

8. 前后端接口联调：前端直接调用AI Agent能力

9. 线上日志监控与排错：实时查看运行状态

三、第一步：项目标准化规整（上线前置）

3.1 标准项目目录结构

上线项目必须规范目录，杜绝杂乱文件：

langgraph-final-deploy/
├── .env                # 环境变量配置（密钥、模型地址）
├── requirements.txt    # 项目依赖清单
├── main.py             # 项目统一入口
├── Dockerfile          # 容器构建脚本
├── .dockerignore       # 忽略无用文件
└── rag_db/             # 知识库向量库持久化目录

3.2 生成固化依赖清单

锁定所有依赖版本，避免线上版本漂移：

pip freeze > requirements.txt

本项目核心依赖：langgraph、langchain、langchain-openai、python-dotenv、pydantic

3.3 .dockerignore 配置（精简镜像体积）

__pycache__/
*.pyc
*.pyo
*.pyd
.git/
.idea/
.vscode/
*.md
.DS_Store
.env.local
logs/

四、第二步：编写工业级Dockerfile（可直接商用）

该脚本为LangGraph AI项目通用容器模板，适配所有Agent项目：

# 基础镜像（稳定Python3.10企业版）
FROM python:3.10-slim

# 设置工作目录
WORKDIR /app

# 更换国内源，加速安装依赖
RUN pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

# 拷贝依赖文件、优先安装依赖（缓存优化构建速度）
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 拷贝全部项目代码
COPY . .

# 开放端口
EXPOSE 8000

# 项目启动命令
CMD ["python", "main.py"]

五、第三步：改造主程序为线上可服务模式

本地脚本直接运行无法对接前端，我们将终章项目改造为FastAPI接口服务，支持网络请求、跨域访问、前后端联调。

以下为最终上线版 main.py 完整代码（整合所有能力+接口化部署）：

from dotenv import load_dotenv
import os
import json
from typing import TypedDict, Literal, List
from functools import wraps
from pydantic import BaseModel, Field, ValidationError
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.tools import StructuredTool
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

# 加载环境变量
load_dotenv()

# 监控全局开启
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = os.getenv("LANGSMITH_API_KEY")
os.environ["LANGCHAIN_PROJECT"] = "LangGraph-线上商用部署项目"

# 初始化模型
llm = ChatOpenAI(
    api_key=os.getenv("API_KEY"),
    base_url=os.getenv("BASE_URL"),
    model="gpt-3.5-turbo",
    temperature=0.05
)
embedding = OpenAIEmbeddings(
    api_key=os.getenv("API_KEY"),
    base_url=os.getenv("BASE_URL")
)
memory = MemorySaver()

# ===================== 工具模块 =====================
class CalcParam(BaseModel):
    num1: float = Field(description="第一个运算数字")
    num2: float = Field(description="第二个运算数字")
    op: str = Field(description="运算符号：add加、sub减、mul乘、div除")

class TextParam(BaseModel):
    content: str = Field(description="需要处理的文本内容")
    action: str = Field(description="处理方式：upper大写、lower小写、len长度")

def tool_wrapper(param_model):
    def decorator(func):
        @wraps(func)
        def wrapper(params: dict) -> str:
            try:
                valid_params = param_model(**params).model_dump()
            except ValidationError as e:
                return f"【参数异常】{str(e)}"
            try:
                return func(**valid_params)
            except Exception as e:
                return f"【执行异常】{str(e)}"
        return wrapper
    return decorator

@tool_wrapper(CalcParam)
def calc_tool(num1: float, num2: float, op: str) -> str:
    if op == "add": res = num1 + num2
    elif op == "sub": res = num1 - num2
    elif op == "mul": res = num1 * num2
    elif op == "div":
        if num2 == 0: return "【运算错误】除数不能为0"
        res = num1 / num2
    else: return "【参数错误】仅支持add/sub/mul/div"
    return f"运算结果：{num1} {op} {num2} = {res}"

@tool_wrapper(TextParam)
def text_tool(content: str, action: str) -> str:
    if action == "upper": return f"大写结果：{content.upper()}"
    elif action == "lower": return f"小写结果：{content.lower()}"
    elif action == "len": return f"文本长度：{len(content)}"
    return "【参数错误】仅支持upper/lower/len"

tool_list = [StructuredTool.from_function(calc_tool), StructuredTool.from_function(text_tool)]
tool_map = {"calc_tool": calc_tool, "text_tool": text_tool}

# ===================== RAG知识库模块 =====================
private_doc = """
1、LangGraph是企业级AI Agent工作流框架，支持断点续传、分支循环、状态持久化。
2、RAG检索增强可以彻底解决大模型幻觉问题，实现私有文档精准问答。
3、AI工程化核心：架构稳定、输出可控、异常兜底、可监控可迭代。
4、LangGraph商用场景：智能客服、私有知识库、自动化办公、批量数据处理。
"""

text_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30)
split_docs = text_splitter.create_documents([private_doc])
vector_db = Chroma.from_documents(documents=split_docs, embedding=embedding, persist_directory="./rag_db")
vector_db.persist()
retriever = vector_db.as_retriever(search_kwargs={"k": 3})

# ===================== 全局状态 =====================
class FinalAgentState(TypedDict):
    user_query: str
    intent_type: str
    tool_name: str
    tool_params: dict
    tool_result: str
    rag_context: List[str]
    raw_answer: str
    final_answer: str

# ===================== 核心节点 =====================
def intent_node(state: FinalAgentState) -> FinalAgentState:
    prompt = """
    分类为：tool / rag / chat
    tool:数学运算、文本处理
    rag:LangGraph、RAG、AI工程化私有知识
    chat:普通问答
    仅输出分类单词：{query}
    """.format(query=state["user_query"])
    res = llm.invoke(prompt)
    state["intent_type"] = res.content.strip()
    return state

def tool_parse_node(state: FinalAgentState) -> FinalAgentState:
    prompt = """
    输出JSON：{"tool_name":"","tool_params":{}}
    工具：calc_tool、text_tool
    需求：{query}
    """.format(query=state["user_query"])
    res = llm.invoke(prompt)
    content = res.content.strip().replace("```json","").replace("```","")
    try:
        data = json.loads(content)
        state["tool_name"] = data.get("tool_name","")
        state["tool_params"] = data.get("tool_params",{})
    except:
        state["tool_name"] = ""
        state["tool_params"] = {}
    return state

def tool_exec_node(state: FinalAgentState) -> FinalAgentState:
    if state["tool_name"] not in tool_map:
        state["tool_result"] = "暂不支持该工具操作"
    else:
        state["tool_result"] = tool_map[state["tool_name"]](state["tool_params"])
    return state

def rag_retrieve_node(state: FinalAgentState) -> FinalAgentState:
    docs = retriever.get_relevant_documents(state["user_query"])
    state["rag_context"] = [d.page_content.strip() for d in docs]
    return state

def generate_node(state: FinalAgentState) -> FinalAgentState:
    if state["intent_type"] == "tool":
        context = f"工具结果：{state['tool_result']}"
    elif state["intent_type"] == "rag":
        context = f"知识库资料：{'\n'.join(state['rag_context'])}"
    else:
        context = ""
    prompt = f"""
    你是专业AI助手，仅基于参考资料回答，无资料不编造。
    问题：{state['user_query']}
    参考：{context}
    输出简洁、精准、合规的答案。
    """
    res = llm.invoke(prompt)
    state["raw_answer"] = res.content.strip()
    return state

def check_fix_node(state: FinalAgentState) -> FinalAgentState:
    check_prompt = f"""
    修正并标准化回答，杜绝幻觉、答非所问：
    问题：{state['user_query']}
    回答：{state['raw_answer']}
    输出最优标准答案：
    """
    res = llm.invoke(check_prompt)
    state["final_answer"] = res.content.strip()
    return state

# ===================== 工作流路由 =====================
def tool_route(state: FinalAgentState) -> Literal["tool_parse", "generate"]:
    return "tool_parse" if state["intent_type"] == "tool" else "generate"

def rag_route(state: FinalAgentState) -> Literal["rag_retrieve", "generate"]:
    return "rag_retrieve" if state["intent_type"] == "rag" else "generate"

# ===================== 编译工作流 =====================
graph = StateGraph(FinalAgentState)
graph.add_node("intent", intent_node)
graph.add_node("tool_parse", tool_parse_node)
graph.add_node("tool_exec", tool_exec_node)
graph.add_node("rag_retrieve", rag_retrieve_node)
graph.add_node("generate", generate_node)
graph.add_node("check_fix", check_fix_node)

graph.add_edge(START, "intent")
graph.add_conditional_edges("intent", tool_route)
graph.add_edge("tool_parse", "tool_exec")
graph.add_edge("tool_exec", "generate")
graph.add_conditional_edges("intent", rag_route)
graph.add_edge("rag_retrieve", "generate")
graph.add_edge("generate", "check_fix")
graph.add_edge("check_fix", END)

final_agent_workflow = graph.compile(checkpointer=memory)

# ===================== 线上API服务 =====================
app = FastAPI(title="LangGraph商用AI Agent")
# 全局跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class ChatRequest(BaseModel):
    query: str
    thread_id: str = "online_default_001"

@app.post("/api/chat")
def chat(req: ChatRequest):
    res = final_agent_workflow.invoke({
        "user_query": req.query,
        "intent_type": "",
        "tool_name": "",
        "tool_params": {},
        "tool_result": "",
        "rag_context": [],
        "raw_answer": "",
        "final_answer": ""
    }, config={"configurable": {"thread_id": req.thread_id}})
    return {"code":200,"msg":"success","data":res["final_answer"]}

@app.get("/")
def index():
    return {"msg":"LangGraph商用AI Agent服务已在线✅"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

六、第四步：云端服务器完整部署命令

服务器环境：CentOS / Ubuntu 通用，全程一键复制执行。

6.1 安装Docker

# Ubuntu
apt update && apt install docker.io -y

# CentOS
yum install docker -y

# 启动并开机自启
systemctl start docker
systemctl enable docker

6.2 项目上传与镜像构建

将本地完整项目上传至服务器，进入项目目录执行：

# 构建镜像
docker build -t langgraph-agent:v1 .

# 查看镜像
docker images

6.3 启动容器（后台常驻、开机自启）

docker run -d \
--name langgraph-online-agent \
-p 8000:8000 \
--restart always \
-v $(pwd)/rag_db:/app/rag_db \
-v $(pwd)/.env:/app/.env \
langgraph-agent:v1

参数说明：

• -d：后台守护进程运行，关闭终端不中断

• --restart always：服务器重启、程序崩溃自动重启，7×24保活

• 数据挂载：知识库、环境变量持久化，删除容器不丢数据

七、第五步：服务器端口放行（关键）

绝大多数部署失败原因：端口未放行，外部无法访问。

# 放行8000端口
ufw allow 8000

# 查看端口监听
netstat -lnp | grep 8000

同时进入云服务器控制台，在安全组中放行8000端口TCP协议。

八、第六步：线上测试与前后端联调

8.1 服务自测

浏览器访问：服务器IP:8000，显示服务在线即为部署成功。

8.2 接口调用（前端直接对接）

接口地址：http://IP:8000/api/chat

请求方式：POST

请求参数：

{
    "query":"LangGraph商用价值有哪些？",
    "thread_id":"user_1001"
}

前端、小程序、Web页面、客户端均可直接调用，实现完整AI对话功能。

九、线上运维常用命令

# 查看运行容器
docker ps

# 实时查看日志（排错必备）
docker logs -f langgraph-online-agent

# 重启服务
docker restart langgraph-online-agent

# 更新项目：重新构建镜像+重启容器

十、生产级上线优化方案

• 配置域名+HTTPS：备案域名搭配Nginx反向代理，实现公网安全访问

• 日志持久化：挂载日志目录，留存运行记录，方便问题复盘

• 限流防护：增加接口QPS限流，防止恶意请求打爆服务

• 多实例扩容：高并发场景部署多容器实例，负载均衡分流

• 记忆数据库持久化：将长效记忆接入MySQL，支持多实例共享用户记忆

十一、全套22篇教程·终极收官复盘

至此，2026零基础LangGraph AI Agent全套22篇系列教程正式完结。

我们从0基础语法起步，完整走完：

基础工作流 → 分支循环 → 断点持久化 → 监控运维 → 容错重试 → 多智能体协同 → 高阶工具工程化 → Prompt质控 → 批量并发调度 → RAG私有知识库 → 长效记忆人格固化 → 容器化云端上线部署

市面所有LangGraph教程只教「跑通代码」，本套教程独家完成从代码Demo到商业上线的完整闭环，是目前全网最系统、最落地、最贴合企业工程化的AI Agent实战教程。

十二、终章寄语

技术的价值，从来不是“会写代码”，而是能落地、能商用、能解决真实问题。

本套22篇教程，从零带你打破AI开发壁垒，真正掌握工业级AI Agent工程化开发能力，适配面试、毕设、项目变现、企业开发所有场景。

教程完结，学习不止。后续将持续更新多模态Agent、智能体集群、企业级知识库优化、AI自动化办公项目等高阶实战内容，感谢全程陪伴！