refactor(core): 重构配置与入口模块,统一配置结构并调整导入

- 扁平化应用配置类,合并数据库、LLM、爬虫和通知配置
- 重新实现配置加载,统一环境变量前缀和字段命名
- 入口脚本调整,增加源码路径处理,支持模块绝对导入和直接运行
- HRAgentApplication中使用新配置字段访问方式
- 优化通知渠道注册逻辑,适配新的配置字段重命名
- 模块路径统一由ylhp_hr_2.0改为ylhp_hr_2_0,确保导入一致性
- 删除旧配置模块,避免配置重复和混淆
- service.analysis包暴露MockLLMClient,完善LLM客户端选项
- 保留主入口运行示例,演示系统初始化与功能打印
This commit is contained in:
2026-03-24 11:57:45 +08:00
parent a40c239996
commit 6a5495005e
39 changed files with 226 additions and 145 deletions

78
main.py Normal file
View File

@@ -0,0 +1,78 @@
"""Resume Intelligence Agent - Entry Point
简历智能体系统入口
Usage:
# 运行应用
uv run python main.py
# 或使用模块方式
uv run python -m src.main.python.cn.yinlihupo.ylhp_hr_2.0.main
Environment Variables:
# 数据库配置
DB_URL=mysql+pymysql://root:123456@10.200.8.25:3306/hr_agent
# LLM 配置
LLM_PROVIDER=mock
LLM_API_KEY=your_api_key
# 爬虫配置
CRAWLER_BOSS_WT_TOKEN=your_boss_token
# 通知配置
NOTIFY_WECHAT_WORK_WEBHOOK=https://qyapi.weixin.qq.com/cgi-bin/webhook/...
"""
import asyncio
import sys
from pathlib import Path
# 添加源码路径到 sys.path
src_path = Path(__file__).parent / "src" / "main" / "python"
if str(src_path) not in sys.path:
sys.path.insert(0, str(src_path))
# 导入应用
from cn.yinlihupo.ylhp_hr_2_0.main import HRAgentApplication, get_app
from cn.yinlihupo.ylhp_hr_2_0.domain.candidate import CandidateSource
async def demo():
"""演示:使用 HR Agent 进行简历处理"""
print("=" * 50)
print("简历智能体系统 - 演示")
print("=" * 50)
# 初始化应用
app = get_app()
print("\n已注册爬虫:")
for source in app.crawler_factory.get_registered_sources():
print(f" - {source.value}")
print("\n已配置通知渠道:")
if app.notification_service:
for channel_type in app.notification_service.get_configured_channels():
print(f" - {channel_type.value}")
else:
print(" - 无")
print("\n可用的评价方案:")
schemas = app.analyzer.schema_service.list_schemas()
for schema in schemas:
default_mark = " (默认)" if schema.is_default else ""
print(f" - {schema.name}{default_mark}")
print("\n" + "=" * 50)
print("系统初始化完成")
print("=" * 50)
# 示例:爬取并入库(需要配置 CRAWLER_BOSS_WT_TOKEN
# await app.crawl_and_ingest(
# source=CandidateSource.BOSS,
# job_id="your_job_id"
# )
if __name__ == "__main__":
asyncio.run(demo())

View File

