جستجوی کد با استفاده از embeddings

جستجوی کد با استفاده از embeddings

embeddings
preview

این نوت‌بوک نشان می‌دهد چگونه می‌توان از embeddings برای پیاده‌سازی جستجوی معنایی در میان کدهای کامپیوتری استفاده کرد. برای این پست ما از کد openai-python که در گیت‌هاب قایل دسترسی است٬ استفاده می‌کنیم. سپس نسخه ساده‌ای از تجزیه فایل و استخراج توابع از فایل‌های پایتون را پیاده‌سازی می‌کنیم که می‌توانند embed، index و query شوند.

توابع کمکی #

برای شروع به چند تابع تجزیه‌ی ساده برای استخراج توابع داخل کدبیس خود نیاز داریم.

 1import pandas as pd
 2from pathlib import Path
 3
 4DEF_PREFIXES = ['def ', 'async def ']
 5NEWLINE = '\n'
 6
 7def get_function_name(code):
 8    """
 9    Extract function name from a line beginning with 'def' or 'async def'.
10    """
11    for prefix in DEF_PREFIXES:
12        if code.startswith(prefix):
13            return code[len(prefix): code.index('(')]
14
15
16def get_until_no_space(all_lines, i):
17    """
18    Get all lines until a line outside the function definition is found.
19    """
20    ret = [all_lines[i]]
21    for j in range(i + 1, len(all_lines)):
22        if len(all_lines[j]) == 0 or all_lines[j][0] in [' ', '\t', ')']:
23            ret.append(all_lines[j])
24        else:
25            break
26    return NEWLINE.join(ret)
27
28
29def get_functions(filepath):
30    """
31    Get all functions in a Python file.
32    """
33    with open(filepath, 'r') as file:
34        all_lines = file.read().replace('\r', NEWLINE).split(NEWLINE)
35        for i, l in enumerate(all_lines):
36            for prefix in DEF_PREFIXES:
37                if l.startswith(prefix):
38                    code = get_until_no_space(all_lines, i)
39                    function_name = get_function_name(code)
40                    yield {
41                        'code': code,
42                        'function_name': function_name,
43                        'filepath': filepath,
44                    }
45                    break
46
47
48def extract_functions_from_repo(code_root):
49    """
50    Extract all .py functions from the repository.
51    """
52    code_files = list(code_root.glob('**/*.py'))
53
54    num_files = len(code_files)
55    print(f'Total number of .py files: {num_files}')
56
57    if num_files == 0:
58        print('Verify openai-python repo exists and code_root is set correctly.')
59        return None
60
61    all_funcs = [
62        func
63        for code_file in code_files
64        for func in get_functions(str(code_file))
65    ]
66
67    num_funcs = len(all_funcs)
68    print(f'Total number of functions extracted: {num_funcs}')
69
70    return all_funcs

بارگذاری داده‌ها #

ابتدا رپوی openai-python را گلون کرده و اطلاعات مورد نیاز را با استفاده از توابعی که در بالا تعریف کردیم استخراج می‌کنیم.

 1
 2# Set user root directory to the 'openai-python' repository
 3root_dir = Path.home()
 4
 5# Clone the repo
 6git clone https://github.com/openai/openai-python.git
 7
 8code_root = root_dir / 'openai-python'
 9
10# Extract all functions from the repository
11all_funcs = extract_functions_from_repo(code_root)
Total number of .py files: 51
Total number of functions extracted: 97

حالا که محتوای خود را داریم، می‌توانیم داده‌ها را به مدل text-embedding-3-small ارسال کرده تا بردارهای embeddings را دریافت کنیم.

برای اجرای کدهای زیر ابتدا باید یک کلید API را از طریق پنل کاربری گیلاس تولید کنید. برای این کار ابتدا یک حساب کاربری جدید بسازید یا اگر صاحب حساب کاربری هستید وارد پنل کاربری خود شوید. سپس، به صفحه کلید API بروید و با کلیک روی دکمه “ساخت کلید API” یک کلید جدید برای دسترسی به Gilas API بسازید.
 1from openai import OpenAI # for calling the OpenAI API
 2import os
 3
 4client = OpenAI(
 5    api_key=os.environ.get(("GILAS_API_KEY", "<کلید API خود را اینجا بسازید https://dashboard.gilas.io/apiKey>")), 
 6    base_url="https://api.gilas.io/v1/" # Gilas APIs
 7)
 8
 9def get_embedding(query)
10    query_embedding_response = client.embeddings.create(
11        model=embedding_model,
12        input=query,
13    )
14    return query_embedding_response.data[0].embedding
15
16df = pd.DataFrame(all_funcs)
17df['code_embedding'] = df['code'].apply(lambda x: get_embedding(x, model='text-embedding-3-small'))
18df['filepath'] = df['filepath'].map(lambda x: Path(x).relative_to(code_root))
19df.to_csv("data/code_search_openai-python.csv", index=False)
20df.head()

خروجی:

codefunction_namefilepathcode_embedding
0def _console_log_level():\n if openai.log i..._console_log_levelopenai/util.py[0.005937571171671152, 0.05450401455163956, 0....
1def log_debug(message, **params):\n msg = l...log_debugopenai/util.py[0.017557814717292786, 0.05647840350866318, -0...
2def log_info(message, **params):\n msg = lo...log_infoopenai/util.py[0.022524144500494003, 0.06219055876135826, -0...
3def log_warn(message, **params):\n msg = lo...log_warnopenai/util.py[0.030524108558893204, 0.0667714849114418, -0....
4def logfmt(props):\n def fmt(key, val):\n ...logfmtopenai/util.py[0.05337328091263771, 0.03697286546230316, -0....

تست #

بیایید اندپوینت خود را با چند پرس و جوی ساده تست کنیم. اگر با رپوی openai-python آشنا باشید، خواهید دید که می‌توانیم به راحتی توابع مورد نظر خود را تنها با یک توضیح ساده به زبان انگلیسی پیدا کنیم.

برای این کار یک متود search_functions تعریف می‌کنیم که داده‌هایی شامل embeddings، یک پرس و جو و برخی مقادیر پیکربندی دیگر را به عنوان ورودی دریافت می‌کتد. فرآیند جستجوی پایگاه داده‌ی برداری به این صورت عمل می‌کند:

  1. ابتدا پرس و جوی خود (code_query) را با text-embedding-3-small تبدیل به بردار امبدینگ می‌کنیم. دلیل این کار این است که یک متنی مانند “یک تابع که یک رشته را معکوس می‌کند” و یک تابع واقعی مانند def reverse(string): return string[::-1] هنگام embed شدن بسیار مشابه خواهند بود.
  2. سپس شباهت کسینوسی بین embedding رشته پرس و جوی و تمام نقاط داده در پایگاه داده را محاسبه می‌کنیم. این کار فاصله بین هر نقطه و پرس و جوی ما را می‌دهد.
  3. در نهایت تمام نقاط داده خود را بر اساس فاصله آنها با رشته پرس و جوی خود مرتب کرده و تعداد نتایج درخواست شده در پارامترهای تابع را برمی‌گردانیم.
 1def cosine_similarity(vec1, vec2):
 2    """Calculate the cosine similarity between two vectors."""
 3    vec1 = np.array(vec1, dtype=float)
 4    vec2 = np.array(vec2, dtype=float)
 5
 6
 7    dot_product = np.dot(vec1, vec2)
 8    norm_vec1 = np.linalg.norm(vec1)
 9    norm_vec2 = np.linalg.norm(vec2)
10    return dot_product / (norm_vec1 * norm_vec2)
11
12
13def search_functions(df, code_query, n=3, pprint=True, n_lines=7):
14    embedding = get_embedding(code_query, model='text-embedding-3-small')
15    df['similarities'] = df.code_embedding.apply(lambda x: cosine_similarity(x, embedding))
16
17    res = df.sort_values('similarities', ascending=False).head(n)
18
19    if pprint:
20        for r in res.iterrows():
21            print(f"{r[1].filepath}:{r[1].function_name}  score={round(r[1].similarities, 3)}")
22            print("\n".join(r[1].code.split("\n")[:n_lines]))
23            print('-' * 70)
24
25    return res
1res = search_functions(df, 'fine-tuning input data validation logic', n=3)

نتیجه جستجوی متنی روی دیتابیس:

openai/validators.py:format_inferrer_validator  score=0.453
def format_inferrer_validator(df):
    """
    This validator will infer the likely fine-tuning format of the data, and display it to the user if it is classification.
    It will also suggest to use ada and explain train/validation split benefits.
    """
    ft_type = infer_task_type(df)
    immediate_msg = None
----------------------------------------------------------------------
openai/validators.py:infer_task_type  score=0.37
def infer_task_type(df):
    """
    Infer the likely fine-tuning task type from the data
    """
    CLASSIFICATION_THRESHOLD = 3  # min_average instances of each class
    if sum(df.prompt.str.len()) == 0:
        return "open-ended generation"
----------------------------------------------------------------------
openai/validators.py:apply_validators  score=0.369
def apply_validators(
    df,
    fname,
    remediation,
    validators,
    auto_accept,
    write_out_file_func,
----------------------------------------------------------------------

مثال دیگری از جستجو:

1res = search_functions(df, 'find common suffix', n=2, n_lines=10)

نتیجه جستجو:

openai/validators.py:get_common_xfix  score=0.487
def get_common_xfix(series, xfix="suffix"):
    """
    Finds the longest common suffix or prefix of all the values in a series
    """
    common_xfix = ""
    while True:
        common_xfixes = (
            series.str[-(len(common_xfix) + 1) :]
            if xfix == "suffix"
            else series.str[: len(common_xfix) + 1]
----------------------------------------------------------------------
openai/validators.py:common_completion_suffix_validator  score=0.449
def common_completion_suffix_validator(df):
    """
    This validator will suggest to add a common suffix to the completion if one doesn't already exist in case of classification or conditional generation.
    """
    error_msg = None
    immediate_msg = None
    optional_msg = None
    optional_fn = None

    ft_type = infer_task_type(df)
----------------------------------------------------------------------

مثال دیگری از جستجو:

1res = search_functions(df, 'Command line interface for fine-tuning', n=1, n_lines=20)

نتیجه جستجو:

openai/cli.py:tools_register  score=0.391
def tools_register(parser):
    subparsers = parser.add_subparsers(
        title="Tools", help="Convenience client side tools"
    )

    def help(args):
        parser.print_help()

    parser.set_defaults(func=help)

    sub = subparsers.add_parser("fine_tunes.prepare_data")
    sub.add_argument(
        "-f",
        "--file",
        required=True,
        help="JSONL, JSON, CSV, TSV, TXT or XLSX file containing prompt-completion examples to be analyzed."
        "This should be the local file path.",
    )
    sub.add_argument(
        "-q",
----------------------------------------------------------------------