kg_agent/update_functions.py

#!/usr/bin/env python3
"""
更新 4 个评估函数,使用新的 system/user prompt 结构
"""

import re

def read_file(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        return f.read()

def write_file(filepath, content):
    with open(filepath, 'w', encoding='utf-8') as f:
        f.write(content)

def update_evaluate_is_knowledge(content):
    """更新 evaluate_is_knowledge 函数"""
    # 查找并替换
    old_pattern = r'''try:
        prompt_text = PROMPT1_IS_KNOWLEDGE\.format\(
            title=post\.title,
            body_text=post\.body_text or "",
            num_images=len\(image_urls\)
        \)

        data = await _call_openrouter_api\(prompt_text, image_urls, semaphore\)'''

    new_code = '''try:
        user_prompt = PROMPT1_USER_TEMPLATE.format(
            title=post.title,
            body_text=post.body_text or "",
            num_images=len(image_urls)
        )

        data = await _call_openrouter_api(
            system_prompt=PROMPT1_SYSTEM,
            user_prompt=user_prompt,
            image_urls=image_urls,
            semaphore=semaphore
        )'''

    content = re.sub(old_pattern, new_code, content)
    print("✅ evaluate_is_knowledge 更新完成")
    return content

def update_evaluate_is_content_knowledge(content):
    """更新 evaluate_is_content_knowledge 函数"""
    old_pattern = r'''try:
        prompt_text = PROMPT2_IS_CONTENT_KNOWLEDGE\.format\(
            title=post\.title,
            body_text=post\.body_text or "",
            num_images=len\(image_urls\)
        \)

        data = await _call_openrouter_api\(prompt_text, image_urls, semaphore\)'''

    new_code = '''try:
        user_prompt = PROMPT2_USER_TEMPLATE.format(
            title=post.title,
            body_text=post.body_text or "",
            num_images=len(image_urls)
        )

        data = await _call_openrouter_api(
            system_prompt=PROMPT2_SYSTEM,
            user_prompt=user_prompt,
            image_urls=image_urls,
            semaphore=semaphore
        )'''

    content = re.sub(old_pattern, new_code, content)
    print("✅ evaluate_is_content_knowledge 更新完成")
    return content

def update_evaluate_purpose_match(content):
    """更新 evaluate_purpose_match 函数"""
    old_pattern = r'''try:
        prompt_text = PROMPT3_PURPOSE_MATCH\.format\(
            original_query=original_query,
            title=post\.title,
            body_text=post\.body_text or "",
            num_images=len\(image_urls\)
        \)

        data = await _call_openrouter_api\(prompt_text, image_urls, semaphore\)'''

    new_code = '''try:
        user_prompt = PROMPT3_USER_TEMPLATE.format(
            original_query=original_query,
            title=post.title,
            body_text=post.body_text or "",
            num_images=len(image_urls)
        )

        data = await _call_openrouter_api(
            system_prompt=PROMPT3_SYSTEM,
            user_prompt=user_prompt,
            image_urls=image_urls,
            semaphore=semaphore
        )'''

    content = re.sub(old_pattern, new_code, content)
    print("✅ evaluate_purpose_match 更新完成")
    return content

def update_evaluate_category_match(content):
    """更新 evaluate_category_match 函数"""
    old_pattern = r'''try:
        prompt_text = PROMPT4_CATEGORY_MATCH\.format\(
            original_query=original_query,
            title=post\.title,
            body_text=post\.body_text or "",
            num_images=len\(image_urls\)
        \)

        data = await _call_openrouter_api\(prompt_text, image_urls, semaphore\)'''

    new_code = '''try:
        user_prompt = PROMPT4_USER_TEMPLATE.format(
            original_query=original_query,
            title=post.title,
            body_text=post.body_text or "",
            num_images=len(image_urls)
        )

        data = await _call_openrouter_api(
            system_prompt=PROMPT4_SYSTEM,
            user_prompt=user_prompt,
            image_urls=image_urls,
            semaphore=semaphore
        )'''

    content = re.sub(old_pattern, new_code, content)
    print("✅ evaluate_category_match 更新完成")
    return content

def main():
    filepath = 'post_evaluator_v3.py'

    print("📖 读取文件...")
    content = read_file(filepath)

    print("\n🔧 更新评估函数...")
    content = update_evaluate_is_knowledge(content)
    content = update_evaluate_is_content_knowledge(content)
    content = update_evaluate_purpose_match(content)
    content = update_evaluate_category_match(content)

    print("\n💾 保存文件...")
    write_file(filepath, content)

    print("\n✅ 所有评估函数更新完成!")

if __name__ == '__main__':
    main()