打算按这个结构来组织回答:先理解两个框架的本质区别,然后用UML类图展示核心架构,再设计一个完整的实战项目——AI旅行规划助手,最后附上项目文件组织和扩展学习建议。

关于LangGraph vs LangChain v1的关系,搜索结果中没有明确说明,将会根据自己对LangChain生态的了解来解释:LangGraph是LangChain的子项目,LangChain v1简化了Agent创建流程。

下面开始组织这份完整的Agent开发指南。

AI Agent开发完整指南:从框架选型到动手实践

你这个问题问得非常到位——理解概念是一回事,动手实现是另一回事。在这份指南中,我会帮你建立一个完整的AI Agent开发认知框架,涵盖框架对比、架构设计、实战案例和项目组织。

一、框架对比:LangChain vs AutoGPT

在选择开发框架之前,必须先理解两者的本质区别:

对比维度 LangChain AutoGPT
设计理念 模块化构建,开发者控制流程 自主智能体,模型自动执行任务
控制粒度 高,可定义每一步逻辑 低,由模型主导每步行为
使用方式 需要开发者手动构建链条和逻辑 用户只需提供目标,自动执行
适用场景 生产级应用、RAG系统、企业客服 实验探索、自主任务执行、AI研究
稳定性 高,便于测试与调试 弱,出错难以定位
典型用户 企业开发者、技术团队 AI爱好者、研究者

简单类比

  • LangChain 就像用Python写脚本——你可以掌控每一个细节
  • AutoGPT 更像是告诉AI“去完成一个目标”,然后看它如何自己摸索

技术架构差异

LangChain架构:
用户输入 → LLM → Chain(任务链) → Agent(任务决策) → Tools(工具调用) → Memory → 输出

AutoGPT架构:
用户目标 → 任务生成器 → Agent Executor → 工具 → 反馈 → 再规划

选择建议

  • 选择 LangChain:构建定制化AI产品,需要生产级部署
  • 选择 AutoGPT:实验自主性AI能力,研究多步推理
  • 两者可结合:用LangChain编排框架,用AutoGPT的任务生成逻辑辅助拆解

二、核心架构设计

2.1 总体架构图

基础设施

Agent核心

用户层

工具集

Worker Agents

记忆系统

短期记忆
对话上下文

长期记忆
向量数据库

用户输入

Supervisor Agent
调度员

DateAgent
日期助手

WeatherAgent
天气助手

LifeAdvisorAgent
生活顾问

Time工具

Weather工具

Search工具

大模型
GPT-4/DeepSeek

向量数据库

外部API

2.2 核心类图

manages

uses

uses

«abstract»

Agent

+name: str

+role: str

+llm: BaseLLM

+tools: List[Tool]

+memory: BaseMemory

+execute(task: str) : : str

+plan(goal: str) : : List[str]

SupervisorAgent

+workers: List[Agent]

+route(task: str) : : Agent

+coordinate(goal: str) : : str

WorkerAgent

+specialization: str

+execute_subtask(subtask: str) : : str

Tool

+name: str

+description: str

+func: Callable

+execute(*args) : : str

«interface»

Memory

+save(key: str, value: str)

+retrieve(query: str) : : str

+clear()

ShortTermMemory

-buffer: List[Message]

+max_length: int

+add(message: Message)

+get_context() : : str

LongTermMemory

+vectorstore: VectorStore

+embedding_model: EmbeddingModel

+search(query: str, top_k: int) : : List[Document]

2.3 ReAct模式工作流程

ReAct(Reasoning + Acting)是Agent的核心工作模式:

记忆 工具 大模型 Agent User 记忆 工具 大模型 Agent User alt [需要工具] [任务完成] loop [ReAct循环] 输入目标 检索相关记忆 Thought(推理) 决策下一步 Action(执行) Observation(观察) 保存中间结果 Final Answer 最终回答 返回结果

三、实战案例:AI旅行规划助手

下面通过一个完整的项目来串联所有概念。我们将构建一个能够规划旅行行程的AI Agent。

3.1 项目概述

目标:构建一个旅行规划助手,能够:

  • 查询目的地天气
  • 搜索景点信息
  • 规划行程路线
  • 生成旅行报告

