From 437fcd74b1caa0a843403db7b72b288b94732254 Mon Sep 17 00:00:00 2001 From: Marcus <1922576605@qq.com> Date: Mon, 28 Sep 2026 21:13:12 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=BC=95=E5=85=A5=20Loguru?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/exceptions.py | 11 ++++--- app/core/logging.py | 62 ++++++++++++++++++++++++++++++++++++ app/main.py | 8 ++++- app/observability/tracing.py | 13 ++++---- app/processor/base.py | 28 +++++++--------- app/services/agent.py | 23 +++++++++++-- pyproject.toml | 1 + uv.lock | 24 ++++++++++++++ 8 files changed, 139 insertions(+), 31 deletions(-) create mode 100644 app/core/logging.py diff --git a/app/core/exceptions.py b/app/core/exceptions.py index b38bb17..e7e7f48 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -1,19 +1,17 @@ """业务异常与全局 API 异常处理""" -import logging from http import HTTPStatus from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse +from loguru import logger from openai import APIError, APITimeoutError from starlette.exceptions import HTTPException from app.common.response import error_content from app.exceptions import AppError, DecisionError, ImportProcessError -logger = logging.getLogger(__name__) - class BizException(AppError): """可安全返回给调用方的业务错误""" @@ -54,7 +52,12 @@ async def handle_unexpected_exception( _request: Request, exc: Exception ) -> JSONResponse: # 详细异常仅记录在服务端，不将堆栈或内部错误内容返回给前端 - logger.error("未处理的服务器异常", exc_info=(type(exc), exc, exc.__traceback__)) + logger.bind(thread_id=getattr(exc, "_agent_thread_id", "-")).opt( + exception=exc + ).error("未处理的服务器异常") + + exc._logged_by_api = True + # 节点包装异常时沿原始原因查找，区分上游服务故障与内部错误 cause = exc while isinstance(cause, ImportProcessError): diff --git a/app/core/logging.py b/app/core/logging.py new file mode 100644 index 0000000..4fa9506 --- /dev/null +++ b/app/core/logging.py @@ -0,0 +1,62 @@ +"""统一配置应用、Uvicorn 和第三方库日志""" + +import logging +import sys +from pathlib import Path + +from loguru import logger + +LOG_PATH = Path(__file__).resolve().parents[2] / "logs" / "app.log" +LOG_FORMAT = "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {extra[thread_id]} | {extra[node]} | {message}" +_initialized = False + + +class InterceptHandler(logging.Handler): + """将标准 logging 转发到 Loguru，兼容第三方库""" + + def emit(self, record: logging.LogRecord) -> None: + # FastAPI 已记录的异常不再由 Uvicorn 重复打印堆栈 + if record.exc_info and getattr(record.exc_info[1], "_logged_by_api", False): + return + try: + level = logger.level(record.levelname).name + except ValueError: + level = record.levelno + logger.opt(exception=record.exc_info).log(level, record.getMessage()) + + +def setup_logging() -> None: + """应用启动或独立测试入口调用，多次调用不会重复添加输出""" + global _initialized + if _initialized: + return + + LOG_PATH.parent.mkdir(parents=True, exist_ok=True) + logger.remove() + logger.configure(extra={"thread_id": "-", "node": "-"}) + + options = { + "level": "INFO", + "format": LOG_FORMAT, + "enqueue": True, + "diagnose": False, + "backtrace": False, + } + + logger.add(sys.stderr, colorize=True, **options) + logger.add( + LOG_PATH, + rotation="20 MB", + retention="14 days", + compression="zip", + encoding="utf-8", + **options, + ) + + # 清理已有输出处理器，让标准日志只经过根处理器输出一次 + logging.basicConfig(handlers=[InterceptHandler()], level=logging.INFO, force=True) + for name in list(logging.root.manager.loggerDict): + existing = logging.getLogger(name) + existing.handlers.clear() + existing.propagate = True + _initialized = True diff --git a/app/main.py b/app/main.py index ce98ca1..18b2db2 100644 --- a/app/main.py +++ b/app/main.py @@ -4,13 +4,16 @@ import asyncio from contextlib import asynccontextmanager from fastapi import FastAPI +from loguru import logger from app.api.agent import router from app.core.exceptions import register_exception_handlers +from app.core.logging import setup_logging @asynccontextmanager async def lifespan(application: FastAPI): + setup_logging() from app.clients.decision import close_decision_engine from app.observability.tracing import flush_tracing from app.processor.graph import build_graph @@ -24,7 +27,10 @@ async def lifespan(application: FastAPI): try: await close_decision_engine() finally: - await asyncio.to_thread(flush_tracing) + try: + await asyncio.to_thread(flush_tracing) + finally: + await logger.complete() def create_app() -> FastAPI: diff --git a/app/observability/tracing.py b/app/observability/tracing.py index addd611..a34649b 100644 --- a/app/observability/tracing.py +++ b/app/observability/tracing.py @@ -6,16 +6,15 @@ from __future__ import annotations -import logging import os from asyncio import CancelledError from collections.abc import Iterator from contextlib import contextmanager from typing import Any -from app.config import settings +from loguru import logger -logger = logging.getLogger("app.observability") +from app.config import settings # 面板 Trace Name 列用的稳定名字，不要拼 thread_id 或用户原文 TRACE_NAME = "screen-instruct" @@ -52,8 +51,8 @@ def flush_tracing() -> None: return try: client.flush() - except Exception: - logger.warning("Langfuse flush 失败", exc_info=True) + except Exception as exc: # noqa: BLE001 追踪故障不阻断业务 + logger.opt(exception=exc).warning("Langfuse flush 失败") async def ainvoke_screen( @@ -232,9 +231,9 @@ def _ensure_client() -> Any: try: client = Langfuse() - except Exception: + except Exception as exc: # noqa: BLE001 追踪故障不阻断业务 _unavailable = True - logger.warning("Langfuse 不可用，本进程跳过追踪", exc_info=True) + logger.opt(exception=exc).warning("Langfuse 不可用，本进程跳过追踪") return None _client = client return client diff --git a/app/processor/base.py b/app/processor/base.py index 719f241..7706a8c 100644 --- a/app/processor/base.py +++ b/app/processor/base.py @@ -1,10 +1,10 @@ import inspect -import logging import time from abc import ABC, abstractmethod from typing import TYPE_CHECKING, TypeVar from langgraph.errors import GraphInterrupt +from loguru import logger from pydantic import ValidationError if TYPE_CHECKING: @@ -20,7 +20,7 @@ class BaseNode(ABC): def __init__(self, config: "Settings | None" = None): """保存显式配置；全局配置在节点实际使用时才加载。""" self._config = config - self.logger = logging.getLogger(f"import.{self.name}") + self.logger = logger.bind(node=self.name) @property def config(self) -> "Settings": @@ -35,24 +35,22 @@ class BaseNode(ABC): if inspect.iscoroutinefunction(self.process): return self._acall(state) start_time = time.perf_counter() - self.logger.info("--- %s 开始 ---", self.name) + self.logger.info("节点开始") try: result = self.process(state) except GraphInterrupt: raise except (AppError, ValidationError): - self.logger.exception("%s 执行失败", self.name) + self.logger.error("节点执行失败") raise except Exception as e: - self.logger.exception("%s 执行失败", self.name) + self.logger.error("节点执行失败") raise ImportProcessError(str(e), node_name=self.name) from e else: - self.logger.info("--- %s 完成 ---", self.name) + self.logger.info("节点完成") return result finally: - self.logger.info( - "%s 耗时 %.3f 秒", self.name, time.perf_counter() - start_time - ) + self.logger.info(f"节点耗时 {time.perf_counter() - start_time:.3f} 秒") @abstractmethod def process(self, state: T) -> T: @@ -61,24 +59,22 @@ class BaseNode(ABC): async def _acall(self, state: T) -> T: """异步节点等待执行完成后再记录结果和耗时""" start_time = time.perf_counter() - self.logger.info("--- %s 开始 ---", self.name) + self.logger.info("节点开始") try: result = await self.process(state) except GraphInterrupt: raise except (AppError, ValidationError): - self.logger.exception("%s 执行失败", self.name) + self.logger.error("节点执行失败") raise except Exception as error: - self.logger.exception("%s 执行失败", self.name) + self.logger.error("节点执行失败") raise ImportProcessError(str(error), node_name=self.name) from error else: - self.logger.info("--- %s 完成 ---", self.name) + self.logger.info("节点完成") return result finally: - self.logger.info( - "%s 耗时 %.3f 秒", self.name, time.perf_counter() - start_time - ) + self.logger.info(f"节点耗时 {time.perf_counter() - start_time:.3f} 秒") def log_step(self, step_name: str, message: str = ""): """记录步骤日志。""" diff --git a/app/services/agent.py b/app/services/agent.py index 89c647b..ef895aa 100644 --- a/app/services/agent.py +++ b/app/services/agent.py @@ -3,6 +3,7 @@ from uuid import uuid4 from langgraph.types import Command +from loguru import logger from app.core.exceptions import BizException from app.observability.tracing import ainvoke_screen @@ -68,10 +69,26 @@ class AgentService: self._active.discard(thread_id) async def _execute(self, thread_id, payload, session_id) -> AgentResult: - state = await ainvoke_screen( - self.graph, payload, thread_id=thread_id, session_id=session_id - ) + # 上下文只作用于当前异步任务，节点自动继承任务标识 + with logger.contextualize(thread_id=thread_id): + logger.info("任务执行开始") + try: + state = await ainvoke_screen( + self.graph, payload, thread_id=thread_id, session_id=session_id + ) + except Exception as exc: + # API 异常处理发生在上下文退出后，保留任务标识用于定位 + exc._agent_thread_id = thread_id + raise + logger.info( + "任务执行结束，状态 {}", + "waiting_for_input" + if state.get("__interrupt__") + else state.get("status"), + ) + waiting = bool(state.get("__interrupt__")) + # 只交付校验通过的完整动作批次，失败草稿留在后端 return AgentResult( thread_id=thread_id, diff --git a/pyproject.toml b/pyproject.toml index 8994eac..ee908af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,4 +12,5 @@ dependencies = [ "langchain>=1.4.2", "fastapi[standard]>=0.141.1", "uvicorn>=0.54.0", + "loguru>=0.7.3", ] diff --git a/uv.lock b/uv.lock index 6465b0c..e687650 100644 --- a/uv.lock +++ b/uv.lock @@ -928,6 +928,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/12/51/892ff5a2efc8e222ca114dd0e19b56978f78a5ee62d3b33783588dd8e6d4/langsmith-0.13.0-py3-none-any.whl", hash = "sha256:89d53d01dcdd0ff4064b425a0476ce51ecca5453d7284313ca4d8c64d326257c", size = 787579, upload-time = "2026-09-18T18:07:07.167Z" }, ] +[[package]] +name = "loguru" +version = "0.7.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "win32-setctime", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/3a/05/a1dae3dffd1116099471c643b8924f5aa6524411dc6c63fdae648c4f1aca/loguru-0.7.3.tar.gz", hash = "sha256:19480589e77d47b8d85b2c827ad95d49bf31b0dcde16593892eb51dd18706eb6", size = 63559, upload-time = "2024-12-06T11:20:56.608Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/0c/29/0348de65b8cc732daa3e33e67806420b2ae89bdce2b04af740289c5c6c8c/loguru-0.7.3-py3-none-any.whl", hash = "sha256:31a33c10c8e1e10422bfd431aeb5d351c7cf7fa671e3c4df004162264b28220c", size = 61595, upload-time = "2024-12-06T11:20:54.538Z" }, +] + [[package]] name = "markdown-it-py" version = "4.2.0" @@ -1033,6 +1046,7 @@ dependencies = [ { name = "langchain-openai" }, { name = "langfuse" }, { name = "langgraph" }, + { name = "loguru" }, { name = "pydantic" }, { name = "python-dotenv" }, { name = "typesafe-sdk" }, @@ -1046,6 +1060,7 @@ requires-dist = [ { name = "langchain-openai", specifier = ">=1.6.2" }, { name = "langfuse", specifier = "==4.15.6" }, { name = "langgraph", specifier = ">=1.2.12" }, + { name = "loguru", specifier = ">=0.7.3" }, { name = "pydantic", specifier = ">=2.13.5,<3" }, { name = "python-dotenv", specifier = ">=1.2.3" }, { name = "typesafe-sdk", specifier = ">=0.7.1" }, @@ -2328,6 +2343,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/be/4d/2d0d67834092e354d2b0498f014a41249a89556bc406cf86f3e1557bb463/websockets-16.1.1-py3-none-any.whl", hash = "sha256:6abbd3e82c731c8e531714466acd5d87b5e88ac3243465337ba71d68e23ae7e3", size = 173814, upload-time = "2026-07-17T22:51:04.184Z" }, ] +[[package]] +name = "win32-setctime" +version = "1.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b3/8f/705086c9d734d3b663af0e9bb3d4de6578d08f46b1b101c2442fd9aecaa2/win32_setctime-1.2.0.tar.gz", hash = "sha256:ae1fdf948f5640aae05c511ade119313fb6a30d7eabe25fef9764dca5873c4c0", size = 4867, upload-time = "2024-12-07T15:28:28.314Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e1/07/c6fe3ad3e685340704d314d765b7912993bcb8dc198f0e7a89382d37974b/win32_setctime-1.2.0-py3-none-any.whl", hash = "sha256:95d644c4e708aba81dc3704a116d8cbc974d70b3bdb8be1d150e36be6e9d1390", size = 4083, upload-time = "2024-12-07T15:28:26.465Z" }, +] + [[package]] name = "wrapt" version = "2.4.1"