基于FastAPI与豆包大模型构建AI Agent应用：完整流程与可运行代码

开发一个功能完整的AI Agent应用，往往涉及框架选择、模型集成与流程设计等多个环节。对于希望快速上手的开发者而言，FastAPI是一个理想的选择：它轻量、高效，能自动生成API文档，并能便捷地与大模型API集成。

本文将构建AI Agent的复杂过程拆解为六个清晰的步骤，涵盖从环境搭建到部署上线的全流程，并提供可直接运行的完整代码，帮助开发者快速实现从0到1的突破。

一、环境准备：安装核心依赖

首先，需要安装项目运行所必需的核心Python包。

pip install fastapi uvicorn openai python-dotenv requests

解析:

• fastapi: Web应用框架核心。
• uvicorn: 用于运行FastAPI应用的ASGI服务器。
• openai: OpenAI兼容的SDK，用于调用大模型服务。
• python-dotenv: 用于管理环境变量，避免将敏感信息硬编码在代码中。
• requests: 用于调用外部API（如天气查询工具）。

二、配置环境变量

在项目根目录创建.env文件，用于安全地存储API密钥等配置信息。

# .env 文件
DOUBAO_API_KEY="你的豆包大模型API密钥"
DOUBAO_MODEL="doubao-lite"
AMAP_WEATHER_KEY="你的高德天气API密钥"

提示：doubao-lite模型拥有充足的免费额度且响应迅速，非常适合开发测试。

三、搭建FastAPI基础框架

创建一个main.py文件，作为应用的入口，并编写最基础的FastAPI应用代码。

from fastapi import FastAPI
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 创建FastAPI应用实例
app = FastAPI(
    title="AI Agent 应用",
    description="基于FastAPI与豆包大模型构建的智能体",
    version="1.0.0"
)

@app.get("/")
def read_root():
    return {"message": "AI Agent 服务已启动，请访问 /docs 查看接口文档"}

使用以下命令启动开发服务器：

uvicorn main:app --reload

启动后，访问 http://127.0.0.1:8000/docs 即可看到自动生成的交互式API文档。

四、集成大模型思考能力

AI Agent的核心在于其“思考”逻辑，即理解用户意图并决定后续行动。这里我们通过调用豆包大模型来实现。

1. 封装大模型调用函数

在main.py中添加以下代码，实现对豆包大模型API的调用。

import os
import json
from openai import OpenAI

# 初始化客户端（配置为豆包服务地址）
api_key = os.getenv("DOUBAO_API_KEY")
if not api_key:
    raise ValueError("请在 .env 文件中配置 DOUBAO_API_KEY")

client = OpenAI(
    base_url="https://ark.cn-beijing.volces.com/api/v3",
    api_key=api_key,
)

def call_llm(prompt: str) -> str:
    """调用大模型并返回文本响应"""
    try:
        response = client.chat.completions.create(
            model=os.getenv("DOUBAO_MODEL", "doubao-lite"),
            messages=[
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "text",
                            "text": prompt
                        }
                    ]
                }
            ]
        )
        return response.choices[0].message.content.strip()
    except Exception as e:
        return f"模型调用失败：{str(e)}"

2. 实现Agent思考逻辑

该函数负责分析用户任务，判断是否需要调用工具，并结构化输出决策。

def agent_think(task: str) -> dict:
    """
    Agent思考逻辑
    返回格式示例：
    {"need_tool": true, "tool_name": "get_weather", "tool_params": {"city": "北京"}}
    或
    {"need_tool": false, "answer": "这是大模型的直接回答"}
    """
    prompt = f"""
    你是一个AI Agent，需要处理用户任务：{task}
    请根据以下规则判断：
    1. 如果任务是获取实时信息（如天气、新闻）、执行计算、操作数据，则需要调用工具。
    2. 其他任务（如问答、总结、创作）则直接回答，无需调用工具。
    3. 若需要调用工具，请严格按照此JSON格式返回：
    {{"need_tool": true, "tool_name": "工具名", "tool_params": {{"参数名": "参数值"}}}}
    4. 若无需调用工具，请返回：
    {{"need_tool": false, "answer": "你的回答内容"}}

    当前可用工具：
    - get_weather: 获取天气，参数: city (城市名，字符串)
    """

    llm_response = call_llm(prompt)
    try:
        return json.loads(llm_response)
    except json.JSONDecodeError:
        # 如果模型返回非标准JSON，则视作直接回答
        return {"need_tool": False, "answer": llm_response}

