AI-Agent入门实战-AI私厨

AI Agent入门实战：从零搭建一个"AI私厨"应用

配套视频：黑马程序员2026最新版LangChain+LangGraph开发实战

引言

在人工智能飞速发展的今天，AI Agent（智能体）已经成为大模型应用的主流形态。与简单的问答不同，Agent 能够感知环境、决策行动、调用工具、保持记忆，从而完成复杂的多步骤任务。

本篇文章，我们将从零开始，手把手教你搭建一个多模态 AI 私厨应用：

用户上传冰箱/厨房的照片，AI 自动识别食材，搜索相关食谱，并按营养价值和难度排序推荐给用户。

整个项目基于LangChain + LangGraph架构，使用FastAPI提供后端 RESTful 接口，前端采用Next.js构建单页应用。核心技术点包括：

• 多模态大模型：支持图片 + 文本的联合输入
• Web 搜索工具：实时搜索食谱（ Tavily 搜索）
• Agent 记忆系统：基于 SQLite 的会话历史持久化
• 流式输出：SSE 实时推送 AI 响应

一、环境准备

1.1 依赖安装

在项目根目录下创建pyproject.toml，声明所有依赖：

[project] name = "food-recipe-recommender" version = "0.1.0" description = "AI-powered recipe recommender based on uploaded food images" requires-python = ">=3.10" dependencies = [ "langchain>=0.3.0", "langchain-community>=0.3.0", "langchain-openai>=0.2.0", "langchain-core>=0.3.0", "langchain-tavily>=0.1.0", "streamlit>=1.40.0", "pillow>=11.0.0", "python-dotenv>=1.0.0", "fastapi>=0.109.0", "uvicorn>=0.27.0", "python-multipart>=0.0.6", "alibabacloud-oss-v2>=1.2.4", "langgraph-checkpoint-sqlite>=3.0.3", ]

安装依赖：

uv sync# 或使用 pippip install-r requirements.txt

1.2 API Key 配置

在项目根目录创建.env文件：

# 阿里百炼（通义千问）API - 支持多模态输入 DASHSCOPE_API_KEY=你的API_KEY DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1 # Tavily Web 搜索 API - 用于搜索食谱 TAVILY_API_KEY=你的API_KEY # LangSmith 可选 - 用于调试和监控 Agent 运行时 LANGSMITH_API_KEY=你的API_KEY LANGSMITH_TRACING=true LANGSMITH_PROJECT=lc-course # 阿里云 OSS（可选）- 用于图片上传 OSS_ACCESS_KEY_ID=你的AccessKey_ID OSS_ACCESS_KEY_SECRET=你的AccessKey_Secret OSS_BUCKET=你的Bucket名称 OSS_ENDPOINT=oss-cn-heyuan.aliyuncs.com

1.3 项目结构

项目根目录/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── agents/ │ │ ├── __init__.py │ │ └── personal_chief.py │ ├── api/ │ │ └── v1/ │ │ ├── __init__.py │ │ ├── chat.py │ │ └── oss.py │ ├── models/ │ │ ├── __init__.py │ │ └── schemas.py │ ├── common/ │ │ ├── __init__.py │ │ └── logger.py │ └── db/ # SQLite数据库目录（自动创建） ├── .env └── pyproject.toml

二、完整源码

2.1app/__init__.py

# app/__init__.py# 包初始化文件，可以为空

2.2app/common/__init__.py

# app/common/__init__.py# 公共模块初始化文件

2.3app/common/logger.py

