AI Agent 部署完整教程｜附源码

随着大语言模型能力的快速提升，AI Agent（智能体）已经从“简单聊天机器人”逐渐发展为能够调用工具、执行任务、读取资料、访问数据库、自动化工作流的应用形态。相比普通的 ChatGPT 式问答，AI Agent 更强调“目标驱动”和“工具调用”，它不仅能回答问题，还能根据任务规划步骤、调用外部接口、处理数据并返回结果。

本文将从零开始，完整讲解一个 AI Agent 的部署流程，包括环境准备、项目结构设计、后端接口开发、工具调用实现、前端页面搭建、Docker 部署以及上线注意事项。文章中会附带可运行的示例源码，适合希望快速搭建个人 AI Agent、企业知识助手、自动化办公助手或客服机器人的开发者参考。

一、什么是 AI Agent？

AI Agent 可以理解为具备一定自主决策能力的 AI 应用。它通常由以下几个核心部分组成：

1. 大语言模型

   • 负责理解用户意图、生成回答、规划任务。
   • 可以使用 OpenAI、DeepSeek、通义千问、智谱、Claude 等模型。

2. 工具系统

   • Agent 可以调用搜索、数据库、代码执行、文件读取、API 请求等工具。
   • 工具让模型具备“行动能力”。

3. 记忆系统

   • 保存历史对话、用户偏好、任务上下文。
   • 可以是简单的本地 JSON，也可以是 Redis、PostgreSQL、向量数据库。

4. 任务规划能力

   • 将复杂任务拆解成多个步骤。
   • 根据执行结果动态调整后续计划。

5. 应用接口

   • 通常通过 Web API、Web 页面、企业微信、飞书、钉钉、Slack 等方式提供服务。

一个简单的 AI Agent 工作流程如下：

用户输入任务
   ↓
Agent 理解任务
   ↓
判断是否需要调用工具
   ↓
调用搜索 / 数据库 / 文件 / API 等工具
   ↓
整理结果
   ↓
返回最终答案

二、本文要实现的 Agent 功能

本文将实现一个基础版 AI Agent，它具备以下能力：

• 支持用户通过网页输入问题；
• 后端调用大语言模型生成回答；
• 支持简单工具调用，例如获取当前时间、模拟网页搜索；
• 支持多轮对话上下文；
• 提供 REST API；
• 支持 Docker 一键部署；
• 可扩展为知识库问答、自动化工作流或企业内部助手。

技术栈如下：

说明：本文使用 OpenAI 兼容接口写法。如果你使用 DeepSeek、通义千问、智谱等模型，只要它们提供兼容 OpenAI 格式的 API，就可以通过修改 BASE_URL 和 API_KEY 接入。

三、项目目录结构

建议创建如下项目结构：

ai-agent-demo/
├── app/
│   ├── main.py
│   ├── agent.py
│   ├── tools.py
│   └── memory.py
├── static/
│   └── index.html
├── requirements.txt
├── .env.example
├── Dockerfile
├── docker-compose.yml
└── README.md

各文件作用说明：

四、环境准备

1. 安装 Python

建议使用 Python 3.10 或以上版本。

查看版本：

python --version

如果你的系统中同时存在多个 Python 版本，也可以使用：

python3 --version

2. 创建项目目录

mkdir ai-agent-demo
cd ai-agent-demo

3. 创建虚拟环境

python -m venv venv

Windows 激活：

venv\Scripts\activate

macOS / Linux 激活：

source venv/bin/activate

五、安装依赖

创建 requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
python-dotenv==1.0.1
openai==1.45.0
pydantic==2.8.2

安装依赖：

pip install -r requirements.txt

六、配置环境变量

创建 .env.example：

OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o-mini

复制一份为 .env：

cp .env.example .env

然后修改 .env：

OPENAI_API_KEY=你的模型服务API_KEY
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o-mini

如果你使用 DeepSeek，可参考：

OPENAI_API_KEY=你的DeepSeek_API_KEY
OPENAI_BASE_URL=https://api.deepseek.com
MODEL_NAME=deepseek-chat

七、编写工具模块