五、扩展Agent工具库

工具是Agent与外界交互的“手”。我们以实现一个“获取天气”的工具为例。

1. 实现天气查询工具函数

import requests

def get_weather(city: str) -> str:
    """工具函数：调用高德天气API查询实时天气"""
    url = "https://restapi.amap.com/v3/weather/weatherInfo"
    params = {
        "key": os.getenv("AMAP_WEATHER_KEY"),
        "city": city,
        "extensions": "base"  # base为实时天气
    }

    try:
        resp = requests.get(url, params=params)
        data = resp.json()
        if data.get("status") == "1" and data.get("lives"):
            info = data["lives"][0]
            return (f"{city}当前天气：{info['weather']}，"
                    f"温度：{info['temperature']}℃，"
                    f"湿度：{info['humidity']}%，"
                    f"风向：{info['winddirection']}风{info['windpower']}级")
        else:
            return f"获取天气失败：{data.get('info', '未知错误')}"
    except Exception as e:
        return f"工具执行异常：{str(e)}"

2. 创建工具调用调度器

def call_tool(tool_name: str, tool_params: dict) -> str:
    """工具调用中心，根据名称分发任务"""
    tool_map = {
        "get_weather": get_weather,
        # 未来可在此处扩展更多工具，如：”calculator”, “translator”
    }

    if tool_name not in tool_map:
        return f"错误：未找到工具 '{tool_name}'"

    try:
        return tool_map[tool_name](**tool_params)
    except Exception as e:
        return f"工具 '{tool_name}' 调用失败：{str(e)}"

六、创建核心API接口

最后，将上述能力整合为一个完整的API接口，供前端或客户端调用。

1. 定义数据模型

from pydantic import BaseModel

class TaskRequest(BaseModel):
    task: str
    user_id: str | None = None  # 可选，用于多用户会话管理

class AgentResponse(BaseModel):
    code: int = 200
    message: str = "success"
    data: dict  # 包含任务处理详情

2. 实现Agent处理接口

@app.post("/agent/process", response_model=AgentResponse)
def process_task(request: TaskRequest):
    """AI Agent核心处理接口"""
    try:
        # 第一步：Agent思考
        think_result = agent_think(request.task)

        if think_result.get("need_tool"):
            # 第二步：调用工具
            tool_result = call_tool(
                think_result["tool_name"],
                think_result["tool_params"]
            )
            response_data = {
                "task": request.task,
                "thought": "任务需调用工具处理",
                "tool_used": think_result["tool_name"],
                "tool_params": think_result["tool_params"],
                "final_result": tool_result
            }
        else:
            # 无需工具，直接返回模型回答
            response_data = {
                "task": request.task,
                "thought": "任务无需调用工具",
                "final_result": think_result.get("answer", "模型未返回答案")
            }

        return AgentResponse(data=response_data)

    except Exception as e:
        return AgentResponse(code=500, message="处理失败", data={"error": str(e)})

启动服务后，通过 POST /agent/process 接口即可与AI Agent交互。例如，发送 {"task": "查询北京天气"}，Agent将自动调用天气工具并返回结果。

部署与后续扩展

完成开发后，你可以将应用部署到各类支持Python的云服务平台（如PythonAnywhere, Railway, 或你自己的云服务器），使其能够被公开访问。

后续优化方向：

• 增加会话记忆：在TaskRequest中引入session_id，并在后端维护对话历史，使Agent具备上下文理解能力。
• 扩展工具集：在tool_map中添加更多工具函数，如计算器、数据库查询、文件处理等。
• 添加认证中间件：为API接口增加API Key验证，提升安全性。
• 构建前端界面：使用Vue或React开发一个简单的聊天式Web界面，提供更佳用户体验。

通过以上六个步骤，我们完成了一个具备基础思考与工具调用能力的AI Agent应用。这个框架清晰、易于扩展，为后续开发更复杂的智能体功能奠定了坚实基础。