# app/common/logger.pyimportloggingimportsys# 配置日志格式：时间 - 级别 - 模块 - 消息LOG_FORMAT="%(asctime)s - %(levelname)s - %(name)s - %(message)s"defsetup_logging():""" 初始化日志配置。 - StreamHandler：将日志输出到控制台（stdout），方便在终端实时查看 - FileHandler（注释掉）：如果需要持久化到文件，可取消注释 """logging.basicConfig(level=logging.INFO,# INFO级别起步，生产环境可改为WARNINGformat=LOG_FORMAT,handlers=[logging.StreamHandler(sys.stdout),# 输出到控制台# logging.FileHandler("app.log") # 追加模式写入文件])# 创建一个全局的logger实例，供其他模块直接调用# 使用__name__可以自动显示调用方的模块名，便于定位日志来源logger=logging.getLogger("personal_chief")

2.4app/models/__init__.py

# app/models/__init__.py# 数据模型模块初始化文件

2.5app/models/schemas.py

# app/models/schemas.pyfromtypingimportOptional,ListfrompydanticimportBaseModelclassChatRequest(BaseModel):""" 聊天请求的数据模型。 使用Pydantic做运行时类型校验，FastAPI自动从请求体解析字段。 """message:str# 用户输入的文本内容image_url:Optional[str]=None# 可选：图片的OSS URLthread_id:str# 会话线程ID，用于关联同一轮对话的所有消息

2.6app/agents/__init__.py

# app/agents/__init__.py# Agent模块初始化文件

2.7app/agents/personal_chief.py

这是整个应用最核心的文件，包含了 Agent 的完整逻辑：

# app/agents/personal_chief.pyfromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.messagesimportHumanMessage,AIMessageChunk,AIMessagefromlangchain_core.toolsimporttoolfromlangchain_tavilyimportTavilySearchfromlangchain.agentsimportcreate_agentfromapp.common.loggerimportloggerimportosfromlanggraph.checkpoint.sqliteimportSqliteSaverimportsqlite3# 加载环境变量fromdotenvimportload_dotenv load_dotenv()# web搜索工具，使用tavily作为web搜索工具# max_results=5：每次搜索最多返回5条结果# topic="general"：通用主题搜索tavily=TavilySearch(max_results=5,topic="general")# 多模态模型# 使用通义千问的qwen3-omni-flash（多模态模型）# model_provider="openai"：因为通义兼容OpenAI的接口协议，所以填openaimodel=init_chat_model(model="qwen3-omni-flash",# 模型名称，支持图片、文本、音频、视频多模态输入model_provider="openai",base_url=os.getenv("DASHSCOPE_BASE_URL"),api_key=os.getenv("DASHSCOPE_API_KEY"))# 初始化checkpointer# 连接SQLite数据库文件，用于持久化Agent的对话历史# check_same_thread=False：允许跨线程访问（uvicorn多worker时需要）checkpointer=SqliteSaver(sqlite3.connect("db/personal_chief.db",check_same_thread=False))# 自动建表checkpointer.setup()# Agent系统提示词# 这是给AI的"角色设定 + 操作流程"说明书# 提示词的质量直接决定Agent的行为是否符合预期system_prompt=""" 你是一名私人厨师。收到用户提供的食材照片或清单后，请按以下流程操作： 1.识别和评估食材：若用户提供照片，首先辨识所有可见食材。基于食材的外观状态，评估其新鲜度与可用量，整理出一份"当前可用食材清单"。 2.智能食谱检索：优先调用 web_search 工具，以"可用食材清单"为核心关键词，查找可行菜谱。 3.多维度评估与排序：从营养价值和制作难度两个维度对检索到的候选食谱进行量化打分，并根据得分排序，制作简单且营养丰富的排名靠前。 4.结构化方案输出：把排序后的食谱整理为一份结构清晰的建议报告，要包含食谱信息、得分、推荐理由、食谱的参考图片，帮助用户快速做出决策。 请严格按照流程，优先调用 web_search 工具搜索食谱，搜索不到的情况下才能自己发挥。 """# 创建代理# create_agent是LangChain的高级封装，内部基于LangGraph实现# 只需传入model、tools、prompt，框架自动构建推理循环agent=create_agent(model=model,# 多模态大模型（大脑）tools=[tavily],# 工具列表（手脚）checkpointer=checkpointer,# 记忆存储（笔记）system_prompt=system_prompt# 行为规范（手册）)# 流式对话asyncdefsearch_recipes(prompt:str,image:str,thread_id:str):""" 调用agent搜索食谱，支持流式输出（打字机效果）。 Args: prompt: 用户输入的文本 image: 可选的图片URL（来自OSS上传） thread_id: 会话线程ID Yields: str: AI响应的文本片段（用于SSE流式推送） """logger.info(f"[用户]:{prompt}, image:{image}, thread_id:{thread_id}")try:# 判断是否有图片，封装不同格式的消息# LangChain的HumanMessage支持多模态content（列表形式）ifnotimageorimage.strip()=="":# 纯文本场景：直接传字符串message=HumanMessage(content=prompt)else:# 多模态场景：content是list，包含image URL + textmessage=HumanMessage(content=[{"type":"image","url":image},# 图片URL（不传base64，省内存）{"type":"text","text":prompt}# 配合图片的文本问题])# 流式调用Agentforchunk,metadatainagent.stream({"messages":[message]},{"configurable":{"thread_id":thread_id}},# 注入thread_id，关联记忆stream_mode="messages"):ifisinstance(chunk,AIMessageChunk)andchunk.content:yieldchunk.contentexceptExceptionase:logger.error(f"\n[错误]:{str(e)}")yield"信息检索失败，试试看手动输入食物列表？"# 清空会话defclear_messages(thread_id:str):"""清空会话"""logger.info(f"清空历史消息，thread_id:{thread_id}")checkpointer.delete_thread(thread_id)# 查询会话历史defget_messages(thread_id:str)->list[dict[str,str]]:""" 获取会话历史。 LangChain的checkpoint中存储了完整的消息列表， 我们从中提取HumanMessage（用户）和AIMessage（AI）进行返回。 """logger.info(f"获取历史消息，thread_id:{thread_id}")# 根据thread_id查询checkpointcheckpoint=checkpointer.get({"configurable":{"thread_id":thread_id}})# 如果不存在，返回空列表ifnotcheckpoint:return[]# 安全获取messageschannel_values=checkpoint.get("channel_values")ifnotchannel_values:return[]messages=channel_values.get("messages",[])ifnotmessages:return[]# 转换消息格式result=[]formsginmessages:ifnotmsg.content:continueifisinstance(msg,HumanMessage):result.append({"role":"user","content":msg.content})elifisinstance(msg,AIMessage):result.append({"role":"assistant","content":msg.content})returnresult

