chef_agent

lc/chef_agent.py

@@ -1,71 +1,250 @@
-import os
-import sqlite3
-from typing import cast, Optional
+from __future__ import annotations
 
-from langchain.agents import create_agent, AgentState
+import logging
+from typing import Any, cast
+
+from langchain.agents import create_agent
 from langchain.agents.middleware import SummarizationMiddleware
 from langchain.agents.middleware.summarization import ContextMessages
 from langchain.chat_models import init_chat_model
 from langchain_core.language_models import BaseChatModel
-from langchain_core.messages import HumanMessage
-from langchain_core.runnables import RunnableConfig
-from langgraph.checkpoint.sqlite import SqliteSaver # 需要安装langgraph-checkpoint-sqlite
 from langchain_tavily import TavilySearch
+from langgraph.checkpoint.postgres import PostgresSaver
+from psycopg import Connection
+from psycopg_pool import ConnectionPool
 
 from core.config.settings import settings
 
-# 初始化模型 需要多模态大模型 deepseek已支持
-llm_model = init_chat_model(
-    model="deepseek-v4-flash",
-    model_provider="deepseek",
-    api_key=settings.DEEPSEEK_API_KEY
-)
+logger = logging.getLogger(__name__)
 
-# 定义工具 使用tavily搜索工具 langchain-tavily
-search_tool = TavilySearch(
-    tavily_api_key=settings.TAVILY_API_KEY,
-    max_results=5, # 最大搜索结果条数
-    topic="general"
-)
+class ChefAgent:
+    """厨师 Agent。
 
-# 定义记忆策略 使用SummarizationMiddleware摘要策略中间件
-summary = SummarizationMiddleware(
-    model=cast(BaseChatModel, llm_model),  # 消息摘要的记忆管理策略的模型
-    trigger=cast(ContextMessages, ("messages", 10)),  # 触发策略的条件
-    keep=cast(ContextMessages, ("messages", 5))  # 触发策略后保留的消息条数
-)
+    负责：
+    - LLM 初始化
+    - Tool 初始化
+    - CheckPointer 初始化
+    - Middleware 初始化
+    - Agent 初始化
+    - 生命周期管理
+    """
 
-# 创建数据库目录（如果不存在）
-os.makedirs("sqlite", exist_ok=True)
-# 创建 SQLite 数据库连接 check_same_thread=False 是为了确保在多线程环境下的安全性
-conn = sqlite3.connect("sqlite/checkpoints.db", check_same_thread=False)
-# 初始化 checkpointer
-checkpointer = SqliteSaver(conn)
-# 自动建表
-checkpointer.setup()
+    def __init__(self) -> None:
+        self._pool: ConnectionPool[Connection[Any]] | None = None
+        self._checkpointer: PostgresSaver | None = None
+        self._agent = None
 
-# agent提示词
-system_prompt = """
-你是一名有名的国宴厨师。收到用户提供的食材照片或清单后，按照以下流程步骤操作：
-1.识别和评估食材：若用户提供照片，首先辨别所有可见食材，基于食材的外观状态，评估其新鲜度和可用量，整理出一份“可用食材清单”。
-2.智能食谱检索：优先调用search_tool工具，以“可用食材清单”为核心关键词，查找可行菜谱。
-3.多维度评估与排序：从营养与烹饪难度这2个维度对检索到的候选食谱进行量化打分，并根据得分进行排序，制作简单且营养丰富的排名靠前。
-4.结构化方案输出：把排序后的食谱整理成一份结构清晰的建议报告，要包括食谱信息、得分、推荐理由、食谱的参考图片，帮助用户快速做出决策。
-"""
+    def initialize(self) -> ChefAgent:
+        """初始化 Agent。
 
-# 定义thread_config 用于记忆存储分组
-thread_config: RunnableConfig  = {"configurable": {"thread_id": "2"}}
+        Returns:
+            当前 Agent 实例，方便链式调用。
+        """
 
-# 创建agent
-agent = create_agent(
-    model=cast(BaseChatModel, llm_model), # 指定类型 避免idea报错
-    tools=[search_tool],
-    checkpointer=checkpointer,
-    middleware=[summary],
-    system_prompt=system_prompt
-)
+        logger.info("开始初始化chef_agent...")
 
