1、解决如果上传文件或图片节省token搭建服务端对接OSS1设置 apikey访问链接https://ram.console.aliyun.com/overview?activeTabworkflow进入RAM访问控制页面创建好后点开用户进行权限配置2、使用oss---创建桶访问链接https://oss.console.aliyun.com/overview进入oss控制台选择创建Bucket输入名字其余默认测试完后记得复原---实际开发不用关闭安全最后我们需要把阿里云OSS的API_KEY配置到环境变量中也就是项目的.env文件内格式如下# 阿里OSS OSS_ACCESS_KEY_ID OSS_ACCESS_KEY_SECRET OSS_BUCKETtmp95273、智能体的最终修改from langchain.chat_models import init_chat_model from langchain_tavily import TavilySearch from langchain.agents import create_agent import os from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 1.加载环境变量 from dotenv import load_dotenv load_dotenv() # 2.web搜索工具使用tavily作为web搜索工具 web_search TavilySearch( max_results5, topicgeneral ) # 3.多模态模型 model init_chat_model( modelqwen3.5-plus, # 模型名称这里选择qwen3.5-plus这是一个多模态模型支持图片、文本、音频、视频 model_provideropenai, base_urlos.getenv(DASHSCOPE_BASE_URL), api_keyos.getenv(DASHSCOPE_API_KEY) ) # 4.初始化checkpointer # 连接sqlite connection sqlite3.connect(../db/personal_chief.db, check_same_threadFalse) # 初始化checkpointer checkpointer SqliteSaver(connection) # 自动建表 checkpointer.setup() # 5.Agent系统提示词 system_prompt 你是一名私人厨师。收到用户提供的食材照片或清单后请按以下流程操作 1.识别和评估食材若用户提供照片首先辨识所有可见食材。基于食材的外观状态评估其新鲜度与可用量整理出一份“当前可用食材清单”。 2.智能食谱检索优先调用 web_search 工具以“可用食材清单”为核心关键词查找可行菜谱。 3.多维度评估与排序从营养价值和制作难度两个维度对检索到的候选食谱进行量化打分并根据得分排序制作简单且营养丰富的排名靠前。 4.结构化方案输出把排序后的食谱整理为一份结构清晰的建议报告要包含食谱信息、得分、推荐理由、食谱的参考图片帮助用户快速做出决策。 请严格按照流程优先调用 web_search 工具搜索食谱搜索不到的情况下才能自己发挥。 # 6.创建Agent agent create_agent( modelmodel, # 模型 tools[web_search], # 工具 checkpointercheckpointer, # 记忆 system_promptsystem_prompt # 系统提示词 ) # 流式对话 async def search_recipes(prompt: str, image: str, thread_id: str): 调用agent搜索食谱 logger.info(f[用户]: {prompt}, image: {image}, thread_id: {thread_id}) try: # 判断是否有图片封装不同格式的消息 if not image or image.strip() : message HumanMessage(contentprompt) else: message HumanMessage(content[ {type: image, url: image}, {type: text, text: prompt} ]) # 流式调用Agent for chunk, metadata in agent.stream( {messages: [message]}, {configurable: {thread_id: thread_id}}, stream_modemessages ): if isinstance(chunk, AIMessageChunk) and chunk.content: yield chunk.content except Exception as e: logger.error(f\n[错误]: {str(e)}) yield 信息检索失败试试看手动输入食物列表 # 清空会话 def clear_messages(thread_id: str): 清空会话 logger.info(f清空历史消息thread_id: {thread_id}) checkpointer.delete_thread(thread_id) # 查询会话历史 def get_messages(thread_id: str) - list[dict[str, str]]: 获取会话历史 logger.info(f获取历史消息thread_id: {thread_id}) # 根据 thread_id 查询 checkpoint checkpoint checkpointer.get({configurable: {thread_id: thread_id}}) # 如果不存在返回空列表 if not checkpoint: return [] # 安全获取 messages channel_values checkpoint.get(channel_values) if not channel_values: return [] messages channel_values.get(messages, []) if not messages: return [] # 转换消息格式 result [] for msg in messages: if not msg.content: continue if isinstance(msg, HumanMessage): result.append({role: user, content: msg.content}) elif isinstance(msg, AIMessage): result.append({role: assistant, content: msg.content}) return result4、编写FastAPIfrom fastapi import APIRouter from app.models.schemas import ChatRequest from fastapi.responses import StreamingResponse # StreamingResponse 异步调用 from app.agents.personal_chief import search_recipes, get_messages, clear_messages router APIRouter() # 流式就是智能体返回一点就输出一点而不是全部返回才输出更加美观些 router.post(/chat/stream) async def chat_endpoint(request: ChatRequest): 流式对话 return StreamingResponse( search_recipes(request.message, request.image_url, request.thread_id), media_typetext/event-stream ) router.get(/chat/messages) async def get_chat_messages(thread_id: str): 获取历史消息 messages get_messages(thread_id) return {messages: messages} router.delete(/chat/messages) async def clear_chat_messages(thread_id: str): 清空历史消息 clear_messages(thread_id) return {success: True}