技术栈:LangChain v1 + Python + Streamlit

3.2 环境准备

# 创建项目目录
mkdir travel-planner-agent
cd travel-planner-agent

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

# 安装依赖
pip install langchain langchain-google-genai streamlit python-dotenv

3.3 定义State(状态管理)

# state.py
from typing import TypedDict, List, Optional
from datetime import datetime

class TravelState(TypedDict):
    """Agent的状态管理"""
    destination: str                    # 目的地
    start_date: str                     # 开始日期
    end_date: str                       # 结束日期
    weather_info: Optional[str]         # 天气信息
    attractions: List[str]              # 景点列表
    route: List[dict]                   # 行程路线
    final_report: Optional[str]         # 最终报告
    current_step: str                   # 当前步骤
    error_count: int                    # 错误计数

3.4 定义工具(Tools)

# tools.py
import json
import requests
from datetime import datetime
from langchain_core.tools import tool

@tool
def get_current_weather(city: str) -> str:
    """获取指定城市的当前天气
    
    Args:
        city: 城市名称,如"北京"
    
    Returns:
        天气信息的JSON字符串
    """
    # 模拟API调用(实际项目中替换为真实API)
    weather_data = {
        "city": city,
        "temperature": "22°C",
        "condition": "晴朗",
        "humidity": "65%",
        "wind_speed": "12km/h",
        "suggestion": "适合户外活动"
    }
    return json.dumps(weather_data, ensure_ascii=False)

@tool
def search_attractions(city: str, category: str = "all") -> str:
    """搜索城市的景点信息
    
    Args:
        city: 城市名称
        category: 景点类型(all/historical/natural/entertainment)
    
    Returns:
        景点列表的JSON字符串
    """
    # 模拟景点数据
    attractions_db = {
        "北京": [
            {"name": "故宫", "type": "historical", "duration": "3-4小时", "price": "60元"},
            {"name": "长城", "type": "historical", "duration": "4-5小时", "price": "40元"},
            {"name": "颐和园", "type": "natural", "duration": "2-3小时", "price": "30元"}
        ],
        "上海": [
            {"name": "外滩", "type": "entertainment", "duration": "1-2小时", "price": "免费"},
            {"name": "迪士尼乐园", "type": "entertainment", "duration": "全天", "price": "599元"},
            {"name": "东方明珠", "type": "entertainment", "duration": "2小时", "price": "199元"}
        ]
    }
    
    attractions = attractions_db.get(city, [])
    return json.dumps(attractions, ensure_ascii=False)

@tool
def calculate_duration(start_date: str, end_date: str) -> str:
    """计算旅行天数
    
    Args:
        start_date: 开始日期,格式YYYY-MM-DD
        end_date: 结束日期,格式YYYY-MM-DD
    
    Returns:
        旅行天数
    """
    start = datetime.strptime(start_date, "%Y-%m-%d")
    end = datetime.strptime(end_date, "%Y-%m-%d")
    days = (end - start).days + 1
    return f"{days}天"

@tool
def format_travel_report(destination: str, days: int, route: list, weather: str) -> str:
    """格式化旅行报告
    
    Args:
        destination: 目的地
        days: 旅行天数
        route: 行程路线
        weather: 天气信息
    
    Returns:
        格式化的旅行报告
    """
    report = f"""
╔══════════════════════════════════════════════════════════╗
║                    {destination} 旅行规划报告                      ║
╚══════════════════════════════════════════════════════════╝

📅 行程天数:{days}天
🌤️  天气概况:{weather}

🗺️  每日行程:
"""
    for i, day_plan in enumerate(route, 1):
        report += f"\n第{i}天:\n"
        for item in day_plan:
            report += f"  • {item}\n"
    
    report += """
💡 温馨提示:
  1. 建议提前预订热门景点门票
  2. 根据天气情况准备合适的衣物
  3. 预留充足的休息时间
  
祝您旅途愉快!🎉
"""
    return report

3.5 构建Multi-Agent系统

