---

作者： 1Lyn-en
项目地址： github.com/1Lyn-en/AgentForge
版权声明： 本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

---

🤖 从零搭建企业级 AI 智能体平台：AgentForge 完整架构解析与实战

一、写在前面

2026 年，大模型早已不是「聊天玩具」。企业真正需要的是能理解业务、调用工具、自主决策的 AI 智能体（Agent）。但在实际落地中，你可能会遇到这些问题：

• 🤯 想做一个带工具调用的 Agent，但 LangChain 文档散乱、版本割裂
• 🔧 需要集成多个 LLM（Qwen、DeepSeek），切换起来手忙脚乱
• 📧 想让 AI 自动发邮件、查数据库，但不知道怎么串联
• 🎬 想接入视频生成能力，还要管理资产库
• 📊 老板要每日数据简报，你还在一张张跑 SQL 贴 Excel

这些问题，AgentForge 都给出了答案。本文将从零开始，带你看懂一个生产级全栈 AI 智能体平台的完整架构与实现细节。

---

二、项目概览

AgentForge 是一个基于 LangChain + FastAPI + Vue.js 的企业级多智能体平台，v0.3.0 版本实现了五大核心模块：

模块	作用	关键技术
🧠 AI Agent 层	4 个智能体，各司其职	LangChain create_agent + AgentExecutor
🔌 Tool 工具层	8 个工具，打通内外系统	LangChain @tool 装饰器
🤖 LLM 模型层	统一工厂，多模型热切换	ChatOpenAI (DashScope / DeepSeek)
🌐 REST API 层	20+ 接口，前后端分离	FastAPI + SSE 流式推送
🎨 前端 SPA 层	单页应用，开箱即用	Vue.js 2 + marked.js

支持的功能包括：

• ✅ 多模型对话：Qwen 3.7 Max / Plus、DeepSeek V4 Pro 无缝切换
• ✅ 流式 SSE 响应：打字机效果实时输出
• ✅ 邮件验证码登录：QQ 邮箱 SMTP + Redis 会话管理
• ✅ AI 视频生成：DashScope Wan2.7-T2V，自动下载归档
• ✅ 视频资产管理：搜索、播放、分页
• ✅ CEO 数据简报：从 MySQL 自动提取业务数据，邮件推送
• ✅ 对话历史管理：Redis 存储，30 天保留
• ✅ 暗色/亮色主题：CSS 自定义属性方案

---

三、系统架构深度解析

3.1 整体架构图

┌──────────────────────────────────────────────────────────────┐
│                     Vue.js SPA (chat.html)                     │
│    登录页 · 聊天页 · 视频库 · 主题切换 · Markdown 渲染         │
└──────────────────────────┬───────────────────────────────────┘
                           │ HTTP / SSE
┌──────────────────────────▼───────────────────────────────────┐
│                     FastAPI (port 9000)                        │
│  CORS · 静态文件挂载 · 20+ REST 端点 · 路由分发                │
└──────────────────────────┬───────────────────────────────────┘
                           │
┌──────────────────────────▼───────────────────────────────────┐
│                   AI Agent 层 (LangChain)                      │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────────────┐   │
│  │ ChatAgent│ │LoginAgent│ │VideoAgent│ │PersonalAssist   │   │
│  └────┬─────┘ └────┬─────┘ └────┬─────┘ └───────┬────────┘   │
│       │            │            │                │            │
│       ▼            ▼            ▼                ▼            │
│  ┌────────────────────────────────────────────────────────┐   │
│  │              工具层 (8 个 LangChain Tools)               │   │
│  │  MySQL · Email · Redis · VideoGen · Download · Archive  │   │
│  └───────────┬────────────┬───────────────┬───────────────┘   │
└──────────────┼────────────┼───────────────┼──────────────────┘
               ▼            ▼               ▼
          ┌────────┐  ┌──────────┐  ┌──────────────┐
          │ MySQL  │  │ QQ SMTP  │  │ Redis        │
          │   DB   │  │ Server   │  │ Cache/Session│
          └────────┘  └──────────┘  └──────────────┘

               ┌──────────────────────────────────┐
               │      LLM 后端（三选一）             │
               │  Qwen Max · Qwen Plus · DeepSeek  │
               │  Wan2.7 T2V (视频生成)             │
               └──────────────────────────────────┘

3.2 核心设计原则

AgentForge 的设计遵循 3 个关键原则：

1. Agent 即服务：每个 Agent 封装一个完整业务场景，输入问题、输出结果，对外屏蔽 LangChain 复杂度
2. 工具即插即用：每个工具独立、无状态、可复用，Agent 只需声明依赖哪些工具
3. 配置即代码：所有敏感信息走环境变量，模型切换通过字符串 key，无需改代码

---

四、核心模块代码拆解

4.1 ChatAgent：通用对话智能体

