编程自动化01：使用提示词工程实现面向特定场景的稳定代码生成

常见的提示词工程的方法有：

• one-shot&few-shot: 通过输入一个或多个类似问题和问题答案，让模型参考学习 (论文地址《Language Models are Few-Shot Learners》)
• zero-shot-CoT: 简单来说就是提示词后面跟上 let's think step by step 或 请一步步进行推理并得出结论，让模型边分析边生成 (论文地址:《Large Language Models are Zero-Shot Reasoners》)
• few-shot-CoT：实在zero-shot-CoT的基础上加上几个类似的问题和对应的答案提供参考(论文地址：《Chain-of-Thought Prompting Elicits Reasoning in Large Language Models》)
• LEAST-TO-MOST PROMPTING（LtM提示法）： LtM提示方法提出的初衷是为了解决CoT提示方法泛化能力不足的问题——即通过人工编写的思维链提示样本可能并不能够很好的迁移到别的问题当中去，换而言之，就是解决问题的流程迁移能力不足，即泛化能力不够。而这种泛化能力不足则会导致“新的问题”无法使用“老的模板”进行解决。例如此前的第四个推理问题就是如此。那即然要找到更加普适的解决问题的流程会非常复杂，那能否“千人千面”让大模型自己找到解决当前问题的思维链呢？答案是肯定的，谷歌大脑基于这个思路开发了一种全新的提示流程，即先通过提示过程让模型找到解决该问题必须要分步解决哪几个问题，然后再通过依次解决这些问题来解决最原始的问题。
  不难看出整个提示过程会分为两个阶段进行，第一个阶段是自上而下的分解问题（Decompose Question into subquestion），第二个阶段是自下而上的依次解决问题（Sequentially Solve Subquestion），而整个依次回答问题的过程，其实就可以看成是CoT的过程，只不过LtM会要求模型根据每个不同的问题，单独生成解决问题的链路，以此做到解决问题流程的“千人千面”，从而能够更加精准的解决复杂推理问题。而整个过程问题的由少变多，则是LEAST-TO-MOST一词的来源。（论文地址：《LEAST-TO-MOST PROMPTING ENABLES COMPLEX REASONING IN LARGE LANGUAGE MODELS》）

最近我实践了这些提示词工程的技巧，让gpt4给自己编写工具来扩展他的能力，我花了两天时间把整个流程做到了自动化，下面两个分别为stage1和stage2的成果，stage1阶段，我仍需要思考如何打磨语言来稳定住gpt的行为，stage2我使用了few-shot-LtM的技巧教它拆解问题并添加了3组演示的例子，把思考提示词工程的步骤也自动化了，我只需要用自然语言提出需求，并且编写一些简单的测试用例。

stage1

In[ ]:

from functools import lru_cache
from typing import Any, Callable, Dict
from abc import ABC, abstractmethod
class AutoFunction(ABC):
    tested_dir = './tested'
    untest_dir = './untest'
    def __init__(self, function_name: str,
                 description: str,
                 use_proxy: bool = True,
                 use_cache: bool = True,
                 api_key: str = None, 
                 api_base: str = None,
                 model: str = None):
        self.api_key = api_key or os.environ['OPENAI_API_KEY']
        self.api_base = api_base or os.environ['OPENAI_API_BASE']
        self.api_model = model or os.environ['OPENAI_API_MODEL']
        self.use_proxy = use_proxy
        self.code = ''
        self.path = ''
        self.tested = False
        self.function_name = function_name
        self.function_name_inst = f'{function_name}__func__'
        self.func = (use_cache and self.load_tested_function()) or self.generate_func(description)
        self.proxy_func = self.proxy(self.func)
        self.docs = self.func.__doc__
    def __call__(self, *args, **kwargs) -> str:
        return self.func(*args, **kwargs) if not self.use_proxy else self.proxy_func(*args, **kwargs)

    def __str__(self) -> str:
        return f"{self.__class__.__name__}({self.function_name}) {{\n\ninstance:\n{self.func}\n\ndescription:\n{self.docs}\n\npath:\n{self.path}\n\ncode:\n{self.code}\n}}"

    def __repr__(self) -> str:
        return str(self)

    @abstractmethod
    def generate_code(self, description) -> str:
        """
        使用ai生成代码
        """
        ...

    def generate_func(self, description: str) -> Callable:
        raw_code = self.generate_code(description)
        self.code = self.extract_code(raw_code)
        # 为了避免文件中定义的函数名和AutoFunction对象重名时后者被前者替换
        code_str = self.replace_func_name(self.code, self.function_name_inst)
        self.path = f'./{self.untest_dir}/{self.function_name}_module.py'
        self.store()
        exec(code_str, globals())
        return globals()[self.function_name_inst]

    def extract_code(self, code_str: str) -> str:
        import re
        if "```" in code_str:
            st = code_str.find('```python') + len("```python") + 1
            ed = code_str.rfind('```')
            code_str = code_str[st: ed]
        return code_str

    def replace_func_name(self, original: str=None, func_name: str=None) -> str:
        import re
        return re.sub(r"(def\s+)\w+", r"\1" + (func_name or self.function_name_inst), original or self.code)

    def load_tested_function(self) -> Callable | None:
        func = None
        import os
        if os.path.exists(f'./{self.tested_dir}/{self.function_name}_module.py'):
            self.path = f'./{self.tested_dir}/{self.function_name}_module.py'
            with open(self.path, 'r', encoding='utf8') as f:
                self.code = self.extract_code(f.read())
                # 为了避免文件中定义的函数名和AutoFunction对象重名时后者被前者替换
                code_str = self.replace_func_name(self.code, self.function_name_inst)
                exec(code_str, globals())
                func = globals()[self.function_name_inst]
                self.tested = True
        return func

    def store(self, temp_path: str = None) -> bool:
        try:
            with open(temp_path or self.path, 'w', encoding='utf8') as f:
                f.write(self.code)
        except Exception as e:
            print(f'function store error {self}: \n', e)
            return False
        return True

    @classmethod
    def proxy(cls, func: Callable) -> Callable:
        import os
        def wappered_func(*args, **kwargs):
            old_http_proxy = os.environ.get('http_proxy', None)
            old_https_proxy = os.environ.get('https_proxy', None)
            os.environ['http_proxy'] = 'http://localhost:7890'
            os.environ['https_proxy'] = 'http://localhost:7890'
            ret = func(*args, **kwargs)
            if old_http_proxy:
                os.environ['http_proxy'] = old_http_proxy
                os.environ['https_proxy'] = old_https_proxy
            else:
                del os.environ['http_proxy']
                del os.environ['https_proxy']
            return ret
        wappered_func.__doc__ = func.__doc__
        wappered_func.__name__ = func.__name__
        return wappered_func

    def test(self, test_func: Callable[[Callable], str]) -> bool:
        ret = test_func(self)
        print(test_func.__doc__, ret and '通过' or '未通过')
        self.tested |= ret
        return self

    def testEnd(self):
        if not self.tested or self.tested_dir in self.path : return self
        import shutil
        shutil.move(self.path, f"{self.tested_dir}/{self.function_name}_module.py")
        self.path = f"{self.tested_dir}/{self.function_name}_module.py"
        print(self,'已测试')
        return self

    @lru_cache(maxsize=None)
    def get_properties(self) -> Dict[str, str]:
        """
        生成用于接入openaiApi functions参数所需的对象
        """
        import json,openai
        openai.api_key = self.api_key
        openai.api_base = self.api_base
        resp = openai.ChatCompletion.create(
            model=self.api_model,
            messages=[
                {'role': 'system', 'content': '请按照以下规则为下面的函数创建一个json对象，遵循以下规则:\
                    1.包含name键，值为该函数名;\
                    2.包含 description键，值为该函数的说明，这里需要说明哪些参数为必填;\
                    3.包含paramter键，值为 json schema格式的函数的信息,discription中需要尽可能详细的描述，对于参数的描述需要指明是否为必填参数;\
                    4.不要输出任何多余的符号;'},
                {'role': 'user', 'content': self.function_name + self.docs}
            ],
            temperature=0
        )
        return json.loads(resp.choices[0].message.content)