-if __name__ == "__main__":
-    question = input("> ")
-    res = agent.invoke({"messages": [HumanMessage(content=question)]}, thread_config)
-    print(res)
+        if self._agent is not None:
+            return self
+
+        # 1. 初始化 LLM
+        llm = self._create_llm()
+
+        # 2. 初始化工具
+        tools = self._create_tools()
+
+        # 3. 初始化 Middleware
+        middleware = self._create_middleware(llm)
+
+        # 4. 初始化 CheckPointer
+        self._checkpointer = self._create_checkpointer()
+
+        # 5. 初始化 Agent
+        logger.info("chef_agent的agent开始初始化...")
+        self._agent = create_agent(
+            model=llm,
+            tools=tools,
+            middleware=middleware,
+            checkpointer=self._checkpointer,
+            system_prompt=self._system_prompt(),
+        )
+        logger.info("chef_agent的agent初始化完成")
+
+        logger.info("chef_agent初始化完成")
+
+        return self
+
+    @staticmethod
+    def _create_llm() -> BaseChatModel:
+        """创建 LLM。"""
+
+        logger.info("chef_agent初始化llm...")
+
+        llm = init_chat_model(
+            model="deepseek-v4-flash",
+            api_key=settings.DEEPSEEK_API_KEY,
+        )
+
+        logger.info("chef_agent的llm初始化完成")
+        return cast(BaseChatModel, llm)
+
+    @staticmethod
+    def _create_tools() -> list[Any]:
+        """创建 Agent Tools。"""
+
+        logger.info("chef_agent工具开始初始化...")
+        search_tool = TavilySearch(
+            tavily_api_key=settings.TAVILY_API_KEY,
+            max_results=5,
+            topic="general",
+        )
+
+        logger.info("chef_agent工具初始化完成")
+        return [search_tool]
+
+    @staticmethod
+    def _create_middleware(
+        llm: BaseChatModel,
+    ) -> list[Any]:
+        """创建 Agent Middleware。"""
+
+        logger.info("chef_agent中间件开始初始化...")
+        summarization = SummarizationMiddleware(
+            model=llm,
+            trigger=cast(ContextMessages, ("messages", 10)),
+            keep=cast(ContextMessages, ("messages", 5)),
+        )
+
+        logger.info("chef_agent中间件初始化完成")
+        return [summarization]
+
+    def _create_checkpointer(self) -> PostgresSaver:
+        """创建 PostgreSQL CheckPointer。"""
+
+        logger.info("chef_agent的CheckPointer开始初始化")
+        logger.info("创建PostgreSQL连接池")
+        self.pool: ConnectionPool[Connection[Any]] = ConnectionPool(
+            conninfo=settings.postgres_url,
+            max_size=10,
+            kwargs={
+                "autocommit": True,
+            },
+        )
+
+        logger.info("创建checkpointer")
+        checkpointer = PostgresSaver(self.pool)
+
+        logger.info("初始化checkpoint数据表")
+        # 创建 LangGraph checkpoint 所需的数据表
+        checkpointer.setup()
+
+        logger.info("chef_agent的CheckPointer初始化完成")
+        return checkpointer
+
+    @staticmethod
+    def _system_prompt() -> str:
+        """Agent System Prompt。"""
+
+        return """
+        你是一名专业厨师。
+        
+        你的任务是根据用户提供的食材照片或食材清单，为用户推荐合适的菜谱。
+        
+        请严格按照以下流程执行：
+        
+        ## 1. 识别和评估食材
+        
+        如果用户提供的是食材照片：
+        
+        - 识别照片中可见的食材
+        - 根据外观判断食材的新鲜程度
+        - 估算大致可用量
+        - 排除明显不可食用或状态异常的食材
+        
+        整理成「可用食材清单」。
+        
+        如果用户直接提供食材清单，则直接使用用户提供的信息。
+        
+        ## 2. 搜索菜谱
+        
+        优先使用搜索工具。
+        
+        以「可用食材清单」作为核心关键词，搜索适合这些食材的菜谱。
+        
+        优先考虑：
+        
+        - 食材匹配度高
+        - 操作简单
+        - 营养均衡
+        - 家庭烹饪可执行
+        
+        除非搜索不到合适结果，否则不要直接凭经验编造菜谱。
+        
+        ## 3. 评估和排序
+        
+        对搜索到的候选菜谱进行综合评价。
+        
+        从以下维度进行评分：
+        
+        - 食材匹配度
+        - 营养价值
+        - 烹饪难度
+        - 烹饪时间
+        
+        综合评分后进行排序。
+        
+        优先推荐：
+        
+        「简单 + 食材匹配度高 + 营养丰富」
+        
+        的菜谱。
+        
+        ## 4. 输出结果
+        
+        最终输出结构化的菜谱推荐报告。
+        
+        每个推荐至少包含：
+        
+        - 菜谱名称
+        - 所需食材
+        - 核心烹饪步骤
+        - 烹饪时间
+        - 难度
+        - 综合评分
+        - 推荐理由
+        - 参考来源
+        
+        如果搜索结果中存在可靠的菜谱图片，则提供图片参考。
+        
+        如果搜索不到合适菜谱，再根据已有知识进行合理推荐，并明确说明这是基于模型知识给出的建议。
+        """.strip()
+
+    @property
+    def agent(self):
+        """获取 Agent。"""
+
+        if self._agent is None:
+            raise RuntimeError(
+                "ChefAgent 尚未初始化，请先调用 initialize()"
+            )
+
+        return self._agent
+
+    def close(self) -> None:
+        """释放资源。"""
+
+        if self._pool is not None:
+            self._pool.close()
+            self._pool = None
+
+        self._checkpointer = None
+        self._agent = None
+
+    # 使用with上下文管理方式来使用的话需要这2个方法
+    # def __enter__(self) -> ChefAgent:
+    #     return self.initialize()
+    #
+    # def __exit__(self, exc_type, exc_value, traceback) -> None:
+    #     self.close()
+
+
+chef_agent = ChefAgent()