2.8app/api/__init__.py

# app/api/__init__.py# API模块初始化文件

2.9app/api/v1/__init__.py

# app/api/v1/__init__.py# API v1版本初始化文件

2.10app/api/v1/chat.py

# app/api/v1/chat.pyfromfastapiimportAPIRouterfromapp.models.schemasimportChatRequestfromfastapi.responsesimportStreamingResponsefromapp.agents.personal_chiefimportsearch_recipes,get_messages,clear_messages router=APIRouter()@router.post("/chat/stream")asyncdefchat_endpoint(request:ChatRequest):"""流式聊天接口"""returnStreamingResponse(search_recipes(request.message,request.image_url,request.thread_id),media_type="text/event-stream")@router.get("/chat/messages")asyncdefget_chat_messages(thread_id:str):"""获取历史消息"""messages=get_messages(thread_id)return{"messages":messages}@router.delete("/chat/messages")asyncdefclear_chat_messages(thread_id:str):"""清空历史消息"""clear_messages(thread_id)return{"success":True}

2.11app/api/v1/oss.py

# app/api/v1/oss.pyimportalibabacloud_oss_v2asossfromfastapiimportAPIRouterfromdatetimeimporttimedeltaimportos# 加载环境变量fromdotenvimportload_dotenv load_dotenv()router=APIRouter()# 从环境变量中加载凭证信息，用于身份验证# 使用EnvironmentVariableCredentialsProvider，密钥不落地代码credentials_provider=oss.credentials.EnvironmentVariableCredentialsProvider()# 加载SDK的默认配置，并设置凭证提供者cfg=oss.config.load_default()cfg.credentials_provider=credentials_provider# 方式一：只填写Region（推荐）# 必须指定Region ID，SDK会根据Region自动构造HTTPS访问域名cfg.region='cn-heyuan'# 使用配置好的信息创建OSS客户端client=oss.Client(cfg)# OSS 域名配置OSS_ENDPOINT=os.getenv("OSS_ENDPOINT","oss-cn-heyuan.aliyuncs.com")OSS_BUCKET=os.getenv("OSS_BUCKET")@router.get("/oss/presign")defchat_endpoint(filename:str):""" 返回一个预签名的PUT请求URL。 前端拿这个URL，直接上传文件到OSS，不经过我们的服务器。 这样设计的好处： 1. 服务器不承担文件传输，省带宽 2. OSS原生支持HTTPS，安全 3. 签名URL有时效，过期无法上传，防滥用 """# 根据文件扩展名判断Content-Type，OSS上传需要准确类型content_type_map={"jpg":"image/jpeg","jpeg":"image/jpeg","png":"image/png","gif":"image/gif","webp":"image/webp",}ext=filename.split(".")[-1].lower()if"."infilenameelse"jpg"content_type=content_type_map.get(ext,"application/octet-stream")# 预签名URL，有效期1小时pre_result=client.presign(oss.PutObjectRequest(bucket=OSS_BUCKET,key=filename,content_type=content_type,),expires=timedelta(seconds=3600))# 返回上传URL和可访问的图片路径return{"uploadUrl":pre_result.url.strip('"'),# 前端用这个URL上传"contentType":content_type,"accessUrl":f"https://{OSS_BUCKET}.{OSS_ENDPOINT}/{filename}"# 上传后的访问地址}

