diff --git a/api/agent_api.py b/api/agent_api.py index a5970b9..a3a4e15 100644 --- a/api/agent_api.py +++ b/api/agent_api.py @@ -38,15 +38,10 @@ def chat(request: ChatRequest): @agent_router.post("/chat_stream") def chat(request: ChatRequest): - def generate_stream(): - result = ChefService.chat_stream( + return StreamingResponse( + ChefService.chat_stream( thread_id=request.thread_id, message=request.message, - ) - for chunk in result: - yield chunk - - return StreamingResponse( - generate_stream(), + ), media_type="text/plain; charset=utf-8" ) \ No newline at end of file diff --git a/api/scheduler_api.py b/api/scheduler_api.py index 32c02be..9b803e9 100644 --- a/api/scheduler_api.py +++ b/api/scheduler_api.py @@ -1,7 +1,5 @@ from fastapi import APIRouter, HTTPException from typing import Any, Dict -from core.db import SessionDep -from core.dto.api_response import ApiResponse from core.task.dlt_scheduler import scheduler # 彩票相关的路由 diff --git a/core/config/logging_cof.py b/core/config/logging_cof.py new file mode 100644 index 0000000..d8f4182 --- /dev/null +++ b/core/config/logging_cof.py @@ -0,0 +1,24 @@ +import logging +import sys + +import structlog + + +def setup_logging(): + logging.basicConfig( + format='%(message)s', + stream=sys.stdout, + level=logging.INFO, + ) + + structlog.configure( + processors=[ + structlog.contextvars.merge_contextvars, + structlog.processors.add_log_level, + structlog.processors.TimeStamper(fmt='iso'), + # structlog.processors.JSONRenderer(ensure_ascii=False), # 生产输出JSON 开发生产2选一 + structlog.dev.ConsoleRenderer() # 开发环境彩色控制台 + ], + logger_factory=structlog.stdlib.LoggerFactory(), + cache_logger_on_first_use=True, + ) diff --git a/core/task/dlt_scheduler.py b/core/task/dlt_scheduler.py index 1f6f8ec..50cdd50 100644 --- a/core/task/dlt_scheduler.py +++ b/core/task/dlt_scheduler.py @@ -1,30 +1,16 @@ -import logging -import sys -from pathlib import Path from datetime import datetime from typing import List +import structlog from apscheduler.job import Job from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.base import STATE_PAUSED, STATE_STOPPED, STATE_RUNNING from apscheduler.triggers.cron import CronTrigger from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR -# 添加项目根目录到路径 -sys.path.append(str(Path(__file__).parent.parent.parent)) - from core.task.dlt_spider import DLTSpider -# 配置日志 -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', - handlers=[ - logging.FileHandler('lottery_scheduler.log', encoding='utf-8'), - logging.StreamHandler() - ] -) -logger = logging.getLogger(__name__) +logger = structlog.getLogger(__name__) def test_fun(): logger.info(f"当前时间---{datetime.now()}") diff --git a/core/task/dlt_spider.py b/core/task/dlt_spider.py index 8e92ab0..7cbc72e 100644 --- a/core/task/dlt_spider.py +++ b/core/task/dlt_spider.py @@ -1,17 +1,16 @@ -import logging from dataclasses import dataclass from datetime import datetime from typing import Any, List, Dict, Union, Optional, NamedTuple import time import requests -from pydantic import BaseModel, field_validator +import structlog from sqlalchemy import desc from core.db import get_db_session from models import LotteryDLT -logger = logging.getLogger(__name__) +logger = structlog.getLogger(__name__) def parse_draw_numbers(result_str: str) -> List[str]: diff --git a/exceptions/__init__.py b/exceptions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/exceptions/errors.py b/exceptions/errors.py new file mode 100644 index 0000000..aa75e5e --- /dev/null +++ b/exceptions/errors.py @@ -0,0 +1,3 @@ +class AgentNotReadyError(RuntimeError): + """agent未初始化错误""" + pass \ No newline at end of file diff --git a/exceptions/handlers.py b/exceptions/handlers.py new file mode 100644 index 0000000..66086db --- /dev/null +++ b/exceptions/handlers.py @@ -0,0 +1,24 @@ +from fastapi import Request, FastAPI +from fastapi.responses import JSONResponse +from exceptions.errors import AgentNotReadyError + + +async def agent_not_ready_handler(request: Request, exc: Exception): + """agent未就绪的错误""" + return JSONResponse( + status_code=503, + content={ + "code": 503, + "message": str(exc) + } + ) + + +# 集中处理 保持main简洁 +def register_exception_handlers(app: FastAPI): + """集中处理注册异常""" + + app.add_exception_handler( + AgentNotReadyError, + agent_not_ready_handler, + ) diff --git a/lc/chef_agent.py b/lc/chef_agent.py index c6555a4..8c44f06 100644 --- a/lc/chef_agent.py +++ b/lc/chef_agent.py @@ -1,8 +1,8 @@ from __future__ import annotations -import logging 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 @@ -14,8 +14,10 @@ from psycopg import Connection from psycopg_pool import ConnectionPool from core.config.settings import settings +from exceptions.errors import AgentNotReadyError + +logger = structlog.getLogger(__name__) -logger = logging.getLogger(__name__) class ChefAgent: """厨师 Agent。 @@ -30,7 +32,7 @@ class ChefAgent: """ def __init__(self) -> None: - self._pool: ConnectionPool[Connection[Any]] | None = None + self._pool = None self._checkpointer: PostgresSaver | None = None self._agent = None @@ -81,7 +83,8 @@ class ChefAgent: llm = init_chat_model( model="deepseek-v4-flash", - api_key=settings.DEEPSEEK_API_KEY, + model_provider="deepseek", + api_key=settings.DEEPSEEK_API_KEY ) logger.info("chef_agent的llm初始化完成") @@ -122,16 +125,16 @@ class ChefAgent: logger.info("chef_agent的CheckPointer开始初始化") logger.info("创建PostgreSQL连接池") - self.pool: ConnectionPool[Connection[Any]] = ConnectionPool( + 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) + checkpointer = PostgresSaver(self._pool) logger.info("初始化checkpoint数据表") # 创建 LangGraph checkpoint 所需的数据表 @@ -223,7 +226,8 @@ class ChefAgent: """获取 Agent。""" if self._agent is None: - raise RuntimeError( + logger.info("ChefAgent 尚未初始化,请先调用 initialize()") + raise AgentNotReadyError( "ChefAgent 尚未初始化,请先调用 initialize()" ) @@ -235,6 +239,7 @@ class ChefAgent: if self._pool is not None: self._pool.close() self._pool = None + logger.info("PostgreSQL已关闭") self._checkpointer = None self._agent = None diff --git a/main.py b/main.py index d9efddc..4f1c52c 100644 --- a/main.py +++ b/main.py @@ -1,4 +1,4 @@ -import logging +import structlog import models from contextlib import asynccontextmanager @@ -7,20 +7,18 @@ from fastapi.middleware.cors import CORSMiddleware from sqlmodel import SQLModel from api.agent_api import agent_router +from core.config.logging_cof import setup_logging from core.db import engine from core.config.settings import settings from api.scheduler_api import scheduler_router from api.lottery_api import lottery_router from core.task.dlt_scheduler import scheduler +from exceptions.handlers import register_exception_handlers from lc.chef_agent import chef_agent -# 配置日志 -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) - -logger = logging.getLogger(__name__) +# 全局设置日志配置 +setup_logging() +logger = structlog.getLogger(__name__) # lifespan生命周期 @asynccontextmanager @@ -30,23 +28,34 @@ async def lifespan(app: FastAPI): # 同步建表 SQLModel.metadata.create_all(engine) + logger.info("服务启动,启动定时调度器") # 启动定时调度器 scheduler.start() - # 初始化chef_agent + logger.info("服务启动,创建携程初始化chef_agent任务") + # 异步初始化chef_agent chef_agent.initialize() - yield # 此处交出控制权,服务开始运行 - # ========== 关闭时执行 ========== - logger.info("服务关闭,释放资源") - chef_agent.close() - scheduler.shutdown() + logger.info("服务启动完成") + try: + yield # 此处交出控制权,服务开始运行 + finally: + # ========== 关闭时执行 ========== + logger.info("服务关闭,释放资源") + chef_agent.close() + scheduler.shutdown() app = FastAPI(lifespan=lifespan) + +# 注册全局异常处理 +register_exception_handlers(app) + +# 注册路由 app.include_router(scheduler_router) app.include_router(lottery_router) app.include_router(agent_router) +# 注册中间件 app.add_middleware( CORSMiddleware, # type: ignore[arg-type] allow_origins=settings.ALLOWED_ORIGINS, diff --git a/pyproject.toml b/pyproject.toml index 3fe9317..101da80 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,5 +25,6 @@ dependencies = [ "sqlalchemy>=2.0.52", "sqlmodel==0.0.42", "starlette>=1.6.0", + "structlog>=26.1.0", "uvicorn==0.52.4", ] diff --git a/services/agent_server/chef_service.py b/services/agent_server/chef_service.py index 308a62d..13e0a15 100644 --- a/services/agent_server/chef_service.py +++ b/services/agent_server/chef_service.py @@ -1,3 +1,5 @@ +from typing import Iterator + from langchain_core.messages import HumanMessage from langchain_core.runnables import RunnableConfig @@ -17,7 +19,11 @@ class ChefService: return result @staticmethod - def chat_stream(thread_id: str, message: str): + def chat_stream(thread_id: str, message: str) -> Iterator[str]: config: RunnableConfig = {"configurable": {"thread_id": thread_id}} result = chef_agent.agent.stream({"messages": [("user", message)]}, config=config, stream_mode="messages") - return result \ No newline at end of file + + for message_chunk, metadata in result: + content = message_chunk.content + if content: + yield content diff --git a/test.py b/test.py index 7a8c24a..79eb8cc 100644 --- a/test.py +++ b/test.py @@ -1,11 +1,20 @@ -sys_prompt = """ -你是一名有名的厨师。收到用户提供的食材照片或清单后,按照以下流程步骤操作: -1.识别和评估食材:若用户提供照片,首先辨别所有可见食材,基于食材的外观状态,评估其新鲜度和可用量,整理出一份“可用食材清单”。 -2.智能食谱检索:优先调用搜索工具,以“可用食材清单”为核心关键词,查找可行菜谱。 -3.多维度评估与排序:从营养与烹饪难度这2个维度对检索到的候选食谱进行量化打分,并根据得分进行排序,制作简单且营养丰富的排名靠前。 -4.结构化方案输出:把排序后的食谱整理成一份结构清晰的建议报告,要包括食谱信息、得分、推荐理由、食谱的参考图片,帮助用户快速做出决策。 -严格按照流程进行操作,搜索不到再自己发挥。 -""" +a: int | None = None -if __name__ == "__main__": - print(sys_prompt) \ No newline at end of file +def f(v: int) -> int: + pass + +def f2(v: int) -> int: + a = 5 + +f(a) + +class Test: + def __init__(self): + self.a: int | None = None + + def f(self, v: int) -> int: + pass + + def f2(self): + self.a = 5 + self.f(self.a) \ No newline at end of file diff --git a/uv.lock b/uv.lock index bb3572b..9ff093b 100644 --- a/uv.lock +++ b/uv.lock @@ -1428,6 +1428,7 @@ dependencies = [ { name = "sqlalchemy" }, { name = "sqlmodel" }, { name = "starlette" }, + { name = "structlog" }, { name = "uvicorn" }, ] @@ -1455,6 +1456,7 @@ requires-dist = [ { name = "sqlalchemy", specifier = ">=2.0.52" }, { name = "sqlmodel", specifier = "==0.0.42" }, { name = "starlette", specifier = ">=1.6.0" }, + { name = "structlog", specifier = ">=26.1.0" }, { name = "uvicorn", specifier = "==0.52.4" }, ] @@ -2638,6 +2640,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c8/cb/6a6a47d5b464bd08695d254f3da6e7986cc70c9fa5d778eda57538edfe56/starlette-1.6.0-py3-none-any.whl", hash = "sha256:a86dd39d14bb45f85a3d18525215a9ef0cfd1f192ac793220e72598c90335f0c", size = 75969, upload-time = "2026-08-08T18:27:56.196Z" }, ] +[[package]] +name = "structlog" +version = "26.1.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5e/89/b4a0bcfdf4f71a3dea31379f095929613d7e4528a0996bca6aa964cd0dca/structlog-26.1.0.tar.gz", hash = "sha256:f63a716cbd1b1291cf7661de7794b455acfa4c43c5bcf1630e6ad5ddc1adb3b7", size = 1459881, upload-time = "2026-06-06T07:33:39.348Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a9/18/489c97b834dfff9cf2fc2507cede4bcd4b11e67f84bc462acd1992496f86/structlog-26.1.0-py3-none-any.whl", hash = "sha256:e081a26d6c373e6d201eca24eede26d8ffab07f88f477822e679183428d3d91e", size = 73764, upload-time = "2026-06-06T07:33:38.046Z" }, +] + [[package]] name = "tenacity" version = "9.1.4"