失业一年了，天天想着怎么翻身。终于走到最后一课了。从第1课的“LangGraph是什么”，到第7课的生产级部署，我们一起把LangGraph从零基础走到了能真正落地的程度。这节课我们把前7课所有知识点全部整合起来，做一个企业级多Agent智能写作与代码生成系统，作为整个课程的完结大项目。

本课目标

• 整合State、Node、Edge、多Agent协作、记忆、Human-in-the-loop、持久化、部署等全部知识
• 完成一个可实际使用的综合大项目
• 展望LangGraph 2026年及未来的发展方向
• 为你后续自主开发打下完整基础

项目名称：AutoTeam —— 企业级多Agent智能助手（写作 + 代码生成双模式）

项目架构

• Supervisor（主管Agent）：负责任务分解和调度
• Researcher（研究员）：负责搜索最新信息
• Writer（写手）：负责生成文章
• Coder（编程助手）：负责生成代码
• Reviewer（审稿人/代码审查）：负责最终质量把关
• Human Review Node：关键节点人工介入
• PostgreSQL持久化 + FastAPI部署

环境准备

pip install -U langgraph langchain langchain-deepseek langchain-core fastapi uvicorn langserve langsmith tavily-python psycopg
pip install langgraph-checkpoint-postgres

核心代码实战（完结大项目）

代码段1：定义统一State

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    task_type: Literal["writing", "coding"]          # 任务类型
    project_requirements: str                        # 原始需求
    current_draft: str                               # 当前文章/代码草稿
    research_notes: str                              # 研究员搜集的信息
    review_feedback: str                             # 审稿人反馈
    review_status: str                               # pending / approved / rejected
    iteration_count: int                             # 当前迭代次数

代码段2：定义各个专业Agent节点

from langchain_deepseek import ChatDeepSeek
from langchain_core.tools import tool
from tavily import TavilyClient

llm = ChatDeepSeek(model="deepseek-chat", temperature=0.6)
tavily = TavilyClient()

# Researcher
def researcher_node(state: State):
    query = state["messages"][-1].content
    results = tavily.search(query, max_results=5)
    notes = "\n\n".join([r['content'] for r in results['results']])
    return {"research_notes": notes, "messages": [AIMessage(content=f"已完成调研，收集到{len(results['results'])}条信息")]}

# Writer / Coder（根据任务类型切换）
def content_agent_node(state: State):
    if state["task_type"] == "writing":
        prompt = f"根据以下调研信息，撰写专业文章：\n{state.get('research_notes', '')}"
    else:
        prompt = f"根据需求生成高质量代码：\n{state['messages'][-1].content}"
    
    response = llm.invoke(prompt)
    return {"current_draft": response.content, "messages": [response]}

# Reviewer
def reviewer_node(state: State):
    draft = state.get("current_draft", "")
    prompt = f"审查以下内容并给出详细反馈：\n{draft}"
    feedback = llm.invoke(prompt).content
    return {"review_feedback": feedback, "messages": [AIMessage(content=feedback)]}

# Supervisor（主管）
def supervisor_node(state: State):
    prompt = f"""
    当前任务类型：{state['task_type']}
    迭代次数：{state.get('iteration_count', 0)}
    请决定下一步执行：researcher / content_agent / reviewer / FINISH
    """
    decision = llm.invoke(prompt).content.lower()
    
    if "researcher" in decision:
        return {"messages": [AIMessage(content="Next: researcher")]}
    elif "reviewer" in decision:
        return {"messages": [AIMessage(content="Next: reviewer")]}
    else:
        return {"messages": [AIMessage(content="Next: FINISH")]}

代码段3：完整多Agent图 + 生产部署（完结版）

from fastapi import FastAPI
from langserve import add_routes
from langgraph.checkpoint.postgres import PostgresSaver
import psycopg

# PostgreSQL持久化
conn_string = "postgresql://postgres:password@localhost:5432/langgraph_db"
with psycopg.connect(conn_string) as conn:
    checkpointer = PostgresSaver(conn)
    checkpointer.setup()