创建 app/tools.py：

from datetime import datetime
import random


def get_current_time() -> str:
    """
    获取当前系统时间。
    """
    now = datetime.now()
    return now.strftime("%Y-%m-%d %H:%M:%S")


def mock_web_search(query: str) -> str:
    """
    模拟网页搜索工具。
    实际项目中可以替换为搜索引擎 API、企业内部搜索接口或知识库检索。
    """
    fake_results = [
        f"关于「{query}」的搜索结果一：该主题近期受到广泛关注。",
        f"关于「{query}」的搜索结果二：可以从技术、业务和部署三个角度分析。",
        f"关于「{query}」的搜索结果三：建议结合实际场景选择方案。"
    ]

    return "\n".join(random.sample(fake_results, k=3))


def calculator(expression: str) -> str:
    """
    简单计算器工具。
    注意：eval 在真实生产环境中存在安全风险，这里仅用于演示。
    生产环境应使用安全表达式解析库。
    """
    try:
        allowed_chars = "0123456789+-*/(). "
        if not all(ch in allowed_chars for ch in expression):
            return "表达式包含非法字符，无法计算。"

        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算失败：{str(e)}"


TOOLS = {
    "get_current_time": get_current_time,
    "mock_web_search": mock_web_search,
    "calculator": calculator,
}

这里我们实现了三个工具：

1. get_current_time：获取当前时间；
2. mock_web_search：模拟搜索；
3. calculator：简单数学计算。

在生产环境中，这些工具可以被替换为：

• 数据库查询；
• CRM 接口；
• ERP 接口；
• 企业知识库；
• 工单系统；
• 邮件发送；
• 文件解析；
• RPA 自动化流程。

八、编写记忆模块

创建 app/memory.py：

from typing import Dict, List


class ConversationMemory:
    """
    简单内存型对话记忆。
    适合演示和小规模使用。
    生产环境建议替换为 Redis、PostgreSQL 或其他持久化存储。
    """

    def __init__(self, max_messages: int = 10):
        self.max_messages = max_messages
        self.sessions: Dict[str, List[dict]] = {}

    def get_messages(self, session_id: str) -> List[dict]:
        return self.sessions.get(session_id, [])

    def add_message(self, session_id: str, role: str, content: str):
        if session_id not in self.sessions:
            self.sessions[session_id] = []

        self.sessions[session_id].append({
            "role": role,
            "content": content
        })

        if len(self.sessions[session_id]) > self.max_messages:
            self.sessions[session_id] = self.sessions[session_id][-self.max_messages:]

    def clear(self, session_id: str):
        self.sessions[session_id] = []

这个模块负责保存多轮对话内容。为了方便演示，我们使用内存保存。需要注意的是，服务重启后内存会丢失。如果你要部署到生产环境，建议改为 Redis。

九、编写 Agent 核心逻辑

创建 app/agent.py：

import json
import os
from typing import Dict, Any

from dotenv import load_dotenv
from openai import OpenAI

from app.tools import TOOLS

load_dotenv()

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL")
)

MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")


SYSTEM_PROMPT = """
你是一个中文 AI Agent 助手。
你不仅可以回答用户问题，还可以根据需要调用工具。

你可使用的工具如下：

1. get_current_time
   - 功能：获取当前时间
   - 参数：无

2. mock_web_search
   - 功能：模拟搜索网络信息
   - 参数：
     - query: 搜索关键词

3. calculator
   - 功能：计算数学表达式
   - 参数：
     - expression: 数学表达式，例如 "12 * (3 + 4)"

当你认为需要调用工具时，请严格返回如下 JSON 格式：
{
  "tool": "工具名称",
  "arguments": {
    "参数名": "参数值"
  }
}

如果不需要调用工具，请直接用中文回答用户。
不要在 JSON 外输出其他内容。
"""


