基于Dify与FastAPI,手把手构建地图场景AI智能体原型
最近在跟进地图应用与AI结合的新趋势时,发现Google Maps的一项重大更新引发了开发者社区的广泛讨论。其内置的“Ask Maps”功能,从最初基于大语言模型的简单问答,进化到了能够直接处理“订餐”、“酒店预订”这类复杂、多步骤任务的智能体(Agent)。这不仅仅是功能的叠加,更标志着地图应用正从一个被动的“信息查询工具”,转变为一个主动的“任务执行平台”。对于开发者而言,理解其背后的“智能体”技术架构,并思考如何在自己的应用中实现类似的能力,变得至关重要。
本文将深入剖析Google Maps这一新功能背后的技术逻辑,并基于当前主流的AI智能体开发框架(如Dify、Coze等),手把手带你从零搭建一个具备“订餐”或“酒店预订”能力的简易智能体原型。无论你是想了解AI智能体如何落地,还是希望为自己的项目添加类似的自动化任务处理能力,这篇文章都将提供一套完整的实战思路和代码示例。
1. 智能体(Agent)在地图场景中的应用与核心概念
在传统的交互模式中,用户使用地图应用通常遵循“搜索 -> 浏览 -> 选择 -> 跳转”的线性路径。例如,找餐厅需要先输入关键词,然后在一堆结果中筛选,再点开详情页查看菜单和评价,最后可能需要跳转到外卖平台或拨打订餐电话。这个过程是割裂且低效的。
智能体(AI Agent) 的出现,旨在彻底改变这一模式。它不是一个简单的聊天机器人,而是一个具备 感知、规划、记忆、行动和反思 能力的软件实体。在地图场景中,智能体的核心价值在于 理解用户的高层意图,并自动协调多个工具(Tools)或服务(Services)来完成一个端到端的任务 。
以“Ask Maps”的新功能为例:
- 用户意图 :“帮我在这附近找一家评价不错的意大利餐厅,并预订今晚7点两人位。”
- 传统模式 :用户需要手动执行上述所有步骤。
- 智能体模式 :
- 感知/理解 :智能体解析用户自然语言,识别核心实体:菜系(意大利)、条件(评价不错)、时间(今晚7点)、人数(两人)、动作(预订)。
- 规划 :智能体内部规划任务流:a) 调用本地搜索API查找符合条件的餐厅;b) 调用评分系统过滤结果;c) 调用餐厅预订系统的API查询空位并下单。
- 行动 :智能体依次执行上述规划,调用相应的工具。
- 反思/反馈 :将每一步的结果(找到的餐厅列表、可预订的时间、预订成功确认)以自然、连贯的方式反馈给用户。
这个过程中,智能体扮演了“虚拟助手”和“自动化流程调度中心”的角色。它背后的关键技术包括:
- 大语言模型(LLM) :负责理解用户意图、拆解任务、生成规划、以及组织自然语言回复。它是智能体的“大脑”。
- 工具调用(Tool Calling) :智能体“手”和“脚”。LLM决定在何时、以何种参数调用哪个外部工具(如搜索API、数据库、支付接口)。
- 记忆(Memory) :分为短期会话记忆(记住当前对话上下文)和长期记忆(存储用户偏好、历史订单),用于实现连贯的个性化服务。
- 工作流(Workflow) :对于“订餐”这类固定流程,可以预定义工作流(如:查询->筛选->确认->支付),智能体按步骤执行,提高可靠性和效率。
理解这些概念,是我们复现类似功能的基础。
2. 环境准备与开发框架选型
要搭建一个演示性质的“订餐智能体”,我们不需要从零开始造轮子。市面上已有多个优秀的智能体开发平台和框架,可以大幅降低开发门槛。这里我们选择 Dify 和 Coze 作为主要工具进行讲解,因为它们提供了可视化的编排界面和丰富的集成能力,非常适合快速原型开发。
2.1 核心工具与框架说明
- Dify :一个开源的LLM应用开发平台,核心功能是让开发者通过可视化工作流的方式,快速构建基于LLM的应用程序,包括智能体。它支持多种模型(GPT、Claude、国产大模型等),内置了知识库、文本提取、代码解释器等丰富工具,并可以轻松通过API接入自定义工具。
- Coze :字节跳动推出的AI Bot开发平台,同样强调低代码和可视化。它集成了丰富的插件(如搜索引擎、天气、日历等),并且发布和分享非常便捷,适合构建面向最终用户的对话式应用。
- 语言模型 :我们将使用OpenAI的GPT系列模型(如gpt-3.5-turbo)作为智能体的“大脑”。你需要准备一个有效的OpenAI API Key。
- 模拟后端服务 :为了演示“订餐”和“酒店预订”,我们需要模拟几个后端API。这里我们将使用 Python + FastAPI 快速搭建几个简单的HTTP接口,模拟餐厅查询、座位预订、酒店搜索、价格比对等功能。
2.2 本地开发环境搭建
我们将创建一个本地开发项目,包含智能体逻辑(使用Dify/Coze云服务或本地部署)和模拟的后端服务。
项目结构预览:
food-hotel-agent-demo/
├── backend/ # 模拟后端服务
│ ├── main.py # FastAPI 主应用
│ ├── requirements.txt # Python依赖
│ └── ...
├── agent-config/ # 智能体配置导出文件(可选)
│ └── dify-workflow.json
└── README.md
第一步:准备Python环境 确保你的系统已安装Python 3.8+。建议使用虚拟环境。
# 创建项目目录
mkdir food-hotel-agent-demo && cd food-hotel-agent-demo
mkdir backend
# 创建并激活虚拟环境 (Windows用户请使用 `venv\Scripts\activate`)
python -m venv venv
source venv/bin/activate # Linux/Mac
# venv\Scripts\activate # Windows
第二步:安装后端依赖 在 backend 目录下创建 requirements.txt 文件:
fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
requests==2.31.0
然后安装:
cd backend
pip install -r requirements.txt
第三步:获取OpenAI API Key 访问 OpenAI平台 ,创建一个新的API Key并妥善保存。我们稍后会在Dify或Coze中配置。
至此,基础开发环境就准备好了。接下来,我们先搭建模拟的后端服务,这是智能体需要调用的“工具”。
3. 构建模拟服务:餐厅与酒店API
智能体需要通过API与真实世界交互。我们先构建几个简单的模拟API,它们将扮演“餐厅数据库”和“酒店预订系统”的角色。
3.1 创建FastAPI应用与数据模型
在 backend/main.py 文件中,我们开始编写代码:
# backend/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
import uuid
app = FastAPI(title="模拟餐厅与酒店服务API", description="用于智能体功能演示的模拟后端")
# ---------- 数据模型定义 ----------
class Restaurant(BaseModel):
id: str
name: str
cuisine: str # 菜系,如 Italian, Chinese
rating: float
address: str
open_time: str # 简化处理,如 "18:00"
close_time: str
class TableQuery(BaseModel):
restaurant_id: str
date: str # YYYY-MM-DD
time: str # HH:MM
party_size: int
class TableBooking(BaseModel):
query: TableQuery
customer_name: str
customer_phone: str
class Hotel(BaseModel):
id: str
name: str
star_rating: int
address: str
price_per_night: float
available_rooms: int
class HotelSearchQuery(BaseModel):
location: str
check_in_date: str
check_out_date: str
guests: int
class HotelBooking(BaseModel):
hotel_id: str
search_query: HotelSearchQuery
customer_name: str
customer_email: str
# ---------- 模拟内存数据库 ----------
# 模拟餐厅数据
mock_restaurants = [
Restaurant(id="R001", name="Mario's Trattoria", cuisine="Italian", rating=4.5, address="123 Main St", open_time="11:00", close_time="22:00"),
Restaurant(id="R002", name="Dragon Palace", cuisine="Chinese", rating=4.2, address="456 Elm St", open_time="10:30", close_time="21:30"),
Restaurant(id="R003", name="Le Bistro Parisien", cuisine="French", rating=4.7, address="789 Oak St", open_time="12:00", close_time="23:00"),
]
# 模拟酒店数据
mock_hotels = [
Hotel(id="H001", name="Grand Plaza Hotel", star_rating=5, address="1 Central Ave", price_per_night=299.99, available_rooms=10),
Hotel(id="H002", name="Cozy Inn", star_rating=3, address="2 River Rd", price_per_night=129.50, available_rooms=5),
Hotel(id="H003", name="Sunset Resort", star_rating=4, address="3 Beach Blvd", price_per_night=220.00, available_rooms=8),
]
# 存储预订记录(实际应用中应使用数据库)
restaurant_bookings = []
hotel_bookings = []
3.2 实现餐厅查询与预订API
接下来,添加处理餐厅相关请求的端点:
# backend/main.py (续)
# ---------- 餐厅相关API ----------
@app.get("/restaurants/search", response_model=List[Restaurant])
async def search_restaurants(cuisine: Optional[str] = None, min_rating: Optional[float] = None):
"""根据菜系和最低评分搜索餐厅"""
results = mock_restaurants
if cuisine:
# 简单过滤,实际应为数据库查询
results = [r for r in results if r.cuisine.lower() == cuisine.lower()]
if min_rating:
results = [r for r in results if r.rating >= min_rating]
return results
@app.get("/restaurants/{restaurant_id}", response_model=Restaurant)
async def get_restaurant(restaurant_id: str):
"""根据ID获取餐厅详情"""
for r in mock_restaurants:
if r.id == restaurant_id:
return r
raise HTTPException(status_code=404, detail="Restaurant not found")
@app.post("/restaurants/check-availability")
async def check_table_availability(query: TableQuery):
"""检查指定时间是否有空位(模拟逻辑)"""
# 这里模拟一个简单的检查逻辑:如果时间在营业时间内,且人数小于6人,则认为有空位
for r in mock_restaurants:
if r.id == query.restaurant_id:
if query.time >= r.open_time and query.time <= r.close_time:
# 模拟随机可用性
import random
is_available = random.choice([True, False])
return {
"available": is_available,
"restaurant_name": r.name,
"message": "Table available" if is_available else "No table available at this time"
}
raise HTTPException(status_code=404, detail="Restaurant not found or outside business hours")
@app.post("/restaurants/book", response_model=dict)
async def book_table(booking: TableBooking):
"""预订餐桌"""
# 1. 检查可用性
availability = await check_table_availability(booking.query)
if not availability.get("available"):
raise HTTPException(status_code=400, detail="Table is not available for the selected time.")
# 2. 创建预订记录(模拟)
booking_id = str(uuid.uuid4())[:8]
booking_record = {
"booking_id": booking_id,
**booking.dict(),
"status": "confirmed",
"created_at": datetime.now().isoformat()
}
restaurant_bookings.append(booking_record)
return {
"success": True,
"booking_id": booking_id,
"message": f"Table successfully booked at {availability['restaurant_name']} for {booking.query.time} on {booking.query.date}.",
"details": booking_record
}
3.3 实现酒店搜索与预订API
同样地,添加酒店相关的端点:
# backend/main.py (续)
# ---------- 酒店相关API ----------
@app.get("/hotels/search", response_model=List[Hotel])
async def search_hotels(location: str, max_price: Optional[float] = None):
"""根据位置和最高价格搜索酒店"""
results = [h for h in mock_hotels if location.lower() in h.address.lower()]
if max_price:
results = [h for h in results if h.price_per_night <= max_price]
# 按评分或价格排序
results.sort(key=lambda x: x.star_rating, reverse=True)
return results
@app.post("/hotels/book", response_model=dict)
async def book_hotel(booking: HotelBooking):
"""预订酒店"""
# 1. 查找酒店并检查房间
hotel = None
for h in mock_hotels:
if h.id == booking.hotel_id:
hotel = h
break
if not hotel:
raise HTTPException(status_code=404, detail="Hotel not found")
if hotel.available_rooms < 1:
raise HTTPException(status_code=400, detail="No rooms available")
# 2. 模拟扣减房间并创建订单
hotel.available_rooms -= 1 # 注意:内存操作,重启后重置
booking_id = str(uuid.uuid4())[:8]
booking_record = {
"booking_id": booking_id,
"hotel_name": hotel.name,
**booking.dict(),
"total_price": hotel.price_per_night * 2, # 简化计算,住两晚
"status": "confirmed",
"created_at": datetime.now().isoformat()
}
hotel_bookings.append(booking_record)
return {
"success": True,
"booking_id": booking_id,
"message": f"Booking confirmed at {hotel.name} for {booking.search_query.check_in_date} to {booking.search_query.check_out_date}.",
"details": booking_record
}
3.4 启动模拟服务
在 backend 目录下,运行以下命令启动服务:
uvicorn main:app --reload --host 0.0.0.0 --port 8000
服务启动后,访问 http://127.0.0.1:8000/docs 即可看到自动生成的API交互文档(Swagger UI),你可以在这里测试各个接口。
我们的“工具”已经准备好了。接下来,最关键的一步:创建能够理解和调用这些工具的智能体。
4. 在Dify平台上构建订餐/酒店预订智能体
Dify的核心优势在于其 可视化工作流 编排能力。我们将创建一个智能体,它能够理解用户关于餐厅和酒店的复杂请求,并自动调用我们刚搭建的模拟API。
4.1 初始化Dify应用与配置
- 访问Dify :如果你没有自托管Dify,可以使用其 云服务 。注册并登录。
- 创建新应用 :点击“创建新应用”,选择“智能体(Agent)类型”。
- 配置模型与提示词 :
- 模型提供商 :选择“OpenAI”。
- 模型 :选择
gpt-3.5-turbo或gpt-4。 - API Key :填入你的OpenAI API Key。
- 提示词(Prompt) :这是智能体的“人格”和核心指令。输入如下内容:
你是一个专业的餐饮和旅行助手,集成在地图应用中。你的目标是帮助用户找到餐厅并预订座位,或者搜索并预订酒店。 能力: 1. 当用户想找餐厅或订座时,你需要询问关键信息:菜系偏好、用餐时间、人数、地点(如果未提供)。 2. 当用户想预订酒店时,你需要询问:入住日期、退房日期、入住人数、地点、价格区间。 3. 根据用户提供的信息,调用相应的工具(函数)来获取数据或执行操作。 4. 将工具返回的结果,用友好、清晰、自然的方式组织成回复给用户。 注意: - 一次只处理一个主要请求(要么是餐厅,要么是酒店)。 - 如果信息不足,主动、礼貌地询问用户。 - 不要编造餐厅或酒店的信息,所有数据必须来自工具调用。 - 预订完成后,务必提供预订ID和确认信息。
4.2 配置工具(Tools)——连接模拟API
这是智能体拥有“行动力”的关键。我们需要将第3步创建的API封装成Dify能调用的工具。
-
创建“搜索餐厅”工具 :
- 在应用编辑页,进入“工具”标签页,点击“添加工具”。
- 选择“自定义工具” -> “通过 HTTP 请求”。
- 工具名称 :
search_restaurants - 描述 :
根据菜系和最低评分搜索附近的餐厅。 - 请求方法 :
GET - URL :
http://127.0.0.1:8000/restaurants/search(确保你的后端服务正在运行) - 参数设置 :点击“添加参数”。
- 参数1:
name: cuisine,type: string,required: false,description: 菜系,例如 Italian, Chinese - 参数2:
name: min_rating,type: number,required: false,description: 最低评分,例如 4.0
- 参数1:
- 请求头 :可留空,或根据需要添加
Content-Type: application/json。 - 响应解析 :这是一个关键步骤。Dify需要知道如何从API返回的JSON中提取文本内容给LLM“阅读”。
- 在“响应内容提取”中,填写JSONPath表达式:
$[*].name。这表示提取返回数组中每个对象的name字段,组合成文本。你也可以提取更多信息,如$[*].{name: name, cuisine: cuisine, rating: rating, address: address}。
- 在“响应内容提取”中,填写JSONPath表达式:
-
创建“检查餐桌可用性”工具 :
- 名称:
check_table_availability - 描述:
检查指定餐厅在特定日期、时间和人数下是否有空位。 - 方法:
POST - URL:
http://127.0.0.1:8000/restaurants/check-availability - 参数(通过请求体Body传递):
{ "restaurant_id": "{{restaurant_id}}", "date": "{{date}}", "time": "{{time}}", "party_size": {{party_size}} }- 注意:
{{}}是Dify的变量插值语法,值来自对话上下文或用户输入。
- 注意:
- 响应内容提取:
$.message(提取返回的message字段)。
- 名称:
-
创建“预订餐桌”工具 :
- 名称:
book_table - 描述:
在指定的餐厅、时间、日期为指定人数的顾客预订座位。需要顾客姓名和电话。 - 方法:
POST - URL:
http://127.0.0.1:8000/restaurants/book - 参数(Body):
{ "query": { "restaurant_id": "{{restaurant_id}}", "date": "{{date}}", "time": "{{time}}", "party_size": {{party_size}} }, "customer_name": "{{customer_name}}", "customer_phone": "{{customer_phone}}" } - 响应内容提取:
$.message。
- 名称:
-
同理,创建“搜索酒店” (
search_hotels, GET,/hotels/search,参数location,max_price)和**“预订酒店”** (book_hotel, POST,/hotels/book)工具。参数配置参考上面的格式和酒店API的模型定义。
4.3 测试与调试智能体
工具配置完成后,回到应用的“对话”或“预览”界面。
- 基础测试 :尝试输入“帮我找一家意大利餐厅”。智能体应该会询问你用餐时间、人数等信息。然后,它应该自动调用
search_restaurants工具(你可以在Dify的对话日志中看到工具调用记录),并将结果返回给你。 - 完整流程测试 :进行一个端到端的对话。
- 用户:“我想预订今晚7点,2个人的意大利餐厅座位。”
- 智能体:可能会询问餐厅名称或从搜索结果中选择。假设它列出了“Mario's Trattoria”。
- 用户:“就订Mario's吧。”
- 智能体:应调用
check_table_availability检查空位,然后询问你的姓名和电话。 - 用户:“我叫张三,电话是13800138000。”
- 智能体:应调用
book_table工具,并返回预订成功的确认信息和预订ID。
通过这个可视化编排的过程,你无需编写复杂的逻辑代码,就创建了一个能理解意图、规划步骤、调用工具并完成任务的智能体。这模拟了“Ask Maps”新功能的核心交互逻辑。
5. 进阶:使用工作流实现更复杂的决策逻辑
单纯依赖LLM的零样本(Zero-shot)工具调用有时不可靠,比如在需要严格顺序或复杂条件判断时。Dify的 工作流(Workflow) 功能可以解决这个问题。我们可以为“订餐”设计一个固定流程。
- 创建工作流 :在Dify中新建一个“工作流”类型的应用。
- 设计节点 :
- 开始节点 :接收用户输入。
- LLM节点(意图分类) :使用Prompt让LLM判断用户意图是“餐厅”还是“酒店”。
- 条件分支节点 :根据意图,将流程导向“餐厅预订子流程”或“酒店预订子流程”。
- 餐厅预订子流程 :
- 变量提取节点 :用LLM或正则从对话中提取
cuisine,time,date,party_size。 - HTTP请求节点 :调用
search_restaurantsAPI。 - LLM节点(选择餐厅) :让LLM根据用户可能偏好(如评分最高)从结果中推荐一个。
- HTTP请求节点 :调用
check_table_availability。 - 提问节点 :如果可用,向用户提问获取
customer_name和customer_phone。 - HTTP请求节点 :调用
book_table。 - 结束节点 :返回成功信息。
- 变量提取节点 :用LLM或正则从对话中提取
- 酒店预订子流程 :类似结构。
工作流提供了更强的流程控制、错误处理和变量传递能力,适合对稳定性和确定性要求更高的生产场景。你可以将编排好的工作流发布为一个API,供你的前端应用调用。
6. 常见问题与排查思路
在开发和集成此类智能体时,你可能会遇到以下典型问题:
| 问题现象 | 可能原因 | 排查与解决思路 |
|---|---|---|
| 智能体不调用工具 | 1. Prompt描述不清,LLM未理解需要调用工具。 2. 工具的描述(Description)不够准确,LLM无法匹配。 3. 用户请求过于模糊,LLM在等待更多信息。 |
1. 在Prompt中明确指令,如“你必须使用工具来获取真实数据”。 2. 优化工具描述,使其更贴近自然语言问题,例如将 search_restaurants 描述为“当用户想找餐厅时使用此工具”。 3. 设计对话逻辑,让智能体主动询问缺失的关键参数(如时间、人数)。 |
| 工具调用失败(API错误) | 1. 网络不通,本地服务未启动或URL错误。 2. API参数格式错误(如类型不匹配、缺少必填字段)。 3. 后端API返回非2xx状态码。 |
1. 在Dify的工具配置页面,使用“测试”功能直接调用,查看原始请求和响应。 2. 检查后端API日志,确认收到的请求体是否符合预期。 3. 确保Dify中参数映射的变量名与后端接口定义一致。使用 curl 或Postman单独测试后端API。 |
| LLM无法解析API返回结果 | 1. API返回的JSON结构过于复杂或嵌套太深。 2. Dify中“响应内容提取”的JSONPath表达式写错。 |
1. 简化API返回的数据结构,只返回LLM需要的关键信息。 2. 使用在线JSONPath验证器测试你的表达式。在Dify工具测试中,观察“模型接收到的内容”是否是你期望的文本。 |
| 智能体陷入循环或逻辑混乱 | 1. Prompt中存在矛盾指令。 2. 工作流中的条件分支设置错误。 3. 会话历史过长导致模型困惑。 |
1. 精简和优化Prompt,确保指令单一、明确。 2. 在工作流中增加调试节点,输出变量值以检查流程。 3. 在Dify中设置“会话记忆”条数限制,或使用“总结式记忆”来压缩历史。 |
| 处理多轮对话时状态丢失 | 智能体没有记住之前的关键信息(如选中的餐厅ID)。 | 1. 在Dify的“上下文变量”中,将关键信息(如 restaurant_id )设置为“变量”,使其能在整个会话中传递。 2. 在工作流中,使用“变量分配器”节点来存储和更新状态。 |
7. 最佳实践与工程化建议
将演示原型转化为一个稳定、可扩展的生产级应用,需要考虑更多因素:
-
工具设计的原子性与复用性 :每个工具(API)应职责单一。例如,
search_restaurants只负责搜索,check_availability只负责检查。这便于组合和复用。避免创建“一站式”的庞杂API。 -
全面的错误处理与用户反馈 :智能体必须能优雅地处理失败。例如,当API调用失败时,不应将内部错误信息直接抛给用户,而应回复:“抱歉,餐厅查询服务暂时不可用,请您稍后再试或尝试其他选择。” 在工作流中,要为每个HTTP节点配置失败分支。
-
安全性 :
- API密钥管理 :切勿在前端或客户端代码中硬编码API Key。Dify等平台应配置在环境变量中。对自建的后端API,应实施认证(如API Token、JWT)。
- 用户输入净化 :传递给后端API的所有用户输入都必须进行验证和转义,防止SQL注入或命令注入。
- 权限控制 :确保智能体只能调用其被授权的工具。例如,预订工具可能需要用户先登录。
-
性能与成本优化 :
- 缓存 :对频繁查询且变化不快的(如餐厅列表、酒店信息),引入缓存层(Redis),减少对LLM和下游API的调用。
- LLM调用优化 :使用更高效的模型(如
gpt-3.5-turbo而非gpt-4)处理简单任务。精心设计Prompt以减少Token消耗。 - 异步处理 :对于耗时的操作(如确认邮件发送、支付处理),智能体可以立即响应“已受理”,然后通过Webhook或轮询告知用户最终结果。
-
可观测性 :在生产环境中,必须记录智能体的完整决策链路。
- 日志记录 :记录每轮对话的用户输入、LLM的完整思考过程(如果支持)、工具调用详情(请求/响应)、最终回复。
- 监控指标 :监控API调用延迟、失败率、Token使用量、用户会话满意度等。
通过以上步骤,你不仅能够复现一个类似“Ask Maps”智能体功能的演示,更能掌握构建实用AI智能体的核心方法论——即围绕 意图理解、工具抽象、流程编排、可靠交互 这四个支柱进行系统设计。这种模式可以扩展到客服、数据分析、内部流程自动化等无数场景,是当前AI应用落地最具潜力的方向之一。
更多推荐
所有评论(0)