In[ ]:

class AutoGmailApiFunction(AutoFunction):
    def generate_code(self, description):
        import openai,os
        openai.api_key = self.api_key
        openai.api_base = self.api_base
        resp = openai.ChatCompletion.create(
            model=self.api_model,
            temperature=0,
            # 给gpt举了一个样例(few-shot)
            messages=[
                {'role': 'system', 'content': 'Gmail API授权已经完成，授权文件为本地文件token.json，你只需要生成相关的代码和文档注释而不需要添加任何多余的描述、旁白、前后缀、修饰、说明之类'},
                {'role': 'user', 'content': f'请帮我编写一个python函数，函数要求如下:\
                1.函数的功能为查看我的Gmail邮箱中最后一封邮件信息\
                2.函数参数userId，userId是字符串参数，默认情况下取值为"me"，表示查看我的邮件；\
                3.函数返回结果是一个包含最后一封邮件信息的对象，返回结果本身必须是一个json格式对象； \
                4.请将全部功能封装在一个函数内；\
                5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: -From: 发信人 -Date: 时间 -Subject: 主题 -Body: 内容；\
                6.需要处理参数的特殊输入'},
                {'role': 'assistant', 'content' : example_assistant_code},
                {'role': 'user', 'content': f'请帮我编写一个python函数，函数要求如下:\
                {description}'},
            ]
        )
        return resp.choices[0].message.content

In[ ]:

def test_case1_get_current_time(get_current_time):
    """
    用例1：检查返回值是否为json
    """
    import json
    try:
        json.loads(get_current_time())
    except ValueError as e:
        return False
    return True

def test_case2_get_current_time(get_current_time):
    """
    用例2：检查返回时间是否为当前时间
    """
    import json, datetime
    result = get_current_time()
    data = json.loads(result)
    expected = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    for _,real in data.items():
        return abs((datetime.datetime.strptime(real, '%Y-%m-%d %H:%M:%S') - datetime.datetime.strptime(expected, '%Y-%m-%d %H:%M:%S')).seconds) <= 1

In[ ]:

get_current_time = AutoGmailApiFunction('get_current_time',
    '1.函数的功能为查看当前时间\
    2.函数的参数为空\
    3.函数返回结果是一个包含当前时间信息的json对象,时间表示为%Y-%m-%d %H:%M:%S \
    4.请将全部功能封装在一个函数内，只需要一个函数定义而不需要任何调用代码；\
    5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: -From: 发信人 -Date: 时间 -Subject: 主题 -Body: 内容；\
    6.需要处理参数的特殊输入', use_cache=True).test(test_case1_get_current_time).test(test_case2_get_current_time).testEnd()

Out[ ]:

    用例1：检查返回值是否为字符串
     通过

    用例2：检查返回时间是否为当前时间
     通过
AutoGoogleApiFunction(get_current_time) {

instrance:
<function get_current_time__func__ at 0x0000021F20216680>

description:

    函数功能：获取当前时间
    参数：无
    返回结果：一个包含当前时间信息的json对象，时间表示为%Y-%m-%d %H:%M:%S
    - CurrentTime: 当前时间

path:
./tested/get_current_time_module.py

code:
import datetime
import json

def get_current_time():
    """
    函数功能：获取当前时间
    参数：无
    返回结果：一个包含当前时间信息的json对象，时间表示为%Y-%m-%d %H:%M:%S
    - CurrentTime: 当前时间
    """
    current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    return json.dumps({"CurrentTime": current_time})

} 已测试

In[ ]:

print(get_current_time.docs)
get_current_time()

Out[ ]:

    函数功能：获取当前时间
    参数：无
    返回结果：一个包含当前时间信息的json对象，时间表示为%Y-%m-%d %H:%M:%S
    - CurrentTime: 当前时间

'{"CurrentTime": "2023-12-29 16:25:34"}'

In[ ]:

def test_case1_get_last_n_emails(get_last_n_emails):
    """
    用例1：检查返回值是否为json
    """
    import json
    try:
        json.loads(get_last_n_emails(1))
    except ValueError as e:
        return False
    return True

def test_case2_get_last_n_emails(get_last_n_emails):
    """
    用例2：检查返回邮件的数量是否正确
    """
    import json
    result = get_last_n_emails(5)
    return len(json.loads(result)) == 5

In[ ]:

get_last_n_emails = AutoGmailApiFunction('get_last_n_emails',
    '1.函数的功能为获取最后n条邮件信息\
    2.函数的参数为n、userId,其中n为整型表示要查看的邮件数量，userId表示要查看的用户，默认值为"me"表示我的账号\
    3.函数返回结果是一个包含n封邮件信息的列表，每封邮件是一个对象，邮件信息只需要包含少量的必要信息，包括邮件主题，内容(内容超过50字，就省略中间部分)，发件人，时间等，返回结果本身必须是一个json格式列表； \
    4.请将全部功能封装在一个函数内；\
    5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: -From: 发信人 -Date: 时间 -Subject: 主题 -Body: 内容；\
    6.需要处理参数的特殊输入').test(test_case1_get_last_n_emails).test(test_case2_get_last_n_emails).testEnd()

Out[ ]:

    用例1：检查返回值是否为字符串
     通过

    用例2：检查返回邮件的数量是否正确
     通过

In[ ]:

get_email_count = AutoGmailApiFunction('get_email_count','请帮我编写一个python函数，函数要求如下:\
1.函数的功能为我的Gmail邮箱中总共有多少封邮件；\
2.函数参数userId，userId是字符串参数，默认情况下取值为"me"，表示查看我的邮件；\
3.函数返回结果是一个包含当前邮件数的json对象；\
4.请将全部功能封装在一个函数内，只需要一个函数定义而不需要任何调用代码；\
5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: - Count: 邮件数；\
6.需要处理参数的特殊输入；')

In[ ]:

get_email_count()

Out[ ]:

'{"Count": 100}'

In[ ]:

get_email_count.test(test_case1_get_current_time)

Out[ ]:

    用例1：检查返回值是否为字符串
     通过
AutoGoogleApiFunction(get_email_count) {

instrance:
<function get_email_count__func__ at 0x0000021F47F9A3B0>

description:

    函数功能：查看Gmail邮箱中的邮件总数
    参数：
    userId：字符串参数，默认情况下取值为'me'，表示查看我的邮件，非必填
    返回结果：一个包含当前邮件数的json对象
    - Count: 邮件数

path:
././untest/get_email_count_module.py

code:
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
import json

# If modifying these SCOPES, delete the file token.json.
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']

def get_email_count(userId='me'):
    """
    函数功能：查看Gmail邮箱中的邮件总数
    参数：
    userId：字符串参数，默认情况下取值为'me'，表示查看我的邮件，非必填
    返回结果：一个包含当前邮件数的json对象
    - Count: 邮件数
    """
    creds = Credentials.from_authorized_user_file('token.json', SCOPES)
    # Call the Gmail API
    service = build('gmail', 'v1', credentials=creds)
    results = service.users().messages().list(userId=userId).execute()
    messages = results.get('messages', [])
    return json.dumps({"Count": len(messages)})

}

In[ ]:

get_current_time.get_properties()

Out[ ]:

{'name': 'get_current_time',
 'description': '这个函数的功能是获取当前时间，没有必填参数。',
 'parameter': {'type': 'object', 'properties': {}, 'required': []}}

In[ ]:

get_last_n_emails.get_properties()

Out[ ]:

{'name': 'get_last_n_emails',
 'description': "获取最后n条邮件的信息。其中参数n为必填，表示要查看的邮件数量；参数userId为非必填，表示要查看的用户，默认值为'me'表示我的账号。",
 'parameter': {'type': 'object',
  'properties': {'n': {'type': 'integer', 'description': '要查看的邮件数量，必填参数'},
   'userId': {'type': 'string', 'description': "要查看的用户，默认值为'me'表示我的账号，非必填参数"}},
  'required': ['n']}}

In[ ]:

functions = [get_current_time.get_properties(),get_last_n_emails.get_properties()]

In[ ]:

def chat_with_func_call(question, functions = None):
    """
    通过ChatGPT与用户进行对话
    Args:
        question(string): 用户输入的问题
        functions(List[Callbale]): 可供gpt调用的函数
    Returns:
        string: ChatGPT的回答
    """
    import json,openai
    openai.api_key = os.environ['OPENAI_API_KEY']
    openai.api_base = os.environ['OPENAI_API_BASE']
    messages = []
    messages.append({'role': 'user', 'content': question})
    while True:
        print('history',messages)
        resp = openai.ChatCompletion.create(
            model=os.environ['OPENAI_API_MODEL'],
            messages=messages,
            functions=functions,
            function_call='auto',
            temperature=0
        )
        if resp.choices[0].finish_reason == 'function_call':
            print('function_calling',resp.choices[0].message)
            arguments = json.loads(resp.choices[0].message.function_call.arguments)
            answer = globals()[resp.choices[0].message.function_call.name](**arguments)
            messages.append({'role': 'function', 'name': resp.choices[0].message.function_call.name, 'content': answer})
        else:
            return resp.choices[0].message.content

In[ ]:

chat_with_func_call('昨天12点后我有收到任何人的邮件吗',functions)

Out[ ]:

history [{'role': 'user', 'content': '昨天12点后我有收到任何人的邮件吗'}]
function_calling {
  "role": "assistant",
  "content": "\u4e3a\u4e86\u56de\u7b54\u60a8\u7684\u95ee\u9898\uff0c\u6211\u9700\u8981\u5148\u83b7\u53d6\u5f53\u524d\u65f6\u95f4\uff0c\u7136\u540e\u83b7\u53d6\u60a8\u6700\u8fd1\u7684\u90ae\u4ef6\uff0c\u7136\u540e\u8fdb\u884c\u6bd4\u8f83\u3002\u8ba9\u6211\u6765\u4e3a\u60a8\u505a\u8fd9\u4e2a\u3002",
  "function_call": {
    "name": "get_current_time",
    "arguments": "{}"
  }
}
history [{'role': 'user', 'content': '昨天12点后我有收到任何人的邮件吗'}, {'role': 'function', 'name': 'get_current_time', 'content': '{"CurrentTime": "2023-12-29 02:03:30"}'}]
function_calling {
  "role": "assistant",
  "content": null,
  "function_call": {
    "name": "get_last_n_emails",
    "arguments": "{\n  \"n\": 50\n}"
  }
}
history [{'role': 'user', 'content': '昨天12点后我有收到任何人的邮件吗'}, {'role': 'function', 'name': 'get_current_time', 'content': '{"CurrentTime": "2023-12-29 02:03:30"}'}, {'role': 'function', 'name': 'get_last_n_emails', 'content': '[{"From": "Google Developer Profile <googledev-noreply@google.com>", "Date": "Wed, 27 Dec 2023 01:25:04 -0800", "Subject": "Welcome to your Google Developer Profile!", "Body": "Compl...\\u200c \\u200c \\u200c"}, {"From": "Steam <noreply@steampowered.com>", "Date": "Wed, 20 Dec 2023 09:17:31 -0800", "Subject": "\\u60a8\\u7684 2023 \\u5e74 Steam \\u5e74\\u5ea6\\u56de\\u987e\\u5df2\\u51c6\\u5907\\u5c31\\u7eea\\uff01", "Body": "\\u60a8\\u7684 20...\\u200c \\u200c \\u200c"}, {"From": "ClickHouse Cloud Team <clickhouse.marketing@clickhouse.com>", "Date": "Tue, 19 Dec 2023 10:03:01 -0600 (CST)", "Subject": "Celebrating a year of growth \\ud83c\\udf89", "Body": "As th...\\u200c \\u200c \\u200c"}, {"From": "Discord <notifications@discord.com>", "Date": "Tue, 19 Dec 2023 15:56:46 +0000 (UTC)", "Subject": "davinci798\\u5728\\u81ea\\u7531AI\\u9635\\u7ebf \\u4e2d\\u63d0\\u53ca\\u4e86\\u60a8", "Body": "\\u60a8\\u6709 1 ...\\u5ba2 \\u2022 @"}, {"From": "Poe <product-updates@poe.com>", "Date": "Thu, 14 Dec 2023 20:38:14 +0000", "Subject": "Knowledge bases, DALL-E-3, Playground v2, international expansion, and more", "Body": "Hi ev... bots"}, {"From": "\\"Excalidraw+\\" <newsletter@excalidraw.com>", "Date": "Wed, 06 Dec 2023 11:27:33 +0000 (UTC)", "Subject": "Generate diagrams with AI \\ud83e\\udd16", "Body": "Text ...u can"}, {"From": "Kaggle <noreply@kaggle.com>", "Date": "Wed, 22 Nov 2023 10:24:50 -0800", "Subject": "You\\u2019re officially a Kaggler. Now what?", "Body": "Hi he...nt to"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 21 Nov 2023 12:30:29 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "Battl...\\u5e33\\u865f\\u7684\\u96fb\\u5b50"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 21 Nov 2023 12:30:27 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "Google <no-reply@accounts.google.com>", "Date": "Fri, 10 Nov 2023 15:23:35 GMT", "Subject": "\\u5b89\\u5168\\u63d0\\u9192", "Body": "\\u5728 Win...s \\u6211\\u4eec\\u5411"}, {"From": "Nick Peart <nickpeart@clickhouse.com>", "Date": "Thu, 9 Nov 2023 03:42:36 -0600 (CST)", "Subject": "ClickHouse at AWS re:Invent 2023", "Body": "Hi du...ll be"}, {"From": "\\"pixiv\\u4e8b\\u52a1\\u5c40\\" <no-reply@pixiv.net>", "Date": "Sun, 05 Nov 2023 14:33:20 +0000 (UTC)", "Subject": "[pixiv] \\u65b0\\u767b\\u5f55\\u901a\\u77e5\\uff08\\u767b\\u5f55\\u5730\\u70b9\\uff1a\\u65b0\\u52a0\\u5761\\uff09", "Body": "\\u4eb2\\u7231\\u7684du...\\u4eba\\u767b\\u5f55\\u611f\\u8c22"}, {"From": "Poe <product-updates@poe.com>", "Date": "Fri, 27 Oct 2023 17:38:12 +0000", "Subject": "Introducing creator monetization on Poe", "Body": "Hi Ev...ryone"}, {"From": "Discord <noreply@discord.com>", "Date": "Sun, 22 Oct 2023 02:45:36 +0000 (UTC)", "Subject": "\\u9a8c\\u8bc1\\u6765\\u81ea\\u65b0\\u5730\\u70b9\\u7684Discord\\u767b\\u5f55", "Body": "hedu_... \\u53cd\\u9988\\u7f51\\u9875"}, {"From": "Discord <noreply@discord.com>", "Date": "Sat, 21 Oct 2023 09:48:34 +0000 (UTC)", "Subject": "\\u9a8c\\u8bc1\\u6765\\u81ea\\u65b0\\u5730\\u70b9\\u7684Discord\\u767b\\u5f55", "Body": "hedu_...\\u9988\\u7f51\\u9875 \\u8ba9"}, {"From": "Discord <noreply@discord.com>", "Date": "Sat, 21 Oct 2023 09:21:05 +0000 (UTC)", "Subject": "\\u9a8c\\u8bc1 Discord \\u7684\\u7535\\u5b50\\u90ae\\u4ef6\\u5730\\u5740", "Body": "Hey h...\\u6211\\u4eec\\u7684\\u652f\\u6301"}, {"From": "ClickHouse Learning Team <learn@clickhouse.com>", "Date": "Mon, 16 Oct 2023 18:15:13 -0500 (CDT)", "Subject": "Master ClickHouse: Join our FREE Fundamentals Training", "Body": "Are y...\\u200c \\u200c \\u200c"}, {"From": "Poe <product-updates@poe.com>", "Date": "Thu, 05 Oct 2023 21:08:12 +0000", "Subject": "Image generating bots, web search, API v2, and Android updates!", "Body": "Hi ev...ty in"}, {"From": "Nicholas Kontopoulos <teamtwilio@team.twilio.com>", "Date": "Thu, 05 Oct 2023 19:37:47 +0000 (UTC)", "Subject": "David Rowan headlines SIGNAL Singapore", "Body": "Hear ...ers a"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 26 Sep 2023 13:43:20 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 26 Sep 2023 13:41:09 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "Battl...\\u5e33\\u865f\\u7684\\u96fb\\u5b50"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 26 Sep 2023 13:41:09 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "Patreon <no-reply@patreon.com>", "Date": "Mon, 25 Sep 2023 18:05:35 +0000", "Subject": "\\u6709\\u5173 Patreon \\u9690\\u79c1\\u653f\\u7b56\\u7684\\u66f4\\u65b0", "Body": "Patre...\\u7684\\u4e3b\\u8981\\u987b\\u77e5"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Sun, 24 Sep 2023 13:44:53 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "Battl...\\u5e33\\u865f\\u7684\\u96fb\\u5b50"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Sun, 24 Sep 2023 13:44:54 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "Ivan Zhao <ivan@mail.notion.so>", "Date": "Fri, 22 Sep 2023 17:22:59 +0000", "Subject": "Notion 2.33", "Body": "Impro...\\u034f \\u034f \\u034f"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Mon, 18 Sep 2023 13:47:09 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Mon, 18 Sep 2023 13:45:06 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "Battl...\\u5e33\\u865f\\u7684\\u96fb\\u5b50"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Mon, 18 Sep 2023 13:45:03 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Mon, 11 Sep 2023 13:10:32 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Mon, 11 Sep 2023 13:10:27 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "Battl...\\u5e33\\u865f\\u7684\\u96fb\\u5b50"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 5 Sep 2023 13:30:13 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 5 Sep 2023 13:29:45 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Tue, 5 Sep 2023 13:29:43 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "Battl...\\u5e33\\u865f\\u7684\\u96fb\\u5b50"}, {"From": "Poe <product-updates@poe.com>", "Date": "Sun, 27 Aug 2023 17:23:15 +0000", "Subject": "Poe updates: Threading, Bot search, Llama 2, custom temperature, Poe for Mac, and Poe in Japanese", "Body": "Hi ev... Your"}, {"From": "\\"Excalidraw+\\" <newsletter@excalidraw.com>", "Date": "Thu, 24 Aug 2023 13:23:15 +0000 (UTC)", "Subject": "\\ud83d\\ude80 Read-Only Presentations: The Next Logical Step in Visual Storytelling", "Body": "Easil...th we"}, {"From": "Google Accounts Team <no-reply@accounts.google.com>", "Date": "Mon, 07 Aug 2023 19:14:25 -0700", "Subject": "\\u300a\\u95f2\\u7f6e Google \\u5e10\\u53f7\\u653f\\u7b56\\u300b\\u66f4\\u65b0\\u4e86", "Body": "\\u4e3a\\u5207\\u5b9e\\u4fdd\\u969c...\\u548c\\u670d\\u52a1\\u4e2d\\u7684"}, {"From": "Ivan Zhao <ivan@mail.notion.so>", "Date": "Thu, 03 Aug 2023 17:10:40 +0000", "Subject": "Notion 2.32", "Body": "New i...\\u034f \\u034f \\u034f"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Thu, 20 Jul 2023 15:27:09 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Thu, 20 Jul 2023 15:25:45 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...\\u5230\\u9019\\u5c01\\u96fb\\u5b50"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Thu, 20 Jul 2023 15:25:29 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Excalidraw+\\" <newsletter@excalidraw.com>", "Date": "Wed, 12 Jul 2023 17:25:18 +0000 (UTC)", "Subject": "\\ud83d\\udcfd Presentations. Our latest feature.", "Body": "Say i...illed"}, {"From": "Poe <product-updates@poe.com>", "Date": "Wed, 12 Jul 2023 14:16:22 +0000 (UTC)", "Subject": "Poe updates: File uploading, URL retrieval, Claude 2, and more!", "Body": "Hi ev...f new"}, {"From": "Brilliant <learn@account.brilliant.org>", "Date": "Wed, 05 Jul 2023 23:09:52 +0000", "Subject": "Welcome to Brilliant!", "Body": "Learn...solve"}, {"From": "Ivan Zhao <ivan@mail.notion.so>", "Date": "Thu, 22 Jun 2023 17:56:01 +0000", "Subject": "Notion 2.31: New Template Gallery", "Body": "Redes...\\u200c \\u200c \\u200c"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Thu, 22 Jun 2023 13:26:40 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Thu, 22 Jun 2023 13:25:45 +0000 (UTC)", "Subject": "\\u5354\\u52a9\\u6211\\u5011\\u900f\\u904e\\u5b89\\u5168\\u6aa2\\u67e5\\u4f86\\u7dad\\u8b77\\u4f60\\u7684 Battle.net \\u5e33\\u865f\\u5b89\\u5168", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...\\u5230\\u9019\\u5c01\\u96fb\\u5b50"}, {"From": "\\"Battle.net\\" <noreply@battle.net>", "Date": "Thu, 22 Jun 2023 13:25:14 +0000 (UTC)", "Subject": "Battle.net \\u5e33\\u865f\\u9a57\\u8b49", "Body": "\\u60a8\\u597d\\uff01\\u5982\\u679c...ZZARD"}, {"From": "Figma <support@figma.com>", "Date": "Wed, 21 Jun 2023 20:38:48 -0500 (CDT)", "Subject": "Everything we announced at Config 2023", "Body": "Disco...\\u034f \\u034f \\u034f"}, {"From": "Google <no-reply@accounts.google.com>", "Date": "Wed, 14 Jun 2023 07:18:01 GMT", "Subject": "\\u5b89\\u5168\\u63d0\\u9192", "Body": "\\u5728 Win...s \\u6211\\u4eec\\u5411"}]'}]
'对不起，我没有找到昨天12点后收到的邮件。'