# agents.py
from langchain.agents import create_agent
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents.middleware import AgentMiddleware
from pydantic import BaseModel, Field
from typing import List
import os
from dotenv import load_dotenv
from tools import get_current_weather, search_attractions, calculate_duration, format_travel_report

load_dotenv()

# 初始化LLM
model = ChatGoogleGenerativeAI(
    model="gemini-2.0-flash-exp",
    google_api_key=os.getenv("GOOGLE_API_KEY"),
    temperature=0.7
)

# ============ 1. 定义结构化输出模型 ============
class TravelPlan(BaseModel):
    """旅行计划的结构化输出"""
    destination: str = Field(description="目的地城市")
    start_date: str = Field(description="出发日期 YYYY-MM-DD")
    end_date: str = Field(description="返回日期 YYYY-MM-DD")
    attractions: List[str] = Field(description="推荐的景点列表")
    daily_route: List[str] = Field(description="每日行程安排")

# ============ 2. 定义中间件(质量验证) ============
class TravelPlanValidator(AgentMiddleware):
    """验证旅行计划的质量"""
    
    def after_model(self, state, runtime):
        if "structured_response" in state:
            plan = state["structured_response"]
            
            # 检查是否有景点推荐
            if len(plan.attractions) < 1:
                return {
                    "messages": [{
                        "role": "user",
                        "content": "请至少推荐1个景点。"
                    }]
                }
            
            # 检查日期格式
            if not plan.start_date or not plan.end_date:
                return {
                    "messages": [{
                        "role": "user",
                        "content": "请提供具体的开始和结束日期。"
                    }]
                }
        
        return None

# ============ 3. 创建Worker Agent ============
def create_weather_agent():
    """创建天气查询Agent"""
    system_prompt = """你是一个天气查询助手。
    使用 get_current_weather 工具查询指定城市的天气。
    返回格式:温度、天气状况、湿度、出行建议。"""
    
    return create_agent(
        model=model,
        tools=[get_current_weather],
        system_prompt=system_prompt,
        response_format=WeatherReport  # 自定义输出格式
    )

def create_attraction_agent():
    """创建景点查询Agent"""
    system_prompt = """你是一个旅游景点推荐助手。
    使用 search_attractions 工具查询景点信息。
    根据用户偏好推荐合适的景点。"""
    
    return create_agent(
        model=model,
        tools=[search_attractions],
        system_prompt=system_prompt
    )

# ============ 4. 创建Supervisor Agent ============
def create_travel_supervisor():
    """创建旅行规划总控Agent"""
    system_prompt = """你是旅行规划总控助手。
    你的职责是:
    1. 理解用户的旅行需求
    2. 调用天气助手获取目的地天气
    3. 调用景点助手获取推荐景点
    4. 规划每日行程
    5. 生成完整的旅行报告
    
    请确保回答完整、信息准确、建议实用。"""
    
    return create_agent(
        model=model,
        tools=[get_current_weather, search_attractions, calculate_duration, format_travel_report],
        system_prompt=system_prompt,
        middleware=[TravelPlanValidator()],
        response_format=TravelPlan
    )

3.6 主程序:Streamlit界面

# app.py
import streamlit as st
from agents import create_travel_supervisor
from dotenv import load_dotenv
import os

load_dotenv()

# 页面配置
st.set_page_config(
    page_title="AI旅行规划助手",
    page_icon="✈️",
    layout="wide"
)

st.title("✈️ AI旅行规划助手")
st.markdown("输入你的旅行需求,AI将为你规划完整的行程")

# 初始化session state
if "messages" not in st.session_state:
    st.session_state.messages = []

if "agent" not in st.session_state:
    with st.spinner("正在初始化AI助手..."):
        st.session_state.agent = create_travel_supervisor()

