langchain开发-AI私厨实战

AI私厨实战

一、agent开发

1.需求分析

整体架构如上

2.具体架构

后端 agent python 代码

load_dotenv()

# 多模态模型
multimodal_model = init_chat_model(
    model="qwen3.5-plus",  # 模型名称，这里选择qwen3.5-plus，这是一个多模态模型，支持图片、文本、音频、视频
    model_provider="openai",
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    api_key=os.getenv("DASHSCOPE_API_KEY")
)


# web搜索工具，使用tavily作为web搜索工具
web_search = TavilySearch(
    max_results=5,
    topic="general"
)

# 4.初始化checkpointer
# 连接sqlite
connection = sqlite3.connect("../db/personal_chief.db", check_same_thread=False)
# 初始化checkpointer
checkpointer = SqliteSaver(connection)
# 自动建表
checkpointer.setup()



system_prompt = """
你是一名私人厨师。收到用户提供的食材照片或清单后，请按以下流程操作：
1.识别和评估食材：若用户提供照片，首先辨识所有可见食材。基于食材的外观状态，评估其新鲜度与可用量，整理出一份“当前可用食材清单”。
2.智能食谱检索：优先调用 web_search 工具，以“可用食材清单”为核心关键词，查找可行菜谱。
3.多维度评估与排序：从营养价值和制作难度两个维度对检索到的候选食谱进行量化打分，并根据得分排序，制作简单且营养丰富的排名靠前。
4.结构化方案输出：把排序后的食谱整理为一份结构清晰的建议报告，要包含食谱信息、得分、推荐理由，帮助用户快速做出决策。

请严格按照流程，优先调用 web_search 工具搜索食谱，再搜索不到的情况下才能自己发挥。
"""

agent = create_agent(
    model=multimodal_model,
    tools=[web_search],
    checkpointer=checkpointer,
    system_prompt=system_prompt,
)

二、langsmith对接

1.langsmith

langsmith平台是官方原生

LangSmith 是 LangChain 团队推出的一个 LLM 应用可观测性与评估平台，主要用于调试、测试、监控 AI 应用

可以直接将自己本地·启动的agent对接到平台中

https://smith.langchain.com/

配置官方 apikey

LANGSMITH_API_KEY=lsv2_pt_ff002467960b400c9e4a293a8dd227
LANGSMITH_TRACING=true

LANGSMITH_PROJECT=lc-course

langsmith官方提供了

2.配置依赖

加载依赖

uv add langgraph-cli[inmem]

配置文件

/langgraph.json

{
    "dependencies": ["."],
    "graphs": {
        "chief_agent": "./app/agent/personal_chief.py:agent"  //这个是agnet对象入口
    },
    "env": ".env"
}

启动命令

uv run langgraph dev

接着就能在官网页面找到我们启动的agent

三、本地部署

1.架构

由于官网部署需要额外付费，所以一般都是配合fastapi提供接口，配合前端对外服务

通常再发给模型时，会将图片上传到图片服务器，接着提交url给大语言模型，这样会节省一些token

具体架构如下

2.开发

流式调用

from langchain.messages import AIMessageChunk

for chunk, metadata in agent.stream(
            {"messages": [multimodal_message]},
            config,
            stream_mode="messages"
        ):
            if isinstance(chunk, AIMessageChunk) and chunk.content:
                print(chunk.content, end="", flush=True)

只是返回工具调用的消息

会话历史

既然会话记忆交给了checkpointer管理，我们自然可以从中获取会话历史

for m in checkpointer.get(config)['channel_values']['messages']:
    print(type(m))
    print(m)

清空历史记录

# 清空会话
def clear_messages(thread_id: str):
    """清空会话"""
    logger.info(f"清空历史消息，thread_id: {thread_id}")
    checkpointer.delete_thread(thread_id)

然后启动main函数就可以啦