最核心的 Agent，掌管所有通用对话。它拥有 4 个工具：发邮件、查数据库、生成视频、下载视频。

# ai/agent/chat_agent.py
from ai.tool.mysql_tool import mysql_tool
from ai.tool.send_email_tool import send_email_tool
from ai.tool.generate_video_tool import generate_video
from ai.tool.download_video_tool import download_video_tool
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

class ChatAgent:
    def __init__(self, model_key: str = None):
        config = MODEL_CONFIG.get(model_key, MODEL_CONFIG["qwen3.7-max"])
        self.model = ChatOpenAI(
            model=config["model_name"],
            api_key=config["api_key"],
            base_url=config["base_url"]
        )
        self.tools = [send_email_tool, mysql_tool, generate_video, download_video_tool]
        self.agent = create_agent(
            model=self.model,
            tools=self.tools,
            system_prompt="你是一个智能助手...",
            debug=True
        )

多模型配置

支持 3 个模型热切换，通过 model_key 参数选择：

MODEL_CONFIG = {
    "qwen3.7-max": {
        "model_name": os.getenv("QWEN_MODEL_NAME"),
        "base_url": os.getenv("QWEN_MODEL_URL"),
        "api_key": os.getenv("DASHSCOPE_API_KEY"),
    },
    "qwen3.7-plus": { ... },
    "deepseek-v4pro": {
        "model_name": os.getenv("DEEPSEEK_MODEL_NAME"),
        "base_url": os.getenv("DEEPSEEK_MODEL_URL"),
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
    },
}

流式 SSE 响应

使用 astream_events + generator 实现打字机效果：

async def chat_stream(self, question: str):
    async for event in self.agent.astream_events(
        {"messages": [{"role": "user", "content": question}]},
        version="v2"
    ):
        event_type = event["event"]
        if event_type == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if hasattr(chunk, "content") and chunk.content:
                yield {"type": "answer", "content": chunk.content}
        elif event_type == "on_tool_start":
            yield {"type": "thinking", "content": f"正在调用工具：{event['name']}"}

前端通过 EventSource 接收，体验丝滑流畅。

4.2 LoginAgent：邮件验证码登录

这是一个纯 Prompt 驱动的 Agent——没有手写逻辑，完全靠系统提示词让 LLM 理解流程并自主调用工具：

# ai/agent/login_agent.py
LOGIN_SYSTEM_PROMPT = """
你是一个邮件登录助手

你可以发送邮件，你有以下工具：
一 send_email_tool 发送邮件
二 mysql_tool 数据库查询工具
三 redis_tool redis 设置数据
四 redis_get_tool redis 获取数据

请严格按照以下步骤执行：
一：意图识别
  如果用户输入只有邮箱，请执行发送验证码流程
  如果用户输入含有邮箱和验证码，请执行登录流程

二：发送验证码流程
  步骤一 随机生成一个4位数的不规则随机数
  步骤二 调用 send_email_tool 发送邮件
  步骤三 调用 redis_tool 存入验证码，键名格式 code:邮箱

三：登录流程
  步骤一 调用 redis_get_tool 获取验证码并比对
  步骤二 调用 mysql_tool 根据邮箱查询用户名
"""

登录成功后，API 层还会生成会话 Token 存入 Redis（7 天 TTL）：

token = str(uuid.uuid4())
redis_client.set(
    f"session:{token}", 
    f"{email}|{username}|{user_id}", 
    ex=SESSION_TTL  # 604800秒
)

4.3 VideoAgent：文生视频全流程

从文字描述到最终视频文件落地，一条龙自动化：

# ai/tool/generate_video_tool.py
import dashscope

@tool(args_schema=VideoGenInput)
def generate_video(prompt: str) -> str:
    """根据文本描述生成视频"""
    rsp = dashscope.VideoSynthesis.call(
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        model='wan2.7-t2v-2026-04-25',  # 通义万相 2.7
        prompt=prompt,
        resolution="720P",
        ratio="16:9",
        duration=2,
        watermark=True
    )
    if rsp.status_code == HTTPStatus.OK:
        return f"视频生成成功！视频链接：{rsp.output.video_url}"
    return f"视频生成失败: {rsp.message}"

生成后自动下载并归档到 MySQL：

# ai/tool/download_video_tool.py
@tool("download_video_tool")
def download_video_tool(url: str) -> Dict[str, Any]:
    """下载视频到本地 static/download 目录，并自动归档到数据库"""
    filename = time.strftime("%Y%m%d%H%M%S") + ".mp4"
    save_path = os.path.join("static", "download", filename)
    # ... 流式下载逻辑 ...
    local_url = f"http://localhost:9000/static/download/{filename}"
    archive_result = _archive_video(filename, local_url, total_size)
    return {
        "type": "video",
        "video_url": local_url,
        "status": "success",
        "archive_info": archive_result
    }

