business-assistant/src/knowledge_profiles.py

"""Explicit knowledge destinations shared by preparation and offline import."""
import re
from pathlib import Path

AGENTS = ('business', 'talent')


def prepared_path(root, agent):
    if agent not in AGENTS:
        raise ValueError('unknown_knowledge_agent')
    return Path(root) / 'knowledge' / ('prepared' if agent == 'business' else 'prepared-talent')


def for_agent(settings, agent, db_prefix=None):
    if agent not in AGENTS:
        raise ValueError('unknown_knowledge_agent')
    result = dict(settings)
    business = settings['MILVUS_DB_PREFIX']
    talent = settings.get('TALENT_MILVUS_DB_PREFIX', 'talent_knowledge')
    prefix = db_prefix if db_prefix is not None else business if agent == 'business' else talent
    if not isinstance(prefix, str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9_]{0,99}', prefix):
        raise ValueError('invalid_database_prefix')
    if business == talent or prefix == (talent if agent == 'business' else business):
        raise ValueError('knowledge_agent_prefix_collision')
    result.update(MILVUS_DB_PREFIX=prefix, KNOWLEDGE_AGENT=agent)
    return result


def records_agent(records):
    agents = {record.get('knowledge_agent', 'business') for record in records}
    if len(agents) != 1 or not agents <= set(AGENTS):
        raise ValueError('mixed_or_unknown_knowledge_agents')
    return next(iter(agents))


def check_records(records, settings):
    if records_agent(records) != settings.get('KNOWLEDGE_AGENT', 'business'):
        raise ValueError('knowledge_agent_mismatch')


def talent_library(record):
    if record.get('knowledge_agent') != 'talent':
        return None
    libraries = {'人才政策和服务': 'article', '人才政策问答': 'qa', '青浦区惠企政策事项': 'policy'}
    try:
        return libraries[record['collection']]
    except KeyError:
        raise ValueError('unknown_talent_collection') from None