# 显示历史消息
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# 用户输入
if prompt := st.chat_input("比如:我想去北京玩3天,帮我规划一下行程"):
    # 添加用户消息
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)
    
    # 获取AI响应
    with st.chat_message("assistant"):
        with st.spinner("正在规划行程..."):
            try:
                response = st.session_state.agent.invoke({
                    "messages": [{"role": "user", "content": prompt}]
                })
                
                # 处理结构化响应
                if "structured_response" in response:
                    plan = response["structured_response"]
                    
                    # 显示规划结果
                    col1, col2 = st.columns(2)
                    
                    with col1:
                        st.subheader("📍 目的地信息")
                        st.write(f"**城市:** {plan.destination}")
                        st.write(f"**日期:** {plan.start_date}{plan.end_date}")
                    
                    with col2:
                        st.subheader("🏛️ 推荐景点")
                        for attr in plan.attractions:
                            st.write(f"• {attr}")
                    
                    st.subheader("🗓️ 每日行程")
                    for day_plan in plan.daily_route:
                        st.write(f"• {day_plan}")
                    
                    final_output = f"""
                    **行程规划完成!**

                    目的地:{plan.destination}
                    日期:{plan.start_date}{plan.end_date}
                    推荐景点:{', '.join(plan.attractions)}
                    """
                else:
                    final_output = response.get("output", "无法生成规划,请重试")
                
                st.markdown(final_output)
                
            except Exception as e:
                st.error(f"出错了:{str(e)}")
                final_output = f"抱歉,规划过程中出现错误:{str(e)}"
        
        # 保存消息
        st.session_state.messages.append({"role": "assistant", "content": final_output})

四、项目文件结构

travel-planner-agent/
├── README.md                 # 项目说明
├── requirements.txt          # 依赖列表
├── .env                      # 环境变量(API密钥)
├── .gitignore               # Git忽略文件
│
├── app.py                   # Streamlit主程序
├── agents.py                # Agent定义
├── tools.py                 # 工具函数
├── state.py                 # 状态定义
│
├── config/
│   ├── __init__.py
│   └── settings.py          # 配置参数
│
├── memory/
│   ├── __init__.py
│   ├── short_term.py        # 短期记忆
│   └── long_term.py         # 长期记忆(向量数据库)
│
├── utils/
│   ├── __init__.py
│   ├── logger.py            # 日志工具
│   └── validators.py        # 数据验证
│
├── tests/
│   ├── test_agents.py
│   ├── test_tools.py
│   └── test_memory.py
│
├── data/
│   ├── attractions.json     # 景点数据
│   └── prompts/             # 提示词模板
│
└── docs/
    ├── architecture.md      # 架构文档
    └── api.md              # API文档

五、扩展与进阶

5.1 从LangChain v1迁移到LangGraph

对于更复杂的Agent系统,可以使用LangGraph:

from langgraph.graph import StateGraph, END

# 定义状态
class AgentState(TypedDict):
    messages: List[BaseMessage]
    next_step: str
    tool_results: dict

# 构建图
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("classify", classification_node)
workflow.add_node("search", search_node)
workflow.add_node("summarize", summarize_node)

# 添加边
workflow.add_edge("classify", "search")
workflow.add_conditional_edges("search", should_summarize)
workflow.add_edge("summarize", END)

# 编译
app = workflow.compile()

5.2 添加记忆系统

from langchain.memory import ConversationBufferWindowMemory
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

# 短期记忆
short_term_memory = ConversationBufferWindowMemory(
    k=10,  # 保留最近10轮对话
    return_messages=True
)

# 长期记忆(向量数据库)
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    texts=historical_conversations,
    embedding=embeddings
)

5.3 框架选择建议

场景 推荐框架 理由
生产级RAG系统 LangChain v1 稳定性高、生态完善
复杂多步任务 LangGraph 图结构灵活、支持循环
自主探索实验 AutoGPT 目标驱动、自动化程度高
快速原型验证 轻量级Agent 自定义程度高、易于调试

六、总结

通过这份指南,你应该能够:

  1. 理解框架差异:LangChain是开发者框架(高控制),AutoGPT是自主智能体(高自动)

  2. 掌握核心架构:Supervisor-Worker模式、ReAct循环、记忆系统

  3. 动手构建项目:完整的旅行规划助手代码实现

  4. 组织项目结构:清晰的文件组织和模块划分

学习路径建议

  1. 先跑通本指南的完整示例
  2. 修改工具函数,接入真实API
  3. 尝试添加新的Worker Agent
  4. 学习LangGraph构建更复杂的流程
Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