workflow = StateGraph(State)

workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("content_agent", content_agent_node)
workflow.add_node("reviewer", reviewer_node)

workflow.add_edge(START, "supervisor")

workflow.add_conditional_edges(
    "supervisor",
    lambda state: "researcher" if "researcher" in state["messages"][-1].content else
                  "content_agent" if "content" in state["messages"][-1].content else
                  "reviewer" if "reviewer" in state["messages"][-1].content else "FINISH",
    {"researcher": "researcher", "content_agent": "content_agent", 
     "reviewer": "reviewer", "FINISH": END}
)

# 各节点完成后回到主管
for node in ["researcher", "content_agent", "reviewer"]:
    workflow.add_edge(node, "supervisor")

# 添加人工审核中断（关键节点）
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["reviewer"]   # 在最终审查前人工介入
)

# FastAPI部署
app = FastAPI(title="AutoTeam - 企业级多Agent智能系统", version="1.0.0")
add_routes(app, graph, path="/autoteam")

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

项目使用方式

# 启动服务
uvicorn main:app --reload

# 调用示例（支持持久化多轮）
config = {"configurable": {"thread_id": "autoteam_2026_demo"}}

graph.invoke({
    "messages": [HumanMessage(content="写一篇2026年LangGraph在金融行业的应用分析报告")],
    "task_type": "writing",
    "iteration_count": 0
}, config=config)

---

LangServe自动生成的接口说明

当你启动服务后，可用的主要接口如下：

方法	接口地址	用途说明
POST	/autoteam/invoke	最常用：同步调用，返回完整结果
POST	/autoteam/stream	流式输出（推荐用于长输出）
POST	/autoteam/batch	批量调用多个输入
GET	/autoteam/input_schema	查看输入参数格式
GET	/autoteam/output_schema	查看输出格式
GET	/autoteam/config_schema	查看可配置参数（如 thread_id）

真实HTTP调用示例（推荐方式）

使用 curl 或 Postman 调用：

# 1. 同步调用（最常用）
curl -X POST http://localhost:8000/autoteam/invoke \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "messages": [{"role": "human", "content": "写一篇2026年LangGraph在金融行业的应用报告"}],
      "task_type": "writing"
    },
    "config": {
      "configurable": {
        "thread_id": "autoteam_demo_2026"
      }
    }
  }'

# 2. 流式调用（推荐用于生成长文章或代码）
curl -X POST http://localhost:8000/autoteam/stream \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "messages": [{"role": "human", "content": "帮我生成一个FastAPI用户管理系统代码"}],
      "task_type": "coding"
    },
    "config": {
      "configurable": {"thread_id": "autoteam_demo_2026"}
    }
  }'

小练习（正好2道）

• 练习1（基础）
  使用Postman或curl调用 /autoteam/invoke 接口，完成一次写作任务，并记录返回结果。

• 练习2（进阶）
  调用 /autoteam/stream 接口，实现流式输出，并在前端（或命令行）实时显示生成的内容。

本课小结
本课程最后一课，我们将前7课知识全部融合，完成了 AutoTeam 这个企业级多Agent系统。通过 FastAPI + LangServe，我们把复杂的LangGraph工作流变成了标准HTTP接口（最常用的是 /autoteam/invoke 和 /autoteam/stream）。至此，你已经具备了从零到生产级部署LangGraph的完整能力。

如果觉得这篇有用，欢迎点赞和关注，一起玩转 LangGraph！

---

课程完结感言
恭喜你！坚持学完8节课的你，已经远远超过大多数开发者对LangGraph的掌握程度。
未来不管是做自动化工作流、智能助手，还是企业级多Agent系统，你都有了坚实的基础。

感谢一路陪伴，祝你在2026年用LangGraph做出真正有价值的产品！

如果后续还有任何问题，随时来找我继续讨论。
我们，下一个项目见！