256 lines
7.1 KiB
Python
256 lines
7.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, cast
|
|
|
|
import structlog
|
|
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_tavily import TavilySearch
|
|
from langgraph.checkpoint.postgres import PostgresSaver
|
|
from psycopg import Connection
|
|
from psycopg_pool import ConnectionPool
|
|
|
|
from core.config.settings import settings
|
|
from exceptions.errors import AgentNotReadyError
|
|
|
|
logger = structlog.getLogger(__name__)
|
|
|
|
|
|
class ChefAgent:
|
|
"""厨师 Agent。
|
|
|
|
负责：
|
|
- LLM 初始化
|
|
- Tool 初始化
|
|
- CheckPointer 初始化
|
|
- Middleware 初始化
|
|
- Agent 初始化
|
|
- 生命周期管理
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._pool = None
|
|
self._checkpointer: PostgresSaver | None = None
|
|
self._agent = None
|
|
|
|
def initialize(self) -> ChefAgent:
|
|
"""初始化 Agent。
|
|
|
|
Returns:
|
|
当前 Agent 实例，方便链式调用。
|
|
"""
|
|
|
|
logger.info("开始初始化chef_agent...")
|
|
|
|
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",
|
|
model_provider="deepseek",
|
|
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[dict[str, Any]]]( # 指定类型[Connection[dict[str, Any]]]防止编译器提示错误
|
|
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:
|
|
logger.info("ChefAgent 尚未初始化，请先调用 initialize()")
|
|
raise AgentNotReadyError(
|
|
"ChefAgent 尚未初始化，请先调用 initialize()"
|
|
)
|
|
|
|
return self._agent
|
|
|
|
def close(self) -> None:
|
|
"""释放资源。"""
|
|
|
|
if self._pool is not None:
|
|
self._pool.close()
|
|
self._pool = None
|
|
logger.info("PostgreSQL已关闭")
|
|
|
|
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()
|