class SimpleAgent:
    """
    简易 AI Agent。
    实现逻辑：
    1. 接收用户问题和历史上下文；
    2. 询问模型是否需要调用工具；
    3. 如果模型返回工具调用 JSON，则执行工具；
    4. 将工具结果再次交给模型总结；
    5. 返回最终答案。
    """

    def __init__(self):
        self.model = MODEL_NAME

    def chat(self, user_input: str, history: list) -> str:
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT}
        ]

        messages.extend(history)
        messages.append({"role": "user", "content": user_input})

        first_response = self._call_llm(messages)

        tool_call = self._parse_tool_call(first_response)

        if not tool_call:
            return first_response

        tool_name = tool_call.get("tool")
        arguments = tool_call.get("arguments", {})

        if tool_name not in TOOLS:
            return f"模型请求调用未知工具：{tool_name}"

        tool_result = self._execute_tool(tool_name, arguments)

        final_messages = messages + [
            {
                "role": "assistant",
                "content": first_response
            },
            {
                "role": "user",
                "content": f"工具 {tool_name} 的执行结果如下：\n{tool_result}\n请基于该结果给用户一个自然、准确的中文回答。"
            }
        ]

        final_response = self._call_llm(final_messages)

        return final_response

    def _call_llm(self, messages: list) -> str:
        response = client.chat.completions.create(
            model=self.model,
            messages=messages,
            temperature=0.3
        )

        return response.choices[0].message.content

    def _parse_tool_call(self, text: str) -> Dict[str, Any] | None:
        try:
            data = json.loads(text)
            if isinstance(data, dict) and "tool" in data:
                return data
            return None
        except Exception:
            return None

    def _execute_tool(self, tool_name: str, arguments: dict) -> str:
        tool_func = TOOLS[tool_name]

        try:
            if tool_name == "get_current_time":
                return tool_func()

            if tool_name == "mock_web_search":
                return tool_func(arguments.get("query", ""))

            if tool_name == "calculator":
                return tool_func(arguments.get("expression", ""))

            return "工具存在，但未配置执行逻辑。"
        except Exception as e:
            return f"工具执行失败：{str(e)}"

这个 Agent 的实现思路比较简单，但已经具备智能体的核心能力：判断是否需要工具、调用工具、基于工具结果生成最终回答。

十、编写 FastAPI 后端入口

创建 app/main.py：

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from app.agent import SimpleAgent
from app.memory import ConversationMemory


app = FastAPI(title="AI Agent Demo")

agent = SimpleAgent()
memory = ConversationMemory(max_messages=12)


class ChatRequest(BaseModel):
    session_id: str = "default"
    message: str


class ChatResponse(BaseModel):
    answer: str


@app.post("/api/chat", response_model=ChatResponse)
def chat(req: ChatRequest):
    history = memory.get_messages(req.session_id)

    answer = agent.chat(req.message, history)

    memory.add_message(req.session_id, "user", req.message)
    memory.add_message(req.session_id, "assistant", answer)

    return ChatResponse(answer=answer)


@app.post("/api/clear")
def clear_session(req: ChatRequest):
    memory.clear(req.session_id)
    return {"message": "会话已清空"}


app.mount("/", StaticFiles(directory="static", html=True), name="static")

这里提供了两个接口：

十一、编写前端页面

创建 static/index.html：

  
  
  



  
AI Agent Demo

  


  

    
    发送
    清空
  

十二、本地运行项目

在项目根目录执行：

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

浏览器访问：

http://localhost:8000

你可以尝试输入：

现在几点？

Agent 可能会调用 get_current_time 工具，然后返回当前时间。

再输入：

帮我算一下 23 * (18 + 6)

Agent 会调用计算器工具，返回计算结果。

还可以输入：

帮我搜索一下 AI Agent 部署方案

Agent 会调用模拟搜索工具，并整理结果。

十三、Docker 部署

为了让项目更容易部署到服务器，我们可以使用 Docker。

1. 创建 Dockerfile

在项目根目录创建 Dockerfile：

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

2. 创建 docker-compose.yml

创建 docker-compose.yml：

version: "3.9"

services:
  ai-agent:
    build: .
    container_name: ai-agent-demo
    ports:
      - "8000:8000"
    env_file:
      - .env
    restart: always

3. 构建并启动

docker compose up -d --build

查看容器状态：

docker ps