stage2

In[ ]:

CD_system_message = '为了更好编写满足用户需求的python函数，我们需要先识别用户需求中的变量，以作为python函数的参数。需要注意的是，当前编写的函数中涉及到的功能，优先考虑调用Gmail Api来完成。'
input_message = '请帮我编写一个查看Gmail邮箱中最后一封邮件的函数。函数名为 get_last_email'
CD_assistant_message = '当前需求中可以作为函数参数的是：1.查看谁的邮箱。'
CM_system_message = 'Gmail Api授权已经完成，授权文件为本地文件token.json。编写的函数说明中需要指定1.函数功能；2.函数参数；3.函数的返回结果；4.将全部功能封装在一个函数内；5.为函数添加说明文档；'
CM_user_message = input_message+CD_assistant_message
CM_assistant_message = \
'请帮我编写一个名为get_last_email的python函数，函数要求如下:\
1.函数的功能为查看我的Gmail邮箱中最后一封邮件信息；\
2.函数参数userId，userId是字符串参数，默认情况下取值为"me"，表示查看我的邮件；\
3.函数返回结果是一个包含最后一封邮件信息的对象，返回结果本身必须是一个json格式对象； \
4.请将全部功能封装在一个函数内；\
5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: -From: 发信人 -Date: 时间 -Subject: 主题 -Body: 内容；\
6.需要处理参数的特殊输入；'
code_system_message = 'Gmail Api授权已经完成，授权文件为本地文件token.json，你只需要生成相关的代码和文档注释而不需要添加任何多余的描述、旁白、前后缀、修饰、说明之类'
code_user_message = CM_assistant_message
# description_assistant_message = example_assistant_code

import json
with open('./prompt/get_last_email/prompts.json', 'w+', encoding='utf-8') as f:
    f.write(json.dumps({'input_message':input_message,
                        'CD_assistant_message':CD_assistant_message,
                        'CM_user_message': CM_user_message,
                        'CM_assistant_message': CM_assistant_message,
                        'code_user_message': code_user_message,
                        }, ensure_ascii=False, indent=4))

with open('./prompt/system_prompts.json', 'w+', encoding='utf-8') as f:
    f.write(json.dumps({'CD_system_message':CD_system_message,
                        'CM_system_message':CM_system_message,
                        'code_system_message': code_system_message
                        }, ensure_ascii=False, indent=4))

In[ ]:

input_message = '请帮我编写一个查看当前时间的函数。函数名为 get_current_time'
CD_assistant_message = '当前需求中可以作为函数参数的是：无参数'
CM_user_message = input_message+CD_assistant_message
CM_assistant_message = \
'请帮我编写一个名为get_current_time的python函数，函数要求如下:\
1.函数的功能为查看当前时间；\
2.函数的参数为空；\
3.函数返回结果是一个包含当前时间信息的json对象,时间表示为%Y-%m-%d %H:%M:%S；\
4.请将全部功能封装在一个函数内，只需要一个函数定义而不需要任何调用代码；\
5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: - CurrentTime: 当前时间；\
6.需要处理参数的特殊输入；'
import json
with open('./prompt/get_current_time/prompts.json', 'w+', encoding='utf-8') as f:
    f.write(json.dumps({'input_message':input_message,
                        'CD_assistant_message':CD_assistant_message,
                        'CM_user_message': CM_user_message,
                        'CM_assistant_message': CM_assistant_message,
                        'code_user_message': code_user_message,
                        }, ensure_ascii=False, indent=4))

In[ ]:

input_message = '请帮我编写一个查看现在邮箱中有多少封邮件的函数。函数名为 get_email_count'
CD_assistant_message = '当前需求中可以作为函数参数的是：1.查看谁的邮箱'
CM_user_message = input_message+CD_assistant_message
CM_assistant_message = \
'请帮我编写一个名为get_email_count的python函数，函数要求如下:\
1.函数的功能为我的Gmail邮箱中总共有多少封邮件；\
2.函数参数userId，userId是字符串参数，默认情况下取值为"me"，表示查看我的邮件；\
3.函数返回结果是一个包含当前邮件数的json对象；\
4.请将全部功能封装在一个函数内，只需要一个函数定义而不需要任何调用代码；\
5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: - Count: 邮件数；\
6.需要处理参数的特殊输入；'
import json
with open('./prompt/get_email_count/prompts.json', 'w+', encoding='utf-8') as f:
    f.write(json.dumps({'input_message':input_message,
                        'CD_assistant_message':CD_assistant_message,
                        'CM_user_message': CM_user_message,
                        'CM_assistant_message': CM_assistant_message,
                        'code_user_message': code_user_message,
                        }, ensure_ascii=False, indent=4))

In[ ]:

from functools import lru_cache
from typing import Any, Callable, Dict
from abc import ABC, abstractmethod
class AutoFunction(ABC):
    tested_dir = './tested'
    untest_dir = './untest'
    def __init__(self, function_name: str,
                 description: str,
                 use_proxy: bool = True,
                 use_cache: bool = True,
                 api_key: str = None, 
                 api_base: str = None,
                 model: str = None):
        import os
        self.api_key = api_key or os.environ['OPENAI_API_KEY']
        self.api_base = api_base or os.environ['OPENAI_API_BASE']
        self.api_model = model or os.environ['OPENAI_API_MODEL']
        self.use_proxy = use_proxy
        self.code = ''
        self.path = ''
        self.tested = False
        self.function_name = function_name
        self.function_name_inst = f'{function_name}__func__'
        self.func = (use_cache and self.load_tested_function()) or self.generate_func(description)
        self.proxy_func = self.proxy(self.func)
        self.docs = self.func.__doc__
    def __call__(self, *args, **kwargs) -> str:
        return self.func(*args, **kwargs) if not self.use_proxy else self.proxy_func(*args, **kwargs)

    def __str__(self) -> str:
        return f"{self.__class__.__name__}({self.function_name}) {{\n\ninstance:\n{self.func}\n\ndescription:\n{self.docs}\n\npath:\n{self.path}\n\ncode:\n{self.code}\n}}"

    def __repr__(self) -> str:
        return str(self)

    @abstractmethod
    def generate_code(self, description) -> str:
        """
        使用ai生成代码
        """
        ...

    def generate_func(self, description: str) -> Callable:
        raw_code = self.generate_code(description)
        self.code = self.extract_code(raw_code)
        # 为了避免文件中定义的函数名和AutoFunction对象重名时后者被前者替换
        code_str = self.replace_func_name(self.code, self.function_name_inst)
        self.path = f'./{self.untest_dir}/{self.function_name}_module.py'
        self.store()
        exec(code_str, globals())
        return globals()[self.function_name_inst]

    def extract_code(self, code_str: str) -> str:
        import re
        if "```" in code_str:
            st = code_str.find('```python') + len("```python") + 1
            ed = code_str.rfind('```')
            code_str = code_str[st: ed]
        return code_str

    def replace_func_name(self, original: str=None, func_name: str=None) -> str:
        import re
        return re.sub(r"(def\s+)\w+", r"\1" + (func_name or self.function_name_inst), original or self.code)

    def load_tested_function(self) -> Callable | None:
        func = None
        import os
        if os.path.exists(f'./{self.tested_dir}/{self.function_name}_module.py'):
            self.path = f'./{self.tested_dir}/{self.function_name}_module.py'
            with open(self.path, 'r', encoding='utf8') as f:
                self.code = self.extract_code(f.read())
                # 为了避免文件中定义的函数名和AutoFunction对象重名时后者被前者替换
                code_str = self.replace_func_name(self.code, self.function_name_inst)
                exec(code_str, globals())
                func = globals()[self.function_name_inst]
                self.tested = True
        return func

    def store(self, temp_path: str = None) -> bool:
        try:
            with open(temp_path or self.path, 'w', encoding='utf8') as f:
                f.write(self.code)
        except Exception as e:
            print(f'function store error {self}: \n', e)
            return False
        return True

    @classmethod
    def proxy(cls, func: Callable) -> Callable:
        import os
        def wappered_func(*args, **kwargs):
            old_http_proxy = os.environ.get('http_proxy', None)
            old_https_proxy = os.environ.get('https_proxy', None)
            os.environ['http_proxy'] = 'http://localhost:7890'
            os.environ['https_proxy'] = 'http://localhost:7890'
            ret = func(*args, **kwargs)
            if old_http_proxy:
                os.environ['http_proxy'] = old_http_proxy
                os.environ['https_proxy'] = old_https_proxy
            else:
                del os.environ['http_proxy']
                del os.environ['https_proxy']
            return ret
        wappered_func.__doc__ = func.__doc__
        wappered_func.__name__ = func.__name__
        return wappered_func

    def test(self, test_func: Callable[[Callable], str]) -> bool:
        ret = test_func(self)
        print(test_func.__doc__, ret and '通过' or '未通过')
        self.tested |= ret
        return self

    def testEnd(self):
        if not self.tested or self.tested_dir in self.path : return self
        import shutil
        shutil.move(self.path, f"{self.tested_dir}/{self.function_name}_module.py")
        self.path = f"{self.tested_dir}/{self.function_name}_module.py"
        print(self,'已测试')
        return self

    @lru_cache(maxsize=None)
    def get_properties(self) -> Dict[str, str]:
        """
        生成用于接入openaiApi functions参数所需的对象
        """
        import json,openai
        openai.api_key = self.api_key
        openai.api_base = self.api_base
        resp = openai.ChatCompletion.create(
            model=self.api_model,
            messages=[
                {'role': 'system', 'content': '请按照以下规则为下面的函数创建一个json对象，遵循以下规则:\
                    1.包含name键，值为该函数名;\
                    2.包含 description键，值为该函数的说明，这里需要说明哪些参数为必填;\
                    3.包含paramter键，值为 json schema格式的函数的信息,discription中需要尽可能详细的描述，对于参数的描述需要指明是否为必填参数;\
                    4.不要输出任何多余的符号;'},
                {'role': 'user', 'content': self.function_name + self.docs}
            ],
            temperature=0
        )
        return json.loads(resp.choices[0].message.content)

In[ ]:

from typing import List, Dict
class AutoGmailApiFunctionAdvance(AutoFunction):
    prompt_dir = './prompt'
    def __init__(self, function_name: str, description: str, use_proxy: bool = True, use_cache: bool = True, api_key: str = None, api_base: str = None, model: str = None):
        self.system_prompts = {}
        self.example_cases = []
        import os,json
        with open(os.path.join(self.prompt_dir, 'system_prompts.json'), 'r', encoding='utf8') as f:
            self.system_prompts = json.loads(f.read())
        self.example_cases = []
        for example_case in [os.path.join(self.prompt_dir, name) for name in os.listdir(self.prompt_dir) if os.path.isdir(os.path.join(self.prompt_dir, name))]:
            with open(os.path.join(example_case, 'prompts.json'), 'r', encoding='utf-8') as f:
                self.example_cases.append(json.loads(f.read()))
            with open(os.path.join(example_case, 'code.py'), 'r', encoding='utf-8') as f:
                self.example_cases[-1]['code'] = f.read()
        super().__init__(function_name, description, use_proxy, use_cache, api_key, api_base, model)

    def get_CD_messages(self) -> List[Dict]:
        message = []
        message.append({'role': 'system', 'content': self.system_prompts['CD_system_message']})
        for example_case in self.example_cases:
            message.append({'role': 'user', 'content': example_case['input_message']})
            message.append({'role': 'assistant', 'content': example_case['CD_assistant_message']})
        return message

    def get_CM_messages(self) -> List[Dict]:
        message = []
        message.append({'role': 'system', 'content': self.system_prompts['CM_system_message']})
        for example_case in self.example_cases:
            message.append({'role': 'user', 'content': example_case['CM_user_message']})
            message.append({'role': 'assistant', 'content': example_case['CM_assistant_message']})
        return message

    def get_code_messages(self) -> List[Dict]:
        message = []
        message.append({'role': 'system', 'content': self.system_prompts['code_system_message']})
        for example_case in self.example_cases:
            message.append({'role': 'user', 'content': example_case['code_user_message']})
            message.append({'role': 'assistant', 'content': example_case['code']})
        return message

    def generate_CD_message(self, description) -> str:
        messages = self.get_CD_messages()
        import openai,os
        openai.api_key = self.api_key
        openai.api_base = self.api_base
        resp = openai.ChatCompletion.create(
            model=self.api_model,
            temperature=0,
            messages=[
                *messages,
                {'role': 'user', 'content': description},
            ]
        )
        return f"{description} {resp.choices[0].message.content}"

    def generate_CM_message(self, description) -> str:
        messages = self.get_CM_messages()
        import openai,os
        openai.api_key = self.api_key
        openai.api_base = self.api_base
        resp = openai.ChatCompletion.create(
            model=self.api_model,
            temperature=0,
            messages=[
                *messages,
                {'role': 'user', 'content': description},
            ]
        )
        return resp.choices[0].message.content

    def generate_code(self, description) -> str:
        CD_message = self.generate_CD_message(description)
        print('CD_message', CD_message)
        CM_message = self.generate_CM_message(CD_message)
        print('CM_message', CM_message)
        messages = self.get_code_messages()
        import openai,os
        openai.api_key = self.api_key
        openai.api_base = self.api_base
        resp = openai.ChatCompletion.create(
            model=self.api_model,
            temperature=0,
            messages=[
                *messages,
                {'role': 'user', 'content': CM_message},
            ]
        )
        return resp.choices[0].message.content

In[ ]:

send_email = AutoGmailApiFunctionAdvance('send_email', '请帮我编写一个发送邮件的函数。函数名为 send_email', use_cache=True)

Out[ ]:

CD_message 请帮我编写一个发送邮件的函数。函数名为 send_email 当前需求中可以作为函数参数的是：1.收件人的邮箱地址 2.邮件的主题 3.邮件的内容 4.发件人的邮箱地址
CM_message 请帮我编写一个名为send_email的python函数，函数要求如下:1.函数的功能为使用我的Gmail账户发送邮件；2.函数参数包括to, subject, body, from，其中to是收件人的邮箱地址，subject是邮件的主题，body是邮件的内容，from是发件人的邮箱地址，默认情况下取值为"me"，表示使用我的Gmail账户发送邮件；3.函数返回结果是一个包含发送邮件的状态信息的json对象，返回结果本身必须是一个json格式对象；4.请将全部功能封装在一个函数内；5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: -Status: 发送状态；6.需要处理参数的特殊输入；

In[ ]:

def test_case1_get_current_time(get_current_time):
    """
    用例1：检查返回值是否为json
    """
    import json
    try:
        json.loads(get_current_time())
    except ValueError as e:
        return False
    return True

def test_case2_get_current_time(get_current_time):
    """
    用例2：检查返回时间是否为当前时间
    """
    import json, datetime
    result = get_current_time()
    data = json.loads(result)
    expected = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    for _,real in data.items():
        return abs((datetime.datetime.strptime(real, '%Y-%m-%d %H:%M:%S') - datetime.datetime.strptime(expected, '%Y-%m-%d %H:%M:%S')).seconds) <= 1

In[ ]:

get_current_time = AutoGmailApiFunctionAdvance('get_current_time', '请帮我编写一个查看当前时间的函数。函数名为 get_current_time')\
    .test(test_case1_get_current_time)\
    .test(test_case2_get_current_time)\
    .testEnd()

Out[ ]:

CD_message 请帮我编写一个查看当前时间的函数。函数名为 get_current_time 当前需求中可以作为函数参数的是：无参数
CM_message 请帮我编写一个名为get_current_time的python函数，函数要求如下:1.函数的功能为查看当前时间；2.函数的参数为空；3.函数返回结果是一个包含当前时间信息的json对象,时间表示为%Y-%m-%d %H:%M:%S；4.请将全部功能封装在一个函数内，只需要一个函数定义而不需要任何调用代码；5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: - CurrentTime: 当前时间；6.需要处理参数的特殊输入；

    用例1：检查返回值是否为json
     通过

    用例2：检查返回时间是否为当前时间
     通过
AutoGmailApiFunctionAdvance(get_current_time) {

instance:
<function get_current_time__func__ at 0x0000024228106D40>

description:

    函数名：get_current_time
    函数功能：获取当前时间
    参数：无
    返回结果：一个包含当前时间信息的json对象，时间表示为%Y-%m-%d %H:%M:%S
    - CurrentTime: 当前时间

path:
./tested/get_current_time_module.py

code:
import datetime
import json

def get_current_time():
    """
    函数名：get_current_time
    函数功能：获取当前时间
    参数：无
    返回结果：一个包含当前时间信息的json对象，时间表示为%Y-%m-%d %H:%M:%S
    - CurrentTime: 当前时间
    """
    current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    return json.dumps({"CurrentTime": current_time})
} 已测试

In[ ]:

get_email_count_by = AutoGmailApiFunctionAdvance('get_email_count_by', '请帮我编一个统计某人在最近n周内给我发过多少封邮件的函数,参数为1.统计谁的邮件 2.最近的n周 3.邮件发送者。函数名为 get_email_count_by')