2.12app/main.py

# app/main.pyimportosfromfastapiimportFastAPIfromfastapi.responsesimportFileResponsefromfastapi.staticfilesimportStaticFilesfromfastapi.middleware.corsimportCORSMiddlewarefromapp.api.v1importchatfromapp.api.v1importossfromapp.common.loggerimportsetup_logging# 初始化日志配置（最早执行，确保后续所有模块都能正确记录日志）setup_logging()# 创建FastAPI应用实例app=FastAPI(title="Personal Chief API",description="私厨",version="0.1.0")# 配置跨域资源共享 (CORS)# 因为我们的前端可能是独立部署的（Next.js静态导出），# 需要允许跨域请求，否则浏览器会阻止AJAX请求app.add_middleware(CORSMiddleware,allow_origins=["*"],# 生产环境建议指定具体域名allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)# 挂载路由# /api/v1/chat/* -> chat.py中的接口# /api/v1/oss/* -> oss.py中的接口app.include_router(chat.router,prefix="/api/v1",tags=["对话"])app.include_router(oss.router,prefix="/api/v1",tags=["申请上传签名url"])# 挂载前端资源（如果有static目录的话）static_dir=os.path.join(os.path.dirname(__file__),"static")ifos.path.exists(static_dir):app.mount("/",StaticFiles(directory=static_dir,html=True),name="static")# 前端fallback路由 - 只处理非API请求# Next.js是单页应用（SPA），所有路由都由index.html处理# FastAPI不知道前端有哪些路由，所以对未匹配的路径返回index.html@app.get("/{path:path}",include_in_schema=False)asyncdefserve_frontend(path:str):# 排除API路径ifpath.startswith("api/"):fromfastapi.responsesimportJSONResponsereturnJSONResponse({"error":"Not Found"},status_code=404)# 如果请求的是静态文件，直接返回file_path=os.path.join(static_dir,path)ifos.path.isfile(file_path):returnFileResponse(file_path)# 否则返回index.html（SPA fallback）index_path=os.path.join(static_dir,"index.html")ifos.path.exists(index_path):returnFileResponse(index_path)return{"message":"你的独家私厨上线了~","status":"ok"}if__name__=="__main__":importuvicorn# 启动命令：python -m app.mainuvicorn.run("app.main:app",host="127.0.0.1",port=8001,reload=True)