查看日志：

docker logs -f ai-agent-demo

停止服务：

docker compose down

十四、服务器上线部署流程

如果你要部署到云服务器，可以按照以下步骤操作。

1. 安装 Docker

以 Ubuntu 为例：

sudo apt update
sudo apt install -y docker.io docker-compose-plugin
sudo systemctl enable docker
sudo systemctl start docker

2. 上传项目

可以使用 Git：

git clone https://github.com/yourname/ai-agent-demo.git
cd ai-agent-demo

也可以使用 scp 上传：

scp -r ai-agent-demo root@服务器IP:/opt/

3. 配置环境变量

cp .env.example .env
vim .env

填写真实 API Key。

4. 启动服务

docker compose up -d --build

访问：

http://服务器IP:8000

如果你的云服务器开启了防火墙或安全组，需要放行 8000 端口。

十五、使用 Nginx 配置域名访问

如果你希望通过域名访问，例如：

https://agent.example.com

可以使用 Nginx 反向代理。

安装 Nginx：

sudo apt install -y nginx

创建配置：

sudo vim /etc/nginx/sites-available/ai-agent.conf

写入：

server {
    listen 80;
    server_name agent.example.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

启用配置：

sudo ln -s /etc/nginx/sites-available/ai-agent.conf /etc/nginx/sites-enabled/
sudo nginx -t
sudo systemctl reload nginx

如果需要 HTTPS，可以使用 Certbot：

sudo apt install -y certbot python3-certbot-nginx
sudo certbot --nginx -d agent.example.com

十六、生产环境优化建议

上面的示例适合学习和小规模使用。如果要应用到生产环境，还需要做进一步优化。

1. 对话记忆持久化

当前记忆保存在内存中，服务重启后会丢失。建议改为 Redis：

session_id -> messages

或者使用 PostgreSQL 保存完整聊天记录，便于审计和分析。

2. 接入真实搜索工具

示例中的 mock_web_search 只是模拟搜索。实际应用中可以接入：

• Bing Search API；
• Google Custom Search；
• SerpAPI；
• 企业内部搜索接口；
• 向量数据库检索。

3. 增加知识库能力

如果要实现企业知识库问答，可以采用 RAG 架构：

用户问题
  ↓
向量化
  ↓
检索相关文档
  ↓
将文档片段放入 Prompt
  ↓
大模型生成答案

常见向量数据库包括：

• Milvus；
• Weaviate；
• Qdrant；
• Chroma；
• pgvector。

4. 工具调用安全控制

Agent 一旦可以调用外部工具，就必须考虑安全问题。例如：

• 不允许模型直接执行危险命令；
• API 调用需要权限校验；
• 数据库查询要限制范围；
• 文件读取要限制目录；
• 所有工具参数都要校验；
• 关键操作需要人工确认。

5. 增加用户鉴权

如果部署到公网，建议加入登录鉴权，例如：

• JWT；
• OAuth2；
• 企业微信登录；
• 飞书登录；
• 单点登录 SSO。

否则任何人都可能访问你的 Agent，并消耗模型额度。

6. 加入限流机制

为了防止接口被刷，可以加入限流：

• IP 限流；
• 用户级限流；
• Redis 计数器；
• Nginx 限流；
• Cloudflare 防护。

7. 日志与监控

生产环境中建议记录：

• 请求时间；
• 用户 ID；
• Token 消耗；
• 工具调用记录；
• 模型响应耗时；
• 错误日志。

可选工具：

• Prometheus；
• Grafana；
• ELK；
• Loki；
• Sentry。

十七、常见问题排查

1. 启动时报 API Key 错误

检查 .env 是否正确配置：

OPENAI_API_KEY=xxx
OPENAI_BASE_URL=xxx
MODEL_NAME=xxx

同时确认 Docker Compose 是否加载了 .env：

env_file:
  - .env

2. 访问网页空白

检查静态文件目录是否存在：

static/index.html

确认 main.py 中挂载路径正确：

app.mount("/", StaticFiles(directory="static", html=True), name="static")

3. 工具调用失败

可以打印模型返回内容：

print(first_response)

有时模型没有严格返回 JSON，会导致解析失败。可以进一步优化 Prompt，或者使用模型原生 function calling / tool calling 能力。

4. Docker 容器启动失败

查看日志：

docker logs -f ai-agent-demo

常见原因包括：

• 依赖安装失败；
• .env 文件缺失；
• API Key 配置错误；
• 端口被占用。

5. 模型响应很慢

可能原因：

• 模型服务网络延迟；
• 上下文过长；
• 工具执行耗时；
• 服务器网络质量差。

可以通过减少历史消息数量、启用流式输出、选择更快模型来优化。

十八、完整源码汇总

为了方便复制，下面给出关键源码清单。

requirements.txt

fastapi==0.115.0
uvicorn==0.30.6
python-dotenv==1.0.1
openai==1.45.0
pydantic==2.8.2

.env.example

OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o-mini

app/tools.py

from datetime import datetime
import random


def get_current_time() -> str:
    now = datetime.now()
    return now.strftime("%Y-%m-%d %H:%M:%S")


def mock_web_search(query: str) -> str:
    fake_results = [
        f"关于「{query}」的搜索结果一：该主题近期受到广泛关注。",
        f"关于「{query}」的搜索结果二：可以从技术、业务和部署三个角度分析。",
        f"关于「{query}」的搜索结果三：建议结合实际场景选择方案。"
    ]

    return "\n".join(random.sample(fake_results, k=3))


def calculator(expression: str) -> str:
    try:
        allowed_chars = "0123456789+-*/(). "
        if not all(ch in allowed_chars for ch in expression):
            return "表达式包含非法字符，无法计算。"

        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算失败：{str(e)}"


TOOLS = {
    "get_current_time": get_current_time,
    "mock_web_search": mock_web_search,
    "calculator": calculator,
}

app/memory.py

from typing import Dict, List


class ConversationMemory:
    def __init__(self, max_messages: int = 10):
        self.max_messages = max_messages
        self.sessions: Dict[str, List[dict]] = {}

    def get_messages(self, session_id: str) -> List[dict]:
        return self.sessions.get(session_id, [])

    def add_message(self, session_id: str, role: str, content: str):
        if session_id not in self.sessions:
            self.sessions[session_id] = []

        self.sessions[session_id].append({
            "role": role,
            "content": content
        })

        if len(self.sessions[session_id]) > self.max_messages:
            self.sessions[session_id] = self.sessions[session_id][-self.max_messages:]

    def clear(self, session_id: str):
        self.sessions[session_id] = []

十九、后续扩展方向

完成基础版本后，你可以继续扩展以下能力：

1. 接入 RAG 知识库

   • 支持上传 PDF、Word、Markdown；
   • 自动切分、向量化、检索；
   • 根据企业资料回答问题。

2. 支持流式输出

   • 类似 ChatGPT 一样逐字显示；
   • 改善用户体验。

3. 接入多工具工作流

   • 例如自动创建工单；
   • 自动查询订单；
   • 自动发送邮件；
   • 自动生成报表。

4. 增加多模型路由

   • 简单任务使用低成本模型；
   • 复杂任务使用高能力模型；
   • 根据任务类型动态选择模型。

5. 增加人工审批

   • 对涉及转账、删除、发送外部消息等高风险操作，必须由人工确认后执行。

二十、总结

本文从零实现并部署了一个基础版 AI Agent，涵盖了项目结构设计、工具调用、对话记忆、FastAPI 接口、前端页面、Docker 部署以及生产环境优化建议。

这个示例虽然简单，但已经包含 AI Agent 的关键思想：模型负责理解与规划，工具负责执行，记忆负责上下文，接口负责交互。在此基础上，你可以继续扩展知识库、数据库查询、自动化流程、企业系统集成等能力，最终构建出真正可落地的智能助手。

如果只是个人学习，可以直接运行本文源码；如果要用于企业生产环境，则建议重点加强权限控制、日志监控、数据安全、工具调用审计和异常处理。只要架构设计合理，AI Agent 可以成为连接大模型与真实业务系统的重要入口。