4.4 PersonalAssistant：CEO 每日数据简报

自动从 MySQL 的 orders/products 表提取 5 个维度的业务数据并邮件推送：

# 总订单数与总销售额
cursor.execute("SELECT COUNT(*), SUM(total_amount) FROM orders")
total_orders, total_amount = cursor.fetchone()

# 最近 7 天订单趋势
cursor.execute("""SELECT order_date, COUNT(*), SUM(total_amount)
    FROM orders GROUP BY order_date ORDER BY order_date DESC LIMIT 7""")

# 热销商品 TOP 5 (JOIN products 表)
cursor.execute("""SELECT p.product_name, p.category, 
    SUM(o.quantity), SUM(o.total_amount)
    FROM orders o JOIN products p ON o.product_id = p.product_id
    GROUP BY o.product_id ORDER BY SUM(o.total_amount) DESC LIMIT 5""")

# 品类销售额分布 + 支付方式分布...

生成的简报格式：

==================================================
          CEO 每日数据简报
==================================================
📊 总订单数：1,286
💰 总销售额：¥589,234.00
📈 平均订单金额：¥458.20

── 最近 7 日订单趋势 ──
  2026-06-28: 187 单，¥85,432.00
  ...

── 热销商品 TOP 5 ──
  1. 笔记本电脑（数码）：销量 89，销售额 ¥534,000.00
  ...

4.5 工具即插即用：LangChain @tool 模式

每个工具都用 @tool 装饰器定义，配合 Pydantic 做参数校验：

# ai/tool/mysql_tool.py
from pydantic import BaseModel, Field
from langchain.tools import tool

class MysqlToolParams(BaseModel):
    sql: str = Field(..., description="sql语句")

@tool("mysql_tool", args_schema=MysqlToolParams)
def mysql_tool(sql: str) -> str:
    """查询sql语句"""
    con = pymysql.connect(
        host=os.getenv("DATABASE_HOST"),
        port=int(os.getenv("DATABASE_PORT")),
        user=os.getenv("DATABASE_USER"),
        password=os.getenv("DATABASE_PASSWORD"),
        database=os.getenv("DATABASE_NAME"),
    )
    cursor = con.cursor()
    cursor.execute(sql)
    rs = cursor.fetchall()
    return str(rs)

Agent 只需声明 tools = [mysql_tool, send_email_tool, ...] 即可获得能力。新增工具只需新建一个 @tool 函数，无缝接入。

---

五、API 接口体系

系统提供 14+ REST 接口，覆盖完整业务闭环：

端点	方法	功能	鉴权
/chat_invoke	GET	同步对话	✅
/chat_stream	GET	流式对话 (SSE)	✅
/sendCode	GET	发送验证码	❌
/login	GET	验证码登录	❌
/logout	GET	退出登录	✅
/check_login	GET	检查登录状态	✅
/save_conversation	POST	保存会话	✅
/get_conversations	GET	会话列表	✅
/get_conversation	GET	获取特定会话	✅
/delete_conversation	POST	删除会话	✅
/chat_video	GET	视频生成 (SSE)	✅
/get_video_list	GET	视频资产列表	✅
/archive_video	POST	视频归档	✅
/personal_assistant	GET	CEO 简报	✅

流式接口使用 StreamingResponse + text/event-stream，前端通过 EventSource 接收：

// 前端 SSE 接收示例
const source = new EventSource(`/chat_stream?question=${encodeURIComponent(msg)}&token=${token}`);
source.onmessage = (event) => {
    const data = JSON.parse(event.data);
    if (data.done) { source.close(); return; }
    // data.type: "answer" | "thinking" | "video"
    // data.data: 文本内容
};

---

六、环境搭建与配置

6.1 前置依赖

组件	版本要求	用途
Python	>= 3.10	运行环境
MySQL	8.0+	业务数据持久化
Redis	6.0+	会话、缓存、验证码
DashScope API Key	-	Qwen 模型 + 视频生成
DeepSeek API Key	-	DeepSeek 模型（可选）

6.2 配置指南

复制 .env.example 为 .env，填入真实配置：

# LLM 模型
DASHSCOPE_API_KEY=sk-你的Key
DEEPSEEK_API_KEY=sk-你的Key（可选）

# 数据库
DATABASE_HOST=localhost
DATABASE_PORT=3306
DATABASE_USER=root
DATABASE_PASSWORD=你的密码
DATABASE_NAME=shiyou03

# Redis
REDIS_HOST=127.0.0.1
REDIS_PORT=6379

# SMTP 邮件
SMTP_USER=你的QQ邮箱@qq.com
SMTP_PASSWORD=你的QQ邮箱SMTP授权码

6.3 数据库表结构