三、快速运行

3.1 创建所有文件

按照上面的项目结构，创建好所有目录和文件，将对应的代码复制进去。

3.2 启动服务

# 1. 安装依赖uvsync# 2. 启动服务python-mapp.main# 3. 打开浏览器访问# http://localhost:8001

四、运行效果演示

4.1 基础聊天界面

启动后访问http://localhost:8001，可以看到完整的聊天界面：

• 支持图片上传（点击上传按钮选择冰箱/厨房照片）
• 支持纯文本对话（直接输入食材列表）
• AI自动识别食材并搜索食谱

4.2 上传食材图片

上传一张冰箱图片后，AI 会自动识别食材：

4.3 智能食谱推荐

AI 调用 Tavily 搜索相关食谱，并按营养价值 + 制作难度打分排序：

4.4 多轮对话（Agent 记忆）

用户：我喜欢第1道菜，可以说得更详细点吗？ AI：当然可以！下面是"三文鱼西兰花烤盘料理"的详细步骤... （包含食材清单、详细步骤、小贴士、营养分析）

Agent 自动记住了之前的上下文，知道"第1道菜"指的是什么食谱。

4.5 LangSmith 调试界面

LangChain 提供了基于 LangSmith 的 GUI 控制台，可以方便地调试 Agent：

4.6 详细的调用过程追踪

五、总结与扩展

5.1 Agent 核心原理回顾

┌─────────────────────────────────────────────┐ │ Agent 架构 │ ├─────────────────────────────────────────────┤ │ ┌─────────┐ ┌──────────┐ ┌────────┐ │ │ │ Model │◄──►│ Tools │◄──►│ Memory │ │ │ │ (大脑) │ │ (手脚) │ │ (笔记) │ │ │ └────┬────┘ └──────────┘ └────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────┐ │ │ │ 推理循环 │ │ │ │ (Agent Loop)│ │ │ └─────────────┘ │ │ │ │ 1. 接收用户输入（文本 + 图片） │ │ 2. 模型理解意图，决定是否调用工具 │ │ 3. 执行工具（Tavily 搜索） │ │ 4. 将工具结果注入模型，生成最终回复 │ │ 5. 将本轮对话存入 Memory，供后续使用 │ └─────────────────────────────────────────────┘

5.2 扩展练习

扩展 1：添加更多工具

目前 Agent 只有 Web 搜索一个工具。可以尝试添加：

• 天气 API：根据当地天气推荐适合的菜品（如雨天推荐热汤）
• 营养计算器：接入营养数据库，计算每道菜的热量/碳水/蛋白质

扩展 2：支持语音输入

利用浏览器的Web Speech API采集语音，通过 ASR 转文字后发送给 Agent，实现"拍照 + 语音提问"的交互方式。

扩展 3：换用不同的模型

将qwen3-omni-flash替换为gpt-4o或claude-3-opus，只需修改model和base_url/api_key，提示词和工具代码无需改动——这就是 LangChain 接口统一抽象的好处。

推荐阅读

注意事项

1. API Key 安全：生产环境中务必将 Key 放在环境变量或密钥管理服务中，切勿硬编码到代码
2. OSS 权限：示例中将 Bucket 设置为"公共读"仅用于演示，生产环境应设为私有并通过 CDN 对外暴露
3. 流式输出兼容性：部分 HTTP 客户端（如某些版本requests）不支持 SSE，请使用浏览器 Fetch API 或 Postman 测试