@@ -1,95 +0,0 @@
"""Application settings"""
from typing import Optional
from pydantic_settings import BaseSettings
from pydantic import Field
class DatabaseSettings(BaseSettings):
"""数据库配置"""
url: str = Field(default="sqlite:///./hr_agent.db", description="数据库连接URL")
echo: bool = Field(default=False, description="是否打印SQL语句")
class Config:
env_prefix = "DB_"
class LLMSettings(BaseSettings):
"""LLM 配置"""
provider: str = Field(default="openai", description="LLM提供商: openai, claude, mock")
api_key: Optional[str] = Field(default=None, description="API密钥")
base_url: Optional[str] = Field(default=None, description="自定义API地址")
model: str = Field(default="gpt-4", description="模型名称")
temperature: float = Field(default=0.7, description="温度参数")
max_tokens: int = Field(default=2000, description="最大token数")
class Config:
env_prefix = "LLM_"
class NotificationSettings(BaseSettings):
"""通知配置"""
# 企业微信
wechat_work_webhook: Optional[str] = Field(default=None, description="企业微信Webhook")
wechat_work_mentioned: Optional[str] = Field(default=None, description="@提醒列表,逗号分隔")
# 钉钉
dingtalk_webhook: Optional[str] = Field(default=None, description="钉钉Webhook")
dingtalk_secret: Optional[str] = Field(default=None, description="钉钉加签密钥")
dingtalk_at_mobiles: Optional[str] = Field(default=None, description="@手机号列表,逗号分隔")
# 邮件
email_smtp_host: Optional[str] = Field(default=None, description="SMTP服务器")
email_smtp_port: int = Field(default=587, description="SMTP端口")
email_username: Optional[str] = Field(default=None, description="邮箱用户名")
email_password: Optional[str] = Field(default=None, description="邮箱密码")
email_from: Optional[str] = Field(default=None, description="发件人地址")
email_to: Optional[str] = Field(default=None, description="收件人地址,逗号分隔")
class Config:
env_prefix = "NOTIFY_"
class CrawlerSettings(BaseSettings):
"""爬虫配置"""
boss_wt_token: Optional[str] = Field(default=None, description="Boss直聘WT Token")
class Config:
env_prefix = "CRAWLER_"
class Settings(BaseSettings):
"""应用配置"""
# 应用信息
app_name: str = Field(default="ylhp_hr_2.0", description="应用名称")
app_version: str = Field(default="0.1.0", description="应用版本")
debug: bool = Field(default=False, description="调试模式")
# 子配置
database: DatabaseSettings = Field(default_factory=DatabaseSettings)
llm: LLMSettings = Field(default_factory=LLMSettings)
notification: NotificationSettings = Field(default_factory=NotificationSettings)
crawler: CrawlerSettings = Field(default_factory=CrawlerSettings)
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
# 全局配置实例
_settings: Optional[Settings] = None
def get_settings() -> Settings:
"""获取配置实例(单例)"""
global _settings
if _settings is None:
_settings = Settings()
return _settings
def reload_settings() -> Settings:
"""重新加载配置"""
global _settings
_settings = Settings()
return _settings

View File

@@ -0,0 +1,69 @@
"""Application settings"""
from typing import Optional
from pydantic_settings import BaseSettings
from pydantic import Field
class Settings(BaseSettings):
"""应用配置 - 扁平化结构,所有配置项都在此类中"""
# 应用信息
app_name: str = Field(default="ylhp_hr_2.0", description="应用名称")
app_version: str = Field(default="0.1.0", description="应用版本")
debug: bool = Field(default=False, description="调试模式")
# 数据库配置 (前缀: DB_)
db_url: str = Field(default="sqlite:///./hr_agent.db", description="数据库连接URL")
db_echo: bool = Field(default=False, description="是否打印SQL语句")
# LLM 配置 (前缀: LLM_)
llm_provider: str = Field(default="mock", description="LLM提供商: openai, claude, mock")
llm_api_key: Optional[str] = Field(default=None, description="API密钥")
llm_base_url: Optional[str] = Field(default=None, description="自定义API地址")
llm_model: str = Field(default="gpt-4", description="模型名称")
llm_temperature: float = Field(default=0.7, description="温度参数")
llm_max_tokens: int = Field(default=2000, description="最大token数")
# 爬虫配置 (前缀: CRAWLER_)
crawler_boss_wt_token: Optional[str] = Field(default=None, description="Boss直聘WT Token")
# 通知配置 - 企业微信 (前缀: NOTIFY_)
notify_wechat_work_webhook: Optional[str] = Field(default=None, description="企业微信Webhook")
notify_wechat_work_mentioned: Optional[str] = Field(default=None, description="@提醒列表")
# 通知配置 - 钉钉 (前缀: NOTIFY_)
notify_dingtalk_webhook: Optional[str] = Field(default=None, description="钉钉Webhook")
notify_dingtalk_secret: Optional[str] = Field(default=None, description="钉钉加签密钥")
notify_dingtalk_at_mobiles: Optional[str] = Field(default=None, description="@手机号列表")
# 通知配置 - 邮件 (前缀: NOTIFY_)
notify_email_smtp_host: Optional[str] = Field(default=None, description="SMTP服务器")
notify_email_smtp_port: int = Field(default=587, description="SMTP端口")
notify_email_username: Optional[str] = Field(default=None, description="邮箱用户名")
notify_email_password: Optional[str] = Field(default=None, description="邮箱密码")
notify_email_from: Optional[str] = Field(default=None, description="发件人地址")
notify_email_to: Optional[str] = Field(default=None, description="收件人地址")
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
extra = "ignore" # 忽略额外的环境变量
# 全局配置实例
_settings: Optional[Settings] = None
def get_settings() -> Settings:
"""获取配置实例(单例)"""
global _settings
if _settings is None:
_settings = Settings()
return _settings
def reload_settings() -> Settings:
"""重新加载配置"""
global _settings
_settings = Settings()
return _settings

