mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2026-10-04 03:10:01 +08:00
922d092650
* ♻️ refactor(core): 重构 AI 能力与定时任务调度系统 - 【AI 能力与工具】重构 Capability 注册与管理机制，引入 CapabilityManager 统一管理 - 移除全局能力注册表，改用声明式装饰器 `@capability` 进行解耦注册 - 重构工具解析器链，使用统一的 BaseToolResolver 代替原有的多个特定解析器 - 增强工具查询过滤，支持通配符匹配、工具箱过滤和排除标签 - 【定时任务调度】重构定时任务管理器，引入 SchedulerRegistry 统一管理任务元数据 - 引入 JobConfig 聚合定时任务配置，支持用户维度的定时任务调度 - 重构执行分发器，支持并发限制、串行间隔和随机延迟打散 - 【运行上下文】引入 ScheduledDeps 以支持后台和定时任务环境下的依赖注入 - 优化 RunContext，支持从定时任务上下文快速构造，并提供 emit 辅助方法 - 【日志与监控】引入 AILoggerProxy，实现 AI 各模块的专属日志输出 - 将各模块的全局 logger 替换为对应的模块专属日志代理 - 【其他优化】修复 Pydantic V1 兼容层中 model_validator 的装饰器兼容性问题 - 在非交互式环境（如定时任务）中自动隐藏 HITL 交互工具以节省 Token * ♻️ refactor(core): 优化内部导入路径并提升 Pydantic 兼容性 - 【重构】将 `services/ai` 模块内的绝对导入重构为相对导入，优化包结构 - 【重构】移除不必要的 `if TYPE_CHECKING` 保护，通过 `from __future__ import annotations` 直接导入类型 - 【清理】清理 `core/messages/types.py` 中未使用的 `AssistantContentUnion` 等联合类型定义 - 【优化】在 `utils/pydantic_compat.py` 中新增 `model_rebuild` 兼容函数，统一 Pydantic V1/V2 的模型重建逻辑 - 【优化】将部分函数内部的延迟导入提升至模块顶部，规范代码结构 * ♻️ refactor(imports): 优化导入路径为相对导入并清理冗余导入 - 【重构】将 AI 服务相关模块中的绝对导入路径修改为相对导入，提升模块内聚性与可移植性 - 【清理】移除多处函数内部或类方法中未使用的冗余导入，避免循环引用和资源浪费 - 【格式化】微调部分工具装饰器和返回语句的格式与尾随逗号 * 🚨 auto fix by pre-commit hooks --------- Co-authored-by: webjoin111 <455457521@qq.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
104 lines
3.7 KiB
Python
104 lines
3.7 KiB
Python
from collections.abc import Callable
|
||
from typing import TYPE_CHECKING, Any, Generic, cast
|
||
from typing_extensions import TypeVar
|
||
|
||
from jinja2 import Environment
|
||
|
||
from zhenxun.services.ai.utils.logger import log_core as logger
|
||
from zhenxun.utils.pydantic_compat import model_dump
|
||
|
||
if TYPE_CHECKING:
|
||
from zhenxun.services.ai.run.context import RunContext
|
||
|
||
AgentDepsT = TypeVar("AgentDepsT", default=Any)
|
||
|
||
|
||
class PromptTemplate(Generic[AgentDepsT]):
|
||
"""
|
||
核心 Prompt 渲染引擎。
|
||
基于 Jinja2 提供强大的模板变量替换功能。
|
||
支持沙盒隔离的自定义过滤器(filters)、全局函数(globals)，
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
template_string: str,
|
||
custom_filters: dict[str, Callable] | None = None,
|
||
custom_globals: dict[str, Any] | None = None,
|
||
):
|
||
"""
|
||
初始化 Prompt 渲染引擎。
|
||
|
||
参数:
|
||
template_string: Jinja2 模板格式的提示词原文本。
|
||
custom_filters: 挂载到 Jinja2 渲染环境的自定义过滤器字典，默认 None。
|
||
custom_globals: 挂载到 Jinja2 渲染环境的全局变量或辅助函数字典，默认 None。
|
||
"""
|
||
self.template_string = template_string
|
||
|
||
self._env = Environment(autoescape=False)
|
||
if custom_filters:
|
||
self._env.filters.update(custom_filters)
|
||
if custom_globals:
|
||
self._env.globals.update(custom_globals)
|
||
|
||
try:
|
||
self._template = (
|
||
self._env.from_string(template_string) if template_string else None
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"PromptTemplate 模板语法编译失败: {e}")
|
||
self._template = None
|
||
|
||
def format_with_context(self, context: "RunContext[AgentDepsT]") -> str:
|
||
"""从运行上下文中自动提取依赖和状态字典，用于模板渲染"""
|
||
|
||
vars_dict: dict[str, Any] = {
|
||
"ctx": context,
|
||
"ctx_deps": context.deps,
|
||
"ctx_state": context.state,
|
||
"ctx_shared": context.shared_state,
|
||
}
|
||
|
||
flat_deps = {}
|
||
if context.deps is not None:
|
||
if hasattr(context.deps, "model_dump"):
|
||
flat_deps.update(model_dump(cast(Any, context.deps), exclude_none=True))
|
||
elif hasattr(context.deps, "__dict__"):
|
||
flat_deps.update(context.deps.__dict__)
|
||
elif isinstance(context.deps, dict):
|
||
flat_deps.update(context.deps)
|
||
|
||
vars_dict.update(flat_deps)
|
||
|
||
if context.state:
|
||
collisions = set(flat_deps.keys()) & set(context.state.keys())
|
||
if collisions:
|
||
logger.warning(
|
||
f"Prompt 模板变量发生名称冲突: {collisions}。"
|
||
"`state` 已覆盖 `deps` 中的同名变量。"
|
||
"建议在模板中使用安全隔离对象获取（如 {{ ctx_state.xxx }}）。"
|
||
)
|
||
vars_dict.update(context.state)
|
||
|
||
return self.render(**vars_dict)
|
||
|
||
def render(self, **variables: Any) -> str:
|
||
if not self._template:
|
||
return self.template_string or ""
|
||
|
||
try:
|
||
return self._template.render(**variables)
|
||
except Exception as e:
|
||
logger.error(
|
||
f"Jinja2 Prompt 模板渲染失败！\n"
|
||
f"错误信息: {e}\n"
|
||
f"请检查传入的变量或 Prompt 模板语法是否有误。\n"
|
||
f"模板原内容截断: {self.template_string[:150]}...",
|
||
e=e,
|
||
)
|
||
raise ValueError(f"Prompt 模板渲染异常: {e}") from e
|
||
|
||
def __str__(self) -> str:
|
||
return self.template_string
|