-- 用户表
CREATE TABLE user_info (
    user_id   INT PRIMARY KEY,
    user_name VARCHAR(100),
    email     VARCHAR(200),
    department VARCHAR(100)
);

-- 订单表
CREATE TABLE orders (
    id             INT PRIMARY KEY AUTO_INCREMENT,
    order_date     DATE,
    product_id     INT,
    quantity       INT,
    total_amount   DECIMAL(10,2),
    payment_method VARCHAR(50)
);

-- 商品表
CREATE TABLE products (
    product_id   INT PRIMARY KEY,
    product_name VARCHAR(200),
    category     VARCHAR(100)
);

-- 视频归档表
CREATE TABLE video_archive (
    id            INT PRIMARY KEY AUTO_INCREMENT,
    video_name    VARCHAR(255),
    download_time DATETIME,
    video_url     VARCHAR(500),
    file_size     INT
);

6.4 启动运行

# 1. 克隆项目
git clone https://github.com/1Lyn-en/AgentForge.git
cd AgentForge

# 2. 安装依赖
pip install -r requirements.txt

# 3. 配置环境变量
cp .env.example .env
# 编辑 .env 填入你的 API Key 和数据库密码

# 4. 启动服务
uvicorn api.main:app --host 0.0.0.0 --port 9000 --reload

# 5. 打开浏览器访问
# http://localhost:9000

6.5 启动验证

1. 登录流程：在登录页输入邮箱 → 点击发送验证码 → 查收邮件 → 输入 4 位验证码登录
2. 对话测试：选择 Qwen 3.7 Max 模型，输入「查询一下 user_info 表的数据」
3. 视频生成：输入「一只橘猫在阳光下打哈欠，特写镜头」
4. CEO 简报：点击个人助理功能，输入收件邮箱接收业务数据报告

---

七、技术亮点总结

7.1 Prompt 即逻辑

LoginAgent 是一个极佳的例子——零代码的业务流程。传统的登录系统需要手写 if-else 判断、状态机管理，而在 Agent 模式下，只需一段自然语言 Prompt 就能让 LLM 自主理解并执行：“发验证码→存 Redis→验证→查用户”。代码量减少了 80%，且修改流程只需改 Prompt，无需动代码。

7.2 流式交互体验

同步 API 在 AI 场景下体验极差（用户要等 5-10 秒才有响应）。AgentForge 全链路采用 SSE (Server-Sent Events) 推送，工具调用状态实时可见（“正在查询数据库…” → “正在生成视频…” → 逐 token 输出回答），用户感知延迟从 10 秒降到 0。

7.3 Agent + Tool 的解耦设计

每个 Tool 独立、无状态、只做一件事。新增能力 = 新建一个 @tool 函数 + 在 Agent 的 tools 列表中注册。这种插件式架构让系统极易扩展：想接飞书通知？写一个 send_feishu_tool 加上就行。

7.4 全栈开箱即用

后端 FastAPI + 前端 Vue.js SPA（单文件，无构建步骤），部署极其轻量。前端依赖通过 CDN 引入，连 npm install 都不需要。这让项目非常适合企业内部工具或个人开发者快速验证想法的场景。

---

八、踩坑与优化建议

8.1 踩过的坑

坑	解决
LangChain 0.x → 1.0 接口不兼容	使用 create_agent 替代 initialize_agent
SSE 流式乱码	确保 FastAPI 返回 UTF-8 + ensure_ascii=False
Redis 中文验证码比对失败	decode_responses=True 统一编码
视频下载大文件内存溢出	使用 stream=True + iter_content 分块写入

8.2 优化方向

1. WebSocket 替代 SSE：SSE 是单向推送，WebSocket 支持双向通信，适合更复杂的协作场景
2. Docker 容器化：当前依赖手动安装 MySQL/Redis，Docker Compose 一键启动体验更好
3. 增加测试覆盖：项目目前缺乏自动化测试，关键 Agent 流的单元测试迫在眉睫
4. 敏感信息管理：SMTP 密码已从代码迁移到 .env，建议后续接入密钥管理服务

---

九、总结

AgentForge 展示了 LangChain Agent 模式在企业场景下的完整落地路径。它不是一个"Hello World"级别的 Demo，而是一个包含认证、对话、视频、数据分析、邮件推送的生产级全栈应用。

它所体现的核心理念——用自然语言替代传统代码编写业务流程逻辑——正在重新定义企业应用开发的方式。当 LLM 不再是聊天玩具，而是成为能理解业务、调用工具、自主决策的"数字员工"，这正是 AI Agent 的真正价值所在。

如果你正在探索如何将 AI Agent 落地到实际业务中，希望 AgentForge 的架构思路能给你一些启发。

---

项目地址： github.com/1Lyn-en/AgentForge
欢迎 Star ⭐、Fork、PR！

觉得有用的话，点个赞再走呗 👍