Out[ ]:

CD_message 请帮我编一个统计某人在最近n周内给我发过多少封邮件的函数,参数为1.统计谁的邮件 2.最近的n周 3.邮件发送者。函数名为 get_email_count_by 当前需求中可以作为函数参数的是：1.统计谁的邮件 2.最近的n周 3.邮件发送者。
CM_message 请帮我编写一个名为get_email_count_by的python函数，函数要求如下:1.函数的功能为统计某人在最近n周内给我发过多少封邮件；2.函数参数包括userId，表示统计谁的邮件，是字符串参数，默认情况下取值为"me"；weeks，表示最近的n周，是整数参数；sender，表示邮件发送者，是字符串参数；3.函数返回结果是一个包含邮件数的json对象；4.请将全部功能封装在一个函数内；5.请在函数编写过程中，在函数内部加入中文编写的详细的函数说明文档，用于说明函数功能、函数参数情况以及函数返回结果等信息，文档注释中需要包括函数名、函数的作用、各个参数及其作用的描述，包括是否为必填参数、返回值键名及介绍 如: - Count: 邮件数；6.需要处理参数的特殊输入；

In[ ]:

functions = list(map(lambda x: x.get_properties(),[send_email, get_current_time, get_email_count_by]))

In[ ]:

get_email_count_by

Out[ ]:

AutoGmailApiFunctionAdvance(get_email_count_by) {

instance:
<function get_email_count_by__func__ at 0x000002424F8DB5B0>

description:

    函数功能：统计某人在最近n周内给我发过多少封邮件
    参数：
    userId：字符串参数，默认情况下取值为'me'，表示统计谁的邮件，非必填
    weeks：整数参数，表示最近的n周，非必填
    sender：字符串参数，表示邮件发送者，非必填
    返回结果：一个包含邮件数的json对象
    - Count: 邮件数

path:
././untest/get_email_count_by_module.py

code:
import dateutil.parser as parser
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
import json
import datetime

SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']

def get_email_count_by(userId='me', weeks=1, sender=None):
    """
    函数功能：统计某人在最近n周内给我发过多少封邮件
    参数：
    userId：字符串参数，默认情况下取值为'me'，表示统计谁的邮件，非必填
    weeks：整数参数，表示最近的n周，非必填
    sender：字符串参数，表示邮件发送者，非必填
    返回结果：一个包含邮件数的json对象
    - Count: 邮件数
    """
    if weeks < 0:
        return json.dumps({"Error": "Weeks must be a positive integer"})
    if not sender:
        return json.dumps({"Error": "Sender must be specified"})

    creds = Credentials.from_authorized_user_file('token.json', SCOPES)
    # Call the Gmail API
    service = build('gmail', 'v1', credentials=creds)
    results = service.users().messages().list(userId=userId).execute()
    messages = results.get('messages', [])

    count = 0
    for message in messages:
        msg = service.users().messages().get(userId=userId, id=message['id']).execute()
        email_data = msg['payload']['headers']
        for values in email_data:
            name = values['name']
            if name == 'From':
                From = values['value']
            if name == 'Date':
                Date = values['value']
        if From == sender:
            date_time_obj = datetime.datetime.strptime(Date, '%a, %d %b %Y %H:%M:%S %z')
            if (datetime.datetime.now(date_time_obj.tzinfo) - date_time_obj).days <= weeks * 7:
                count += 1
    return json.dumps({"Count": count})
}

In[ ]:

def chat_with_func_call(question, functions = None):
    """
    通过ChatGPT与用户进行对话
    Args:
        question(string): 用户输入的问题
        functions(List[Callbale]): 可供gpt调用的函数
    Returns:
        string: ChatGPT的回答
    """
    import json,openai
    openai.api_key = os.environ['OPENAI_API_KEY']
    openai.api_base = os.environ['OPENAI_API_BASE']
    messages = []
    messages.append({'role': 'user', 'content': question})
    while True:
        print('history',messages)
        resp = openai.ChatCompletion.create(
            model=os.environ['OPENAI_API_MODEL'],
            messages=messages,
            functions=functions,
            function_call='auto',
            temperature=0
        )
        if resp.choices[0].finish_reason == 'function_call':
            print('function_calling',resp.choices[0].message)
            arguments = json.loads(resp.choices[0].message.function_call.arguments)
            answer = globals()[resp.choices[0].message.function_call.name](**arguments)
            messages.append({'role': 'function', 'name': resp.choices[0].message.function_call.name, 'content': answer})
        else:
            return resp.choices[0].message.content

In[ ]:

chat_with_func_call('帮我查一下这两周小周给我发过多少封邮件，如果小于10封就通知他明天下午两点来找我开会，小周的邮箱是1297416662@qq.com',functions)

Out[ ]:

history [{'role': 'user', 'content': '帮我查一下这两周小周给我发过多少封邮件，如果小于10封就通知他明天下午两点来找我开会，小周的邮箱是1297416662@qq.com，我的邮箱是me'}]
function_calling {
  "role": "assistant",
  "content": null,
  "function_call": {
    "name": "get_email_count_by",
    "arguments": "{\n  \"userId\": \"me\",\n  \"weeks\": 2,\n  \"sender\": \"1297416662@qq.com\"\n}"
  }
}
history [{'role': 'user', 'content': '帮我查一下这两周小周给我发过多少封邮件，如果小于10封就通知他明天下午两点来找我开会，小周的邮箱是1297416662@qq.com，我的邮箱是me'}, {'role': 'function', 'name': 'get_email_count_by', 'content': '{"Count": 0}'}]
function_calling {
  "role": "assistant",
  "content": null,
  "function_call": {
    "name": "send_email",
    "arguments": "{\n  \"to\": \"1297416662@qq.com\",\n  \"subject\": \"Meeting Invitation\",\n  \"body\": \"Hi, please come to my office for a meeting at 2pm tomorrow.\",\n  \"from_email\": \"me\"\n}"
  }
}
history [{'role': 'user', 'content': '帮我查一下这两周小周给我发过多少封邮件，如果小于10封就通知他明天下午两点来找我开会，小周的邮箱是1297416662@qq.com，我的邮箱是me'}, {'role': 'function', 'name': 'get_email_count_by', 'content': '{"Count": 0}'}, {'role': 'function', 'name': 'send_email', 'content': '{"Status": "Email sent"}'}]
'已经帮你查到，这两周小周给你发的邮件数量为0封。我已经发送了一封邮件通知他明天下午两点来找你开会。'

ps: 最后一步回答时候，gpt4调用get_email_count_by一直搞错发件人和收件人woc！，差点心态崩了，感觉这东西离到靠谱的程度还是挺远的。