View File

@@ -1,29 +1,57 @@
"""Application entry point"""
import asyncio
import sys
from pathlib import Path
from typing import Optional
from .config.settings import get_settings
from .domain.candidate import CandidateSource
from .service.crawler import CrawlerFactory, BossCrawler
from .service.ingestion import (
UnifiedIngestionService,
DataNormalizer,
DataValidator,
DeduplicationService
)
from .service.analysis import (
ResumeAnalyzer,
EvaluationSchemaService,
LLMClient,
OpenAIClient,
MockLLMClient
)
from .service.notification import (
NotificationService,
WeChatWorkChannel,
DingTalkChannel,
EmailChannel
)
# 处理直接运行时的导入问题
try:
from .config.settings import get_settings
from .domain.candidate import CandidateSource
from .service.crawler import CrawlerFactory, BossCrawler
from .service.ingestion import (
UnifiedIngestionService,
DataNormalizer,
DataValidator,
DeduplicationService
)
from .service.analysis import (
ResumeAnalyzer,
EvaluationSchemaService,
LLMClient,
OpenAIClient,
MockLLMClient
)
from .service.notification import (
NotificationService,
WeChatWorkChannel,
DingTalkChannel,
EmailChannel
)
except ImportError:
# 直接运行时,使用绝对导入
from cn.yinlihupo.ylhp_hr_2_0.config.settings import get_settings
from cn.yinlihupo.ylhp_hr_2_0.domain.candidate import CandidateSource
from cn.yinlihupo.ylhp_hr_2_0.service.crawler import CrawlerFactory, BossCrawler
from cn.yinlihupo.ylhp_hr_2_0.service.ingestion import (
UnifiedIngestionService,
DataNormalizer,
DataValidator,
DeduplicationService
)
from cn.yinlihupo.ylhp_hr_2_0.service.analysis import (
ResumeAnalyzer,
EvaluationSchemaService,
LLMClient,
OpenAIClient,
MockLLMClient
)
from cn.yinlihupo.ylhp_hr_2_0.service.notification import (
NotificationService,
WeChatWorkChannel,
DingTalkChannel,
EmailChannel
)
class HRAgentApplication:
@@ -65,8 +93,8 @@ class HRAgentApplication:
def _init_crawlers(self):
"""初始化爬虫"""
# Boss 爬虫
if self.settings.crawler.boss_wt_token:
boss_crawler = BossCrawler(wt_token=self.settings.crawler.boss_wt_token)
if self.settings.crawler_boss_wt_token:
boss_crawler = BossCrawler(wt_token=self.settings.crawler_boss_wt_token)
self.crawler_factory.register(CandidateSource.BOSS, boss_crawler)
print("Boss crawler registered")
@@ -91,16 +119,16 @@ class HRAgentApplication:
def _create_llm_client(self) -> LLMClient:
"""创建 LLM 客户端"""
provider = self.settings.llm.provider.lower()
provider = self.settings.llm_provider.lower()
if provider == "openai":
if self.settings.llm.api_key:
if self.settings.llm_api_key:
return OpenAIClient(
api_key=self.settings.llm.api_key,
model=self.settings.llm.model,
base_url=self.settings.llm.base_url,
temperature=self.settings.llm.temperature,
max_tokens=self.settings.llm.max_tokens
api_key=self.settings.llm_api_key,
model=self.settings.llm_model,
base_url=self.settings.llm_base_url,
temperature=self.settings.llm_temperature,
max_tokens=self.settings.llm_max_tokens
)
else:
print("Warning: OpenAI API key not configured, using mock client")
@@ -118,55 +146,55 @@ class HRAgentApplication:
notification_service = NotificationService()
# 企业微信
if self.settings.notification.wechat_work_webhook:
if self.settings.notify_wechat_work_webhook:
mentioned_list = None
if self.settings.notification.wechat_work_mentioned:
if self.settings.notify_wechat_work_mentioned:
mentioned_list = [
m.strip()
for m in self.settings.notification.wechat_work_mentioned.split(",")
for m in self.settings.notify_wechat_work_mentioned.split(",")
]
channel = WeChatWorkChannel(
webhook_url=self.settings.notification.wechat_work_webhook,
webhook_url=self.settings.notify_wechat_work_webhook,
mentioned_list=mentioned_list
)
notification_service.register_channel(channel)
print("WeChat Work channel registered")
# 钉钉
if self.settings.notification.dingtalk_webhook:
if self.settings.notify_dingtalk_webhook:
at_mobiles = None
if self.settings.notification.dingtalk_at_mobiles:
if self.settings.notify_dingtalk_at_mobiles:
at_mobiles = [
m.strip()
for m in self.settings.notification.dingtalk_at_mobiles.split(",")
for m in self.settings.notify_dingtalk_at_mobiles.split(",")
]
channel = DingTalkChannel(
webhook_url=self.settings.notification.dingtalk_webhook,
secret=self.settings.notification.dingtalk_secret,
webhook_url=self.settings.notify_dingtalk_webhook,
secret=self.settings.notify_dingtalk_secret,
at_mobiles=at_mobiles
)
notification_service.register_channel(channel)
print("DingTalk channel registered")
# 邮件
if (self.settings.notification.email_smtp_host and
self.settings.notification.email_username):
if (self.settings.notify_email_smtp_host and
self.settings.notify_email_username):
to_addrs = []
if self.settings.notification.email_to:
if self.settings.notify_email_to:
to_addrs = [
addr.strip()
for addr in self.settings.notification.email_to.split(",")
for addr in self.settings.notify_email_to.split(",")
]
if to_addrs:
channel = EmailChannel(
smtp_host=self.settings.notification.email_smtp_host,
smtp_port=self.settings.notification.email_smtp_port,
username=self.settings.notification.email_username,
password=self.settings.notification.email_password or "",
from_addr=self.settings.notification.email_from or self.settings.notification.email_username,
smtp_host=self.settings.notify_email_smtp_host,
smtp_port=self.settings.notify_email_smtp_port,
username=self.settings.notify_email_username,
password=self.settings.notify_email_password or "",
from_addr=self.settings.notify_email_from or self.settings.notify_email_username,
to_addrs=to_addrs
)
notification_service.register_channel(channel)

View File

@@ -4,7 +4,7 @@ from .evaluation_schema import EvaluationSchemaService
from .resume_analyzer import ResumeAnalyzer
from .scoring_engine import ScoringEngine
from .prompt_builder import PromptBuilder
from .llm_client import LLMClient, OpenAIClient
from .llm_client import LLMClient, OpenAIClient, MockLLMClient
__all__ = [
"EvaluationSchemaService",
@@ -13,4 +13,5 @@ __all__ = [
"PromptBuilder",
"LLMClient",
"OpenAIClient",
"MockLLMClient",
]