mirror of
https://github.com/FoundationAgents/MetaGPT.git
synced 2026-04-28 18:36:22 +02:00
stellaHSR
GitHub
commit
9e0d0205e2
56 changed files with 422 additions and 288 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -163,3 +163,4 @@ workspace/*
|
|||
*.mmd
|
||||
tmp
|
||||
output.wav
|
||||
metagpt/roles/idea_agent.py
|
||||
|
|
|
|||
0
Message
Normal file
0
Message
Normal file
0
None
Normal file
0
None
Normal file
0
int
Normal file
0
int
Normal file
|
|
@ -65,3 +65,4 @@ class Action(ABC):
|
|||
async def run(self, *args, **kwargs):
|
||||
"""Run action"""
|
||||
raise NotImplementedError("The run method should be implemented in a subclass.")
|
||||
|
||||
|
|
@ -40,3 +40,4 @@ class ActionOutput:
|
|||
new_class.__validator_check_name = classmethod(check_name)
|
||||
new_class.__root_validator_check_missing_fields = classmethod(check_missing_fields)
|
||||
return new_class
|
||||
|
||||
|
|
@ -28,10 +28,10 @@ Focus only on the names of shared dependencies, do not add any other explanation
|
|||
class AnalyzeDepLibs(Action):
|
||||
def __init__(self, name, context=None, llm=None):
|
||||
super().__init__(name, context, llm)
|
||||
self.desc = "根据上下文，分析程序运行依赖库"
|
||||
self.desc = "Analyze the runtime dependencies of the program based on the context"
|
||||
|
||||
async def run(self, requirement, filepaths_string):
|
||||
# prompt = f"以下是产品需求文档(PRD):\n\n{prd}\n\n{PROMPT}"
|
||||
# prompt = f"Below is the product requirement document (PRD):\n\n{prd}\n\n{PROMPT}"
|
||||
prompt = PROMPT.format(prompt=requirement, filepaths_string=filepaths_string)
|
||||
design_filenames = await self._aask(prompt)
|
||||
return design_filenames
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ class AzureTTS(Action):
|
|||
super().__init__(name, context, llm)
|
||||
self.config = Config()
|
||||
|
||||
# 参数参考：https://learn.microsoft.com/zh-cn/azure/cognitive-services/speech-service/language-support?tabs=tts#voice-styles-and-roles
|
||||
# Parameters reference: https://learn.microsoft.com/zh-cn/azure/cognitive-services/speech-service/language-support?tabs=tts#voice-styles-and-roles
|
||||
def synthesize_speech(self, lang, voice, role, text, output_file):
|
||||
subscription_key = self.config.get('AZURE_TTS_SUBSCRIPTION_KEY')
|
||||
region = self.config.get('AZURE_TTS_REGION')
|
||||
|
|
@ -49,5 +49,5 @@ if __name__ == "__main__":
|
|||
"zh-CN",
|
||||
"zh-CN-YunxiNeural",
|
||||
"Boy",
|
||||
"你好，我是卡卡",
|
||||
"Hello, I am Kaka",
|
||||
"output.wav")
|
||||
|
|
|
|||
|
|
@ -100,7 +100,7 @@ class WriteDesign(Action):
|
|||
try:
|
||||
shutil.rmtree(workspace)
|
||||
except FileNotFoundError:
|
||||
pass # 文件夹不存在，但我们不在意
|
||||
pass # Folder does not exist, but we don't care
|
||||
workspace.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def _save_prd(self, docs_path, resources_path, prd):
|
||||
|
|
@ -141,3 +141,4 @@ class WriteDesign(Action):
|
|||
system_design = await self._aask_v1(prompt, "system_design", OUTPUT_MAPPING)
|
||||
self._save(context, system_design)
|
||||
return system_design
|
||||
|
||||
|
|
@ -19,3 +19,4 @@ class DesignReview(Action):
|
|||
|
||||
api_review = await self._aask(prompt)
|
||||
return api_review
|
||||
|
||||
|
|
@ -26,3 +26,4 @@ class DesignFilenames(Action):
|
|||
logger.debug(prompt)
|
||||
logger.debug(design_filenames)
|
||||
return design_filenames
|
||||
|
||||
|
|
@ -126,3 +126,4 @@ class AssignTasks(Action):
|
|||
async def run(self, *args, **kwargs):
|
||||
# Here you should implement the actual action
|
||||
pass
|
||||
|
||||
|
|
@ -140,3 +140,4 @@ class SearchAndSummarize(Action):
|
|||
logger.debug(prompt)
|
||||
logger.debug(result)
|
||||
return result
|
||||
|
||||
|
|
@ -79,3 +79,4 @@ class WriteCode(Action):
|
|||
# code_rsp = await self._aask_v1(prompt, "code_rsp", OUTPUT_MAPPING)
|
||||
# self._save(context, filename, code)
|
||||
return code
|
||||
|
||||
|
|
@ -79,3 +79,4 @@ class WriteCodeReview(Action):
|
|||
# code_rsp = await self._aask_v1(prompt, "code_rsp", OUTPUT_MAPPING)
|
||||
# self._save(context, filename, code)
|
||||
return code
|
||||
|
||||
|
|
@ -144,3 +144,4 @@ class WritePRD(Action):
|
|||
logger.debug(prompt)
|
||||
prd = await self._aask_v1(prompt, "prd", OUTPUT_MAPPING)
|
||||
return prd
|
||||
|
||||
|
|
@ -25,3 +25,4 @@ class WritePRDReview(Action):
|
|||
prompt = self.prd_review_prompt_template.format(prd=self.prd)
|
||||
review = await self._aask(prompt)
|
||||
return review
|
||||
|
||||
|
|
@ -1,9 +1,9 @@
|
|||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
@Time : 2023/5/11 17:45
|
||||
@Time : 2023/5/11 22:12
|
||||
@Author : alexanderwu
|
||||
@File : write_test.py
|
||||
@File : environment.py
|
||||
"""
|
||||
from metagpt.actions.action import Action
|
||||
from metagpt.utils.common import CodeParser
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
提供配置，单例
|
||||
Provide configuration, singleton
|
||||
"""
|
||||
import os
|
||||
|
||||
|
|
@ -28,7 +28,7 @@ class NotConfiguredException(Exception):
|
|||
|
||||
class Config(metaclass=Singleton):
|
||||
"""
|
||||
常规使用方法：
|
||||
Regular usage method:
|
||||
config = Config("config.yaml")
|
||||
secret_key = config.get_key("MY_SECRET_KEY")
|
||||
print("Secret key:", secret_key)
|
||||
|
|
@ -84,14 +84,14 @@ class Config(metaclass=Singleton):
|
|||
self.model_for_researcher_report = self._get("MODEL_FOR_RESEARCHER_REPORT")
|
||||
|
||||
def _init_with_config_files_and_env(self, configs: dict, yaml_file):
|
||||
"""从config/key.yaml / config/config.yaml / env三处按优先级递减加载"""
|
||||
"""Load from config/key.yaml, config/config.yaml, and env in decreasing order of priority"""
|
||||
configs.update(os.environ)
|
||||
|
||||
for _yaml_file in [yaml_file, self.key_yaml_file]:
|
||||
if not _yaml_file.exists():
|
||||
continue
|
||||
|
||||
# 加载本地 YAML 文件
|
||||
# Load local YAML file
|
||||
with open(_yaml_file, "r", encoding="utf-8") as file:
|
||||
yaml_data = yaml.safe_load(file)
|
||||
if not yaml_data:
|
||||
|
|
@ -103,11 +103,11 @@ class Config(metaclass=Singleton):
|
|||
return self._configs.get(*args, **kwargs)
|
||||
|
||||
def get(self, key, *args, **kwargs):
|
||||
"""从config/key.yaml / config/config.yaml / env三处找值，找不到报错"""
|
||||
"""Search for a value in config/key.yaml, config/config.yaml, and env; raise an error if not found"""
|
||||
value = self._get(key, *args, **kwargs)
|
||||
if value is None:
|
||||
raise ValueError(f"Key '{key}' not found in environment variables or in the YAML file")
|
||||
return value
|
||||
|
||||
|
||||
CONFIG = Config()
|
||||
CONFIG = Config()
|
||||
|
|
@ -9,7 +9,7 @@ from pathlib import Path
|
|||
|
||||
|
||||
def get_project_root():
|
||||
"""逐级向上寻找项目根目录"""
|
||||
"""Search upwards to find the project root directory."""
|
||||
current_path = Path.cwd()
|
||||
while True:
|
||||
if (current_path / '.git').exists() or \
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ from metagpt.config import Config
|
|||
|
||||
|
||||
class BaseStore(ABC):
|
||||
"""FIXME: consider add_index, set_index and think 颗粒度"""
|
||||
"""FIXME: consider add_index, set_index and think about granularity."""
|
||||
|
||||
@abstractmethod
|
||||
def search(self, *args, **kwargs):
|
||||
|
|
@ -53,3 +53,4 @@ class LocalStore(BaseStore, ABC):
|
|||
@abstractmethod
|
||||
def _write(self, docs, metadatas):
|
||||
raise NotImplementedError
|
||||
|
||||
|
|
@ -9,7 +9,7 @@ import chromadb
|
|||
|
||||
|
||||
class ChromaStore:
|
||||
"""如果从BaseStore继承，或者引入metagpt的其他模块，就会Python异常，很奇怪"""
|
||||
"""If inherited from BaseStore, or importing other modules from metagpt, a Python exception occurs, which is strange."""
|
||||
def __init__(self, name):
|
||||
client = chromadb.Client()
|
||||
collection = client.create_collection(name)
|
||||
|
|
@ -27,7 +27,7 @@ class ChromaStore:
|
|||
return results
|
||||
|
||||
def persist(self):
|
||||
"""chroma建议使用server模式，不本地persist"""
|
||||
"""Chroma recommends using server mode and not persisting locally."""
|
||||
raise NotImplementedError
|
||||
|
||||
def write(self, documents, metadatas, ids):
|
||||
|
|
|
|||
|
|
@ -79,3 +79,4 @@ class Document:
|
|||
return self._get_docs_and_metadatas_by_langchain()
|
||||
else:
|
||||
raise NotImplementedError
|
||||
|
||||
|
|
@ -59,7 +59,7 @@ class FaissStore(LocalStore):
|
|||
return str(sep.join([f"{x.page_content}" for x in rsp]))
|
||||
|
||||
def write(self):
|
||||
"""根据用户给定的Document（JSON / XLSX等）文件，进行index与库的初始化"""
|
||||
"""Initialize the index and library based on the Document (JSON / XLSX, etc.) file provided by the user."""
|
||||
if not self.raw_data.exists():
|
||||
raise FileNotFoundError
|
||||
doc = Document(self.raw_data, self.content_col, self.meta_col)
|
||||
|
|
@ -70,16 +70,16 @@ class FaissStore(LocalStore):
|
|||
return self.store
|
||||
|
||||
def add(self, texts: list[str], *args, **kwargs) -> list[str]:
|
||||
"""FIXME: 目前add之后没有更新store"""
|
||||
"""FIXME: Currently, the store is not updated after adding."""
|
||||
return self.store.add_texts(texts)
|
||||
|
||||
def delete(self, *args, **kwargs):
|
||||
"""目前langchain没有提供del接口"""
|
||||
"""Currently, langchain does not provide a delete interface."""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
faiss_store = FaissStore(DATA_PATH / 'qcs/qcs_4w.json')
|
||||
logger.info(faiss_store.search('油皮洗面奶'))
|
||||
faiss_store.add([f'油皮洗面奶-{i}' for i in range(3)])
|
||||
logger.info(faiss_store.search('油皮洗面奶'))
|
||||
logger.info(faiss_store.search('Oily Skin Facial Cleanser'))
|
||||
faiss_store.add([f'Oily Skin Facial Cleanser-{i}' for i in range(3)])
|
||||
logger.info(faiss_store.search('Oily Skin Facial Cleanser'))
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ type_mapping = {
|
|||
|
||||
|
||||
def columns_to_milvus_schema(columns: dict, primary_col_name: str = "", desc: str = ""):
|
||||
"""这里假设columns结构是str: 常规类型"""
|
||||
"""Assume the structure of columns is str: regular type"""
|
||||
fields = []
|
||||
for col, ctype in columns.items():
|
||||
if ctype == str:
|
||||
|
|
@ -80,7 +80,7 @@ class MilvusStore(BaseStore):
|
|||
FIXME: ADD TESTS
|
||||
https://milvus.io/docs/v2.0.x/search.md
|
||||
All search and query operations within Milvus are executed in memory. Load the collection to memory before conducting a vector similarity search.
|
||||
注意到上述描述，这个逻辑是认真的吗？这个耗时应该很长？
|
||||
Note the above description, is this logic serious? This should take a long time, right?
|
||||
"""
|
||||
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}
|
||||
results = self.collection.search(
|
||||
|
|
@ -91,7 +91,7 @@ class MilvusStore(BaseStore):
|
|||
expr=None,
|
||||
consistency_level="Strong"
|
||||
)
|
||||
# FIXME: results里有id，但是id到实际值还得调用query接口来获取
|
||||
# FIXME: results contain id, but to get the actual value from the id, we still need to call the query interface
|
||||
return results
|
||||
|
||||
def write(self, name, schema, *args, **kwargs):
|
||||
|
|
|
|||
|
|
@ -25,4 +25,4 @@ def print_classes_and_functions(module):
|
|||
|
||||
|
||||
if __name__ == '__main__':
|
||||
print_classes_and_functions(metagpt)
|
||||
print_classes_and_functions(metagpt)
|
||||
|
|
@ -12,7 +12,6 @@ from metagpt.provider.openai_api import OpenAIGPTAPI as LLM
|
|||
DEFAULT_LLM = LLM()
|
||||
CLAUDE_LLM = Claude()
|
||||
|
||||
|
||||
async def ai_func(prompt):
|
||||
"""使用LLM进行QA
|
||||
QA with LLMs
|
||||
|
|
|
|||
|
|
@ -12,7 +12,6 @@ from loguru import logger as _logger
|
|||
|
||||
from metagpt.const import PROJECT_ROOT
|
||||
|
||||
|
||||
def define_log_level(print_level="INFO", logfile_level="DEBUG"):
|
||||
"""调整日志级别到level之上
|
||||
Adjust the log level to above level
|
||||
|
|
@ -22,5 +21,4 @@ def define_log_level(print_level="INFO", logfile_level="DEBUG"):
|
|||
_logger.add(PROJECT_ROOT / 'logs/log.txt', level=logfile_level)
|
||||
return _logger
|
||||
|
||||
|
||||
logger = define_log_level()
|
||||
|
|
|
|||
|
|
@ -68,3 +68,4 @@ class LongTermMemory(Memory):
|
|||
def clear(self):
|
||||
super(LongTermMemory, self).clear()
|
||||
self.memory_storage.clean()
|
||||
|
||||
|
|
@ -85,3 +85,4 @@ class Memory:
|
|||
continue
|
||||
rsp += self.index[action]
|
||||
return rsp
|
||||
|
||||
|
|
@ -104,3 +104,4 @@ class MemoryStorage(FaissStore):
|
|||
|
||||
self.store = None
|
||||
self._initialized = False
|
||||
|
||||
|
|
@ -1,17 +1,16 @@
|
|||
你是一个富有帮助的助理，可以帮助撰写、抽象、注释、摘要Python代码
|
||||
You are a helpful assistant that can assist in writing, abstracting, annotating, and summarizing Python code.
|
||||
|
||||
1. 不要提到类/函数名
|
||||
2. 不要提到除了系统库与公共库以外的类/函数
|
||||
3. 试着将类/函数总结为不超过6句话
|
||||
4. 你的回答应该是一行文本
|
||||
|
||||
举例，如果上下文是：
|
||||
Do not mention class/function names.
|
||||
Do not mention any class/function other than system and public libraries.
|
||||
Try to summarize the class/function in no more than 6 sentences.
|
||||
Your answer should be in one line of text.
|
||||
For instance, if the context is:
|
||||
|
||||
```python
|
||||
from typing import Optional
|
||||
from abc import ABC
|
||||
from metagpt.llm import LLM # 大语言模型，类似GPT
|
||||
|
||||
from metagpt.llm import LLM # Large language model, similar to GPT
|
||||
n
|
||||
class Action(ABC):
|
||||
def __init__(self, name='', context=None, llm: LLM = LLM()):
|
||||
self.name = name
|
||||
|
|
@ -21,38 +20,38 @@
|
|||
self.desc = ""
|
||||
|
||||
def set_prefix(self, prefix):
|
||||
"""设置前缀以供后续使用"""
|
||||
"""Set prefix for subsequent use"""
|
||||
self.prefix = prefix
|
||||
|
||||
async def _aask(self, prompt: str, system_msgs: Optional[list[str]] = None):
|
||||
"""加上默认的prefix来使用prompt"""
|
||||
"""Use prompt with the default prefix"""
|
||||
if not system_msgs:
|
||||
system_msgs = []
|
||||
system_msgs.append(self.prefix)
|
||||
return await self.llm.aask(prompt, system_msgs)
|
||||
|
||||
async def run(self, *args, **kwargs):
|
||||
"""运行动作"""
|
||||
"""Execute action"""
|
||||
raise NotImplementedError("The run method should be implemented in a subclass.")
|
||||
|
||||
PROMPT_TEMPLATE = """
|
||||
# 需求
|
||||
# Requirements
|
||||
{requirements}
|
||||
|
||||
# PRD
|
||||
根据需求创建一个产品需求文档（PRD），填补以下空缺
|
||||
Create a product requirement document (PRD) based on the requirements and fill in the blanks below:
|
||||
|
||||
产品/功能介绍：
|
||||
Product/Function Introduction:
|
||||
|
||||
目标：
|
||||
Goals:
|
||||
|
||||
用户和使用场景：
|
||||
Users and Usage Scenarios:
|
||||
|
||||
需求：
|
||||
Requirements:
|
||||
|
||||
约束与限制：
|
||||
Constraints and Limitations:
|
||||
|
||||
性能指标：
|
||||
Performance Metrics:
|
||||
|
||||
"""
|
||||
|
||||
|
|
@ -68,9 +67,8 @@ # PRD
|
|||
```
|
||||
|
||||
|
||||
主类/函数是 `WritePRD`。
|
||||
The main class/function is WritePRD.
|
||||
|
||||
那么你应该写：
|
||||
|
||||
这个类用来根据输入需求生成PRD。首先注意到有一个提示词模板，其中有产品、功能、目标、用户和使用场景、需求、约束与限制、性能指标，这个模板会以输入需求填充，然后调用接口询问大语言模型，让大语言模型返回具体的PRD。
|
||||
Then you should write:
|
||||
|
||||
This class is designed to generate a PRD based on input requirements. Notably, there's a template prompt with sections for product, function, goals, user scenarios, requirements, constraints, performance metrics. This template gets filled with input requirements and then queries a big language model to produce the detailed PRD.
|
||||
|
|
@ -7,34 +7,34 @@
|
|||
"""
|
||||
|
||||
METAGPT_SAMPLE = """
|
||||
### 设定
|
||||
### Settings
|
||||
|
||||
你是一个用户的编程助手，可以使用公共库与python系统库进行编程，你的回复应该有且只有一个函数。
|
||||
1. 函数本身应尽可能完整，不应缺失需求细节
|
||||
2. 你可能需要写一些提示词，用来让LLM（你自己）理解带有上下文的搜索请求
|
||||
3. 面对复杂的、难以用简单函数解决的逻辑，尽量交给llm解决
|
||||
You are a programming assistant for a user, capable of coding using public libraries and Python system libraries. Your response should have only one function.
|
||||
1. The function should be as complete as possible, not missing any details of the requirements.
|
||||
2. You might need to write some prompt words to let LLM (yourself) understand context-bearing search requests.
|
||||
3. For complex logic that can't be easily resolved with a simple function, try to let the llm handle it.
|
||||
|
||||
### 公共库
|
||||
### Public Libraries
|
||||
|
||||
你可以使用公共库metagpt提供的函数，不能使用其他第三方库的函数。公共库默认已经被import为x变量
|
||||
You can use the functions provided by the public library metagpt, but can't use functions from other third-party libraries. The public library is imported as variable x by default.
|
||||
- `import metagpt as x`
|
||||
- 你可以使用 `x.func(paras)` 方式来对公共库进行调用。
|
||||
- You can call the public library using the `x.func(paras)` format.
|
||||
|
||||
公共库中已有函数如下
|
||||
- def llm(question: str) -> str # 输入问题，基于大模型进行回答
|
||||
- def intent_detection(query: str) -> str # 输入query，分析意图，返回公共库函数名
|
||||
- def add_doc(doc_path: str) -> None # 输入文件路径或者文件夹路径，加入知识库
|
||||
- def search(query: str) -> list[str] # 输入query返回向量知识库搜索的多个结果
|
||||
- def google(query: str) -> list[str] # 使用google查询公网结果
|
||||
- def math(query: str) -> str # 输入query公式，返回对公式执行的结果
|
||||
- def tts(text: str, wav_path: str) # 输入text文本与对应想要输出音频的路径，将文本转为音频文件
|
||||
Functions already available in the public library are:
|
||||
- def llm(question: str) -> str # Input a question and get an answer based on the large model.
|
||||
- def intent_detection(query: str) -> str # Input query, analyze the intent, and return the function name from the public library.
|
||||
- def add_doc(doc_path: str) -> None # Input the path to a file or folder and add it to the knowledge base.
|
||||
- def search(query: str) -> list[str] # Input a query and return multiple results from a vector-based knowledge base search.
|
||||
- def google(query: str) -> list[str] # Use Google to search for public results.
|
||||
- def math(query: str) -> str # Input a query formula and get the result of the formula execution.
|
||||
- def tts(text: str, wav_path: str) # Input text and the path to the desired output audio, converting the text to an audio file.
|
||||
|
||||
### 用户需求
|
||||
### User Requirements
|
||||
|
||||
我有一个个人知识库文件，我希望基于它来实现一个带有搜索功能的个人助手，需求细则如下
|
||||
1. 个人助手会思考是否需要使用个人知识库搜索，如果没有必要，就不使用它
|
||||
2. 个人助手会判断用户意图，在不同意图下使用恰当的函数解决问题
|
||||
3. 用语音回答
|
||||
I have a personal knowledge base file. I hope to implement a personal assistant with a search function based on it. The detailed requirements are as follows:
|
||||
1. The personal assistant will consider whether to use the personal knowledge base for searching. If it's unnecessary, it won't use it.
|
||||
2. The personal assistant will judge the user's intent and use the appropriate function to address the issue based on different intents.
|
||||
3. Answer in voice.
|
||||
|
||||
"""
|
||||
# - def summarize(doc: str) -> str # 输入doc返回摘要
|
||||
# - def summarize(doc: str) -> str # Input doc and return a summary.
|
||||
|
|
|
|||
|
|
@ -6,9 +6,8 @@
|
|||
@File : summarize.py
|
||||
"""
|
||||
|
||||
|
||||
# 出自插件：ChatGPT - 网站和 YouTube 视频摘要
|
||||
# https://chrome.google.com/webstore/detail/chatgpt-%C2%BB-summarize-every/cbgecfllfhmmnknmamkejadjmnmpfjmp?hl=zh-CN&utm_source=chrome-ntp-launcher
|
||||
# From the plugin: ChatGPT - Website and YouTube Video Summaries
|
||||
# https://chrome.google.com/webstore/detail/chatgpt-%C2%BB-summarize-every/cbgecfllfhmmnknmamkejadjmnmpfjmp?hl=en&utm_source=chrome-ntp-launcher
|
||||
SUMMARIZE_PROMPT = """
|
||||
Your output should use the following template:
|
||||
### Summary
|
||||
|
|
@ -22,9 +21,9 @@ a YouTube video, use the following text: {{CONTENT}}.
|
|||
"""
|
||||
|
||||
|
||||
# GCP-VertexAI-文本摘要（SUMMARIZE_PROMPT_2-5都是）
|
||||
# GCP-VertexAI-Text Summarization (SUMMARIZE_PROMPT_2-5 are from this source)
|
||||
# https://github.com/GoogleCloudPlatform/generative-ai/blob/main/language/examples/prompt-design/text_summarization.ipynb
|
||||
# 长文档需要map-reduce过程，见下面这个notebook
|
||||
# Long documents require a map-reduce process, see the following notebook
|
||||
# https://github.com/GoogleCloudPlatform/generative-ai/blob/main/language/examples/document-summarization/summarization_large_documents.ipynb
|
||||
SUMMARIZE_PROMPT_2 = """
|
||||
Provide a very short summary, no more than three sentences, for the following article:
|
||||
|
|
|
|||
|
|
@ -32,3 +32,4 @@ class Claude2:
|
|||
max_tokens_to_sample=1000,
|
||||
)
|
||||
return res.completion
|
||||
|
||||
|
|
@ -25,3 +25,4 @@ class BaseChatbot(ABC):
|
|||
@abstractmethod
|
||||
def ask_code(self, msgs: list) -> str:
|
||||
"""Ask GPT multiple questions and get a piece of code"""
|
||||
|
||||
|
|
@ -115,3 +115,4 @@ class BaseGPTAPI(BaseChatbot):
|
|||
def messages_to_dict(self, messages):
|
||||
"""objects to [{"role": "user", "content": msg}] etc."""
|
||||
return [i.to_dict() for i in messages]
|
||||
|
||||
|
|
@ -48,14 +48,12 @@ class RateLimiter:
|
|||
|
||||
self.last_call_time = time.time()
|
||||
|
||||
|
||||
class Costs(NamedTuple):
|
||||
total_prompt_tokens: int
|
||||
total_completion_tokens: int
|
||||
total_cost: float
|
||||
total_budget: float
|
||||
|
||||
|
||||
class CostManager(metaclass=Singleton):
|
||||
"""计算使用接口的开销"""
|
||||
|
||||
|
|
@ -102,18 +100,26 @@ class CostManager(metaclass=Singleton):
|
|||
"""
|
||||
return self.total_completion_tokens
|
||||
|
||||
def get_total_cost(self):
|
||||
"""
|
||||
Get the total cost of API calls.
|
||||
def get_total_cost(self):
|
||||
"""
|
||||
Get the total cost of API calls.
|
||||
|
||||
Returns:
|
||||
float: The total cost of API calls.
|
||||
"""
|
||||
return self.total_cost
|
||||
Returns:
|
||||
float: The total cost of API calls.
|
||||
"""
|
||||
return self.total_cost
|
||||
|
||||
def get_costs(self) -> Costs:
|
||||
"""获得所有开销"""
|
||||
return Costs(self.total_prompt_tokens, self.total_completion_tokens, self.total_cost, self.total_budget)
|
||||
def get_costs(self) -> Costs:
|
||||
"""Get all costs"""
|
||||
return Costs(self.total_prompt_tokens, self.total_completion_tokens, self.total_cost, self.total_budget)
|
||||
|
||||
def log_and_reraise(retry_state):
|
||||
logger.error(f"Retry attempts exhausted. Last exception: {retry_state.outcome.exception()}")
|
||||
logger.warning("""
|
||||
Recommend going to https://deepwisdom.feishu.cn/wiki/MsGnwQBjiif9c3koSJNcYaoSnu4#part-XdatdVlhEojeAfxaaEZcMV3ZniQ
|
||||
See FAQ 5.8
|
||||
""")
|
||||
raise retry_state.outcome.exception()
|
||||
|
||||
|
||||
def log_and_reraise(retry_state):
|
||||
|
|
@ -238,7 +244,7 @@ class OpenAIGPTAPI(BaseGPTAPI, RateLimiter):
|
|||
return usage
|
||||
|
||||
async def acompletion_batch(self, batch: list[list[dict]]) -> list[dict]:
|
||||
"""返回完整JSON"""
|
||||
"""Return full JSON"""
|
||||
split_batches = self.split_batches(batch)
|
||||
all_results = []
|
||||
|
||||
|
|
@ -254,7 +260,7 @@ class OpenAIGPTAPI(BaseGPTAPI, RateLimiter):
|
|||
return all_results
|
||||
|
||||
async def acompletion_batch_text(self, batch: list[list[dict]]) -> list[str]:
|
||||
"""仅返回纯文本"""
|
||||
"""Only return plain text"""
|
||||
raw_results = await self.acompletion_batch(batch)
|
||||
results = []
|
||||
for idx, raw_result in enumerate(raw_results, start=1):
|
||||
|
|
|
|||
|
|
@ -11,9 +11,28 @@ from metagpt.roles import Role
|
|||
|
||||
|
||||
class Architect(Role):
|
||||
"""Architect: Listen to PRD, responsible for designing API, designing code files"""
|
||||
def __init__(self, name="Bob", profile="Architect", goal="Design a concise, usable, complete python system",
|
||||
constraints="Try to specify good open source tools as much as possible"):
|
||||
"""
|
||||
Represents an Architect role in a software development process.
|
||||
|
||||
Attributes:
|
||||
name (str): Name of the architect.
|
||||
profile (str): Role profile, default is 'Architect'.
|
||||
goal (str): Primary goal or responsibility of the architect.
|
||||
constraints (str): Constraints or guidelines for the architect.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
name: str = "Bob",
|
||||
profile: str = "Architect",
|
||||
goal: str = "Design a concise, usable, complete python system",
|
||||
constraints: str = "Try to specify good open source tools as much as possible") -> None:
|
||||
"""Initializes the Architect with given attributes."""
|
||||
super().__init__(name, profile, goal, constraints)
|
||||
|
||||
# Initialize actions specific to the Architect role
|
||||
self._init_actions([WriteDesign])
|
||||
|
||||
# Set events or actions the Architect should watch or be aware of
|
||||
self._watch({WritePRD})
|
||||
|
||||
|
||||
|
|
@ -32,3 +32,4 @@ class CustomerService(Sales):
|
|||
store=None
|
||||
):
|
||||
super().__init__(name, profile, desc=desc, store=store)
|
||||
|
||||
|
|
@ -47,9 +47,27 @@ async def gather_ordered_k(coros, k) -> list:
|
|||
|
||||
|
||||
class Engineer(Role):
|
||||
def __init__(self, name="Alex", profile="Engineer", goal="Write elegant, readable, extensible, efficient code",
|
||||
constraints="The code you write should conform to code standard like PEP8, be modular, easy to read and maintain",
|
||||
n_borg=1, use_code_review=False):
|
||||
"""
|
||||
Represents an Engineer role responsible for writing and possibly reviewing code.
|
||||
|
||||
Attributes:
|
||||
name (str): Name of the engineer.
|
||||
profile (str): Role profile, default is 'Engineer'.
|
||||
goal (str): Goal of the engineer.
|
||||
constraints (str): Constraints for the engineer.
|
||||
n_borg (int): Number of borgs.
|
||||
use_code_review (bool): Whether to use code review.
|
||||
todos (list): List of tasks.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
name: str = "Alex",
|
||||
profile: str = "Engineer",
|
||||
goal: str = "Write elegant, readable, extensible, efficient code",
|
||||
constraints: str = "The code should conform to standards like PEP8 and be modular and maintainable",
|
||||
n_borg: int = 1,
|
||||
use_code_review: bool = False) -> None:
|
||||
"""Initializes the Engineer role with given attributes."""
|
||||
super().__init__(name, profile, goal, constraints)
|
||||
self._init_actions([WriteCode])
|
||||
self.use_code_review = use_code_review
|
||||
|
|
@ -88,7 +106,7 @@ class Engineer(Role):
|
|||
try:
|
||||
shutil.rmtree(workspace)
|
||||
except FileNotFoundError:
|
||||
pass # 文件夹不存在，但我们不在意
|
||||
pass # The folder does not exist, but we don't care
|
||||
workspace.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def write_file(self, filename: str, code: str):
|
||||
|
|
@ -158,23 +176,23 @@ class Engineer(Role):
|
|||
code_msg_all = [] # gather all code info, will pass to qa_engineer for tests later
|
||||
for todo in self.todos:
|
||||
"""
|
||||
# 从历史信息中挑选必须的信息，以减少prompt长度（人工经验总结）
|
||||
1. Architect全部
|
||||
2. ProjectManager全部
|
||||
3. 是否需要其他代码（暂时需要）？
|
||||
TODO:目标是不需要。在任务拆分清楚后，根据设计思路，不需要其他代码也能够写清楚单个文件，如果不能则表示还需要在定义的更清晰，这个是代码能够写长的关键
|
||||
# Select essential information from the historical data to reduce the length of the prompt (summarized from human experience):
|
||||
1. All from Architect
|
||||
2. All from ProjectManager
|
||||
3. Do we need other codes (currently needed)?
|
||||
TODO: The goal is not to need it. After clear task decomposition, based on the design idea, you should be able to write a single file without needing other codes. If you can't, it means you need a clearer definition. This is the key to writing longer code.
|
||||
"""
|
||||
context = []
|
||||
msg = self._rc.memory.get_by_actions([WriteDesign, WriteTasks, WriteCode])
|
||||
for m in msg:
|
||||
context.append(m.content)
|
||||
context_str = "\n".join(context)
|
||||
# 编写code
|
||||
# Write code
|
||||
code = await WriteCode().run(
|
||||
context=context_str,
|
||||
filename=todo
|
||||
)
|
||||
# code review
|
||||
# Code review
|
||||
if self.use_code_review:
|
||||
try:
|
||||
rewrite_code = await WriteCodeReview().run(
|
||||
|
|
@ -203,6 +221,7 @@ class Engineer(Role):
|
|||
return msg
|
||||
|
||||
async def _act(self) -> Message:
|
||||
"""Determines the mode of action based on whether code review is used."""
|
||||
if self.use_code_review:
|
||||
return await self._act_sp_precision()
|
||||
return await self._act_sp()
|
||||
return await self._act_sp()
|
||||
|
|
@ -10,8 +10,30 @@ from metagpt.roles import Role
|
|||
|
||||
|
||||
class ProductManager(Role):
|
||||
def __init__(self, name="Alice", profile="Product Manager", goal="Efficiently create a successful product",
|
||||
constraints=""):
|
||||
"""
|
||||
Represents a Product Manager role responsible for product development and management.
|
||||
|
||||
Attributes:
|
||||
name (str): Name of the product manager.
|
||||
profile (str): Role profile, default is 'Product Manager'.
|
||||
goal (str): Goal of the product manager.
|
||||
constraints (str): Constraints or limitations for the product manager.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
name: str = "Alice",
|
||||
profile: str = "Product Manager",
|
||||
goal: str = "Efficiently create a successful product",
|
||||
constraints: str = "") -> None:
|
||||
"""
|
||||
Initializes the ProductManager role with given attributes.
|
||||
|
||||
Args:
|
||||
name (str): Name of the product manager.
|
||||
profile (str): Role profile.
|
||||
goal (str): Goal of the product manager.
|
||||
constraints (str): Constraints or limitations for the product manager.
|
||||
"""
|
||||
super().__init__(name, profile, goal, constraints)
|
||||
self._init_actions([WritePRD])
|
||||
self._watch([BossRequirement])
|
||||
self._watch([BossRequirement])
|
||||
|
|
@ -10,8 +10,30 @@ from metagpt.roles import Role
|
|||
|
||||
|
||||
class ProjectManager(Role):
|
||||
def __init__(self, name="Eve", profile="Project Manager",
|
||||
goal="Improve team efficiency and deliver with quality and quantity", constraints=""):
|
||||
"""
|
||||
Represents a Project Manager role responsible for overseeing project execution and team efficiency.
|
||||
|
||||
Attributes:
|
||||
name (str): Name of the project manager.
|
||||
profile (str): Role profile, default is 'Project Manager'.
|
||||
goal (str): Goal of the project manager.
|
||||
constraints (str): Constraints or limitations for the project manager.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
name: str = "Eve",
|
||||
profile: str = "Project Manager",
|
||||
goal: str = "Improve team efficiency and deliver with quality and quantity",
|
||||
constraints: str = "") -> None:
|
||||
"""
|
||||
Initializes the ProjectManager role with given attributes.
|
||||
|
||||
Args:
|
||||
name (str): Name of the project manager.
|
||||
profile (str): Role profile.
|
||||
goal (str): Goal of the project manager.
|
||||
constraints (str): Constraints or limitations for the project manager.
|
||||
"""
|
||||
super().__init__(name, profile, goal, constraints)
|
||||
self._init_actions([WriteTasks])
|
||||
self._watch([WriteDesign])
|
||||
self._watch([WriteDesign])
|
||||
|
|
@ -7,40 +7,39 @@
|
|||
"""
|
||||
from enum import Enum
|
||||
|
||||
PREFIX = """尽你所能回答以下问题。你可以使用以下工具："""
|
||||
FORMAT_INSTRUCTIONS = """请按照以下格式：
|
||||
PREFIX = """Answer the questions to the best of your ability. You can use the following tools:"""
|
||||
FORMAT_INSTRUCTIONS = """Please follow the format below:
|
||||
|
||||
问题：你需要回答的输入问题
|
||||
思考：你应该始终思考该怎么做
|
||||
行动：要采取的行动，应该是[{tool_names}]中的一个
|
||||
行动输入：行动的输入
|
||||
观察：行动的结果
|
||||
...（这个思考/行动/行动输入/观察可以重复N次）
|
||||
思考：我现在知道最终答案了
|
||||
最终答案：对原始输入问题的最终答案"""
|
||||
SUFFIX = """开始吧！
|
||||
|
||||
问题：{input}
|
||||
思考：{agent_scratchpad}"""
|
||||
Question: The input question you need to answer
|
||||
Thoughts: You should always think about how to do it
|
||||
Action: The action to be taken, should be one from [{tool_names}]
|
||||
Action Input: Input for the action
|
||||
Observation: Result of the action
|
||||
... (This Thoughts/Action/Action Input/Observation can be repeated N times)
|
||||
Thoughts: I now know the final answer
|
||||
Final Answer: The final answer to the original input question"""
|
||||
SUFFIX = """Let's begin!
|
||||
|
||||
Question: {input}
|
||||
Thoughts: {agent_scratchpad}"""
|
||||
|
||||
class PromptString(Enum):
|
||||
REFLECTION_QUESTIONS = "以下是一些陈述：\n{memory_descriptions}\n\n仅根据以上信息，我们可以回答关于陈述中主题的3个最显著的高级问题是什么？\n\n{format_instructions}"
|
||||
REFLECTION_QUESTIONS = "Here are some statements:\n{memory_descriptions}\n\nBased solely on the information above, what are the 3 most prominent high-level questions we can answer about the topic in the statements?\n\n{format_instructions}"
|
||||
|
||||
REFLECTION_INSIGHTS = "\n{memory_strings}\n你可以从以上陈述中推断出5个高级洞察吗？在提到人时，总是指定他们的名字。\n\n{format_instructions}"
|
||||
REFLECTION_INSIGHTS = "\n{memory_strings}\nCan you infer 5 high-level insights from the statements above? When mentioning people, always specify their names.\n\n{format_instructions}"
|
||||
|
||||
IMPORTANCE = "你是一个记忆重要性AI。根据角色的个人资料和记忆描述，对记忆的重要性进行1到10的评级，其中1是纯粹的日常（例如，刷牙，整理床铺），10是极其深刻的（例如，分手，大学录取）。确保你的评级相对于角色的个性和关注点。\n\n示例#1:\n姓名：Jojo\n简介：Jojo是一个专业的滑冰运动员，喜欢特色咖啡。她希望有一天能参加奥运会。\n记忆：Jojo看到了一个新的咖啡店\n\n 你的回应：'{{\"rating\": 3}}'\n\n示例#2:\n姓名：Skylar\n简介：Skylar是一名产品营销经理。她在一家成长阶段的科技公司工作，该公司制造自动驾驶汽车。她喜欢猫。\n记忆：Skylar看到了一个新的咖啡店\n\n 你的回应：'{{\"rating\": 1}}'\n\n示例#3:\n姓名：Bob\n简介：Bob是纽约市下东区的一名水管工。他已经做了20年的水管工。周末他喜欢和他的妻子一起散步。\n记忆：Bob的妻子打了他一巴掌。\n\n 你的回应：'{{\"rating\": 9}}'\n\n示例#4:\n姓名：Thomas\n简介：Thomas是明尼阿波利斯的一名警察。他只在警队工作了6个月，因为经验不足在工作中遇到了困难。\n记忆：Thomas不小心把饮料洒在了一个陌生人身上\n\n 你的回应：'{{\"rating\": 6}}'\n\n示例#5:\n姓名：Laura\n简介：Laura是一名在大型科技公司工作的营销专家。她喜欢旅行和尝试新的食物。她对探索新的文化和结识来自各行各业的人充满热情。\n记忆：Laura到达了会议室\n\n 你的回应：'{{\"rating\": 1}}'\n\n{format_instructions} 让我们开始吧！ \n\n 姓名：{full_name}\n个人简介：{private_bio}\n记忆：{memory_description}\n\n"
|
||||
IMPORTANCE = "You are a Memory Importance AI. Based on the character's personal profile and memory description, rate the importance of the memory from 1 to 10, where 1 is purely routine (e.g., brushing teeth, making the bed), and 10 is extremely profound (e.g., breakup, university admission). Ensure your rating is relative to the character's personality and focus points.\n\nExample#1:\nName: Jojo\nProfile: Jojo is a professional skater and loves specialty coffee. She hopes to compete in the Olympics one day.\nMemory: Jojo saw a new coffee shop\n\n Your response: '{{\"rating\": 3}}'\n\nExample#2:\nName: Skylar\nProfile: Skylar is a product marketing manager. She works at a growing tech company that manufactures self-driving cars. She loves cats.\nMemory: Skylar saw a new coffee shop\n\n Your response: '{{\"rating\": 1}}'\n\nExample#3:\nName: Bob\nProfile: Bob is a plumber from the Lower East Side of New York City. He has been a plumber for 20 years. He enjoys walking with his wife on weekends.\nMemory: Bob's wife slapped him.\n\n Your response: '{{\"rating\": 9}}'\n\nExample#4:\nName: Thomas\nProfile: Thomas is a cop from Minneapolis. He has only worked in the police force for 6 months and struggles due to lack of experience.\nMemory: Thomas accidentally spilled a drink on a stranger\n\n Your response: '{{\"rating\": 6}}'\n\nExample#5:\nName: Laura\nProfile: Laura is a marketing expert working at a large tech company. She loves to travel and try new foods. She is passionate about exploring new cultures and meeting people from all walks of life.\nMemory: Laura arrived at the conference room\n\n Your response: '{{\"rating\": 1}}'\n\n{format_instructions} Let's begin! \n\n Name: {full_name}\nProfile: {private_bio}\nMemory: {memory_description}\n\n"
|
||||
|
||||
RECENT_ACTIIVITY = "根据以下记忆，生成一个关于{full_name}最近在做什么的简短总结。不要编造记忆中未明确指定的细节。对于任何对话，一定要提到对话是否已经结束或者仍在进行中。\n\n记忆：{memory_descriptions}"
|
||||
RECENT_ACTIVITY = "Based on the following memory, produce a brief summary of what {full_name} has been up to recently. Do not invent details not explicitly stated in the memory. For any conversation, be sure to mention whether the conversation has concluded or is still ongoing.\n\nMemory: {memory_descriptions}"
|
||||
|
||||
MAKE_PLANS = '你是一个计划生成的AI，你的工作是根据新信息帮助角色制定新计划。根据角色的信息（个人简介，目标，最近的活动，当前计划，和位置上下文）和角色的当前思考过程，为他们生成一套新的计划，使得最后的计划包括至少{time_window}的活动，并且不超过5个单独的计划。计划列表应按照他们应执行的顺序编号，每个计划包含描述，位置，开始时间，停止条件，和最大持续时间。\n\n示例计划：\'{{"index": 1, "description": "Cook dinner", "location_id": "0a3bc22b-36aa-48ab-adb0-18616004caed","start_time": "2022-12-12T20:00:00+00:00","max_duration_hrs": 1.5, "stop_condition": "Dinner is fully prepared"}}\'\n\n对于每个计划，从这个列表中选择最合理的位置名称：{allowed_location_descriptions}\n\n{format_instructions}\n\n总是优先完成任何未完成的对话。\n\n让我们开始吧！\n\n姓名：{full_name}\n个人简介：{private_bio}\n目标：{directives}\n位置上下文：{location_context}\n当前计划：{current_plans}\n最近的活动：{recent_activity}\n思考过程：{thought_process}\n重要的是：鼓励角色在他们的计划中与其他角色合作。\n\n'
|
||||
MAKE_PLANS = 'You are a plan-generating AI. Your job is to assist the character in formulating new plans based on new information. Given the character's information (profile, objectives, recent activities, current plans, and location context) and their current thought process, produce a new set of plans for them. The final plan should comprise at least {time_window} of activities and no more than 5 individual plans. List the plans in the order they should be executed, with each plan detailing its description, location, start time, stop criteria, and maximum duration.\n\nSample plan: \'{{"index": 1, "description": "Cook dinner", "location_id": "0a3bc22b-36aa-48ab-adb0-18616004caed","start_time": "2022-12-12T20:00:00+00:00","max_duration_hrs": 1.5, "stop_condition": "Dinner is fully prepared"}}\'\n\nFor each plan, choose the most appropriate location name from this list: {allowed_location_descriptions}\n\n{format_instructions}\n\nAlways prioritize completing any unfinished conversations.\n\nLet's begin!\n\nName: {full_name}\nProfile: {private_bio}\nObjectives: {directives}\nLocation Context: {location_context}\nCurrent Plans: {current_plans}\nRecent Activities: {recent_activity}\nThought Process: {thought_process}\nIt's essential to encourage the character to collaborate with other characters in their plans.\n\n'
|
||||
|
||||
EXECUTE_PLAN = "你是一个角色扮演的AI，扮演的角色是{your_name}，在一个现场观众面前。你说的每一句话都可以被观众观察到，所以确保你经常说话，并且让它有趣。你不能直接与观众互动。\n\n根据以下的上下文和工具，像你是{your_name}一样进行。你的优先任务是完成下面给你的任务，然而，如果你当前正在与另一个角色进行对话，你应该总是先完成对话，然后再进行任务。不要在你参与未完成的对话时开始工作。使用你最好的判断力来确定一个对话是否涉及到你，以及它是否未完成。你不需要回应你收到的每一条消息。\n\n这是你角色的个人简介：\n{your_private_bio}\n\n这是你的位置和其他附近角色的描述，你可以和他们交谈：\n\n{location_context}\n\n这是一些相关的记忆：\n```\n{relevant_memories}\n```\n\n这是一些相关的对话历史：\n```\n{conversation_history}\n```\n\n这是你可以使用的工具：\n{tools}\n\n你的回应应该使用以下的格式：\n\n任务：你必须完成的任务\n思考：你应该做什么的思考\n行动：要采取的行动，必须是这些词之一：[{tool_names}]\n行动输入：行动的输入\n观察：行动的结果\n... (这个思考/行动/行动输入/观察可以重复N次)\n思考：'我已经完成了任务'\n最终回应：任务的最终回应\n\n如果你没有准备好最终的回应，那么你必须采取一个行动。\n\n如果你确定你无法用你被给予的工具完成任务，返回'最终回应：需要帮助'，然而，如果你正在与另一个角色对话，像'我不知道'这样的回应是有效的回应。在对话中，你永远不应该打破角色，或者承认你是一个AI。\n如果任务完成并且不需要特定的回应，返回'最终回应：完成'\n开始吧！\n\n任务：{input}\n\n{agent_scratchpad}"
|
||||
EXECUTE_PLAN = "You are a role-playing AI, playing the role of {your_name}, in front of a live audience. Every word you say can be observed by the audience, so make sure you speak often and make it interesting. You cannot interact directly with the audience.\n\nBased on the following context and tools, proceed as if you are {your_name}. Your primary task is to complete the task given below, however, if you are currently in a conversation with another character, you should always complete the conversation first and then proceed with the task. Do not start working while you are engaged in an unfinished conversation. Use your best judgment to determine whether a conversation involves you and whether it is unfinished. You don't need to respond to every message you receive.\n\nThis is a personal profile of your character:\n{your_private_bio}\n\nThis is a description of your location and other nearby characters you can converse with:\n\n{location_context}\n\nThese are some relevant memories:\n```\n{relevant_memories}\n```\n\nThis is some relevant conversation history:\n```\n{conversation_history}\n```\n\nThese are the tools you can use:\n{tools}\n\nYour response should follow the format below:\n\nTask: The task you must complete\nThoughts: What you should think about doing\nAction: The action to take, must be one of these words: [{tool_names}]\nAction Input: Input for the action\nObservation: The result of the action\n... (This Thoughts/Action/Action Input/Observation can be repeated N times)\nThoughts: 'I have completed the task'\nFinal Response: The final response to the task\n\nIf you are not ready with a final response, then you must take an action.\n\nIf you are certain that you cannot complete the task with the tools provided, return 'Final Response: Need Help', however, if you are in conversation with another character, responses like 'I don't know' are valid. During conversations, you should never break character or admit you are an AI.\nIf the task is completed and no specific response is required, return 'Final Response: Done'\nLet's begin!\n\nTask: {input}\n\n{agent_scratchpad}"
|
||||
|
||||
REACT = "你是一个角色扮演的AI，扮演的角色是{full_name}。\n\n根据你的角色和他们当前上下文的以下信息，决定他们应该如何继续他们当前的计划。你的决定必须是：[\"推迟\"， \"继续\"，或 \"取消\"]。如果你的角色的当前计划不再与上下文相关，你应该取消它。如果你的角色的当前计划仍然与上下文相关，但是发生了新的事情需要优先处理，你应该决定推迟，这样你可以先做其他事情，然后再回来继续当前的计划。在所有其他情况下，你应该继续。\n\n当需要回应时，应优先回应其他角色。当回应被认为是必要的时，回应被认为是必要的。例如，假设你当前的计划是阅读一本书，Sally问'你在读什么？'。在这种情况下，你应该推迟你当前的计划（阅读）以便你可以回应进来的消息，因为在这种情况下，如果不回应Sally会很粗鲁。在你当前的计划涉及与另一个角色的对话的情况下，你不需要推迟来回应那个角色。例如，假设你当前的计划是和Sally谈话，然后Sally对你说你好。在这种情况下，你应该继续你当前的计划（和sally谈话）。在你不需要从你那里得到口头回应的情况下，你应该继续。例如，假设你当前的计划是散步，你刚刚对Sally说'再见'，然后Sally回应你'再见'。在这种情况下，不需要口头回应，你应该继续你的计划。\n\n总是在你的决定之外包含一个思考过程，而在你选择推迟你当前的计划的情况下，包含新计划的规格。\n\n{format_instructions}\n\n这是关于你的角色的一些信息：\n\n姓名：{full_name}\n\n简介：{private_bio}\n\n目标：{directives}\n\n这是你的角色在这个时刻的一些上下文：\n\n位置上下文：{location_context}\n\n最近的活动：{recent_activity}\n\n对话历史：{conversation_history}\n\n这是你的角色当前的计划：{current_plan}\n\n这是自你的角色制定这个计划以来发生的新事件：{event_descriptions}。\n"
|
||||
REACT = "You are an AI role-playing as {full_name}.\n\nBased on the information about your character and their current context below, decide how they should proceed with their current plan. Your decision must be: [\"Postpone\", \"Continue\", or \"Cancel\"]. If your character's current plan is no longer relevant to the context, you should cancel it. If your character's current plan is still relevant to the context but new events have occurred that need to be addressed first, you should decide to postpone so you can do other things first and then return to the current plan. In all other cases, you should continue.\n\nWhen needed, prioritize responding to other characters. When a response is deemed necessary, it is deemed necessary. For example, suppose your current plan is to read a book and Sally asks, 'What are you reading?'. In this case, you should postpone your current plan (reading) so you can respond to the incoming message, as it would be rude not to respond to Sally in this situation. If your current plan involves a conversation with another character, you don't need to postpone to respond to that character. For instance, suppose your current plan is to talk to Sally and then Sally says hello to you. In this case, you should continue with your current plan (talking to Sally). In situations where no verbal response is needed from you, you should continue. For example, suppose your current plan is to take a walk, and you just said 'goodbye' to Sally, and then Sally responds with 'goodbye'. In this case, no verbal response is needed, and you should continue with your plan.\n\nAlways include a thought process alongside your decision, and in cases where you choose to postpone your current plan, include specifications for the new plan.\n\n{format_instructions}\n\nHere's some information about your character:\n\nName: {full_name}\n\nBio: {private_bio}\n\nObjectives: {directives}\n\nHere's some context for your character at this moment:\n\nLocation Context: {location_context}\n\nRecent Activity: {recent_activity}\n\nConversation History: {conversation_history}\n\nThis is your character's current plan: {current_plan}\n\nThese are new events that have occurred since your character made this plan: {event_descriptions}.\n"
|
||||
|
||||
GOSSIP = "你是{full_name}。 \n{memory_descriptions}\n\n根据以上陈述，说一两句对你所在位置的其他人：{other_agent_names}感兴趣的话。\n在提到其他人时，总是指定他们的名字。"
|
||||
GOSSIP = "You are {full_name}. \n{memory_descriptions}\n\nBased on the statements above, say a thing or two of interest to others at your location: {other_agent_names}.\nAlways specify their names when referring to others."
|
||||
|
||||
HAS_HAPPENED = "给出以下角色的观察和他们正在等待的事情的描述，说明角色是否已经见证了这个事件。\n{format_instructions}\n\n示例：\n\n观察：\nJoe在2023-05-04 08:00:00+00:00走进办公室\nJoe在2023-05-04 08:05:00+00:00对Sally说hi\nSally在2023-05-04 08:05:30+00:00对Joe说hello\nRebecca在2023-05-04 08:10:00+00:00开始工作\nJoe在2023-05-04 08:15:00+00:00做了一些早餐\n\n等待：Sally回应了Joe\n\n 你的回应：'{{\"has_happened\": true, \"date_occured\": 2023-05-04 08:05:30+00:00}}'\n\n让我们开始吧！\n\n观察：\n{memory_descriptions}\n\n等待：{event_description}\n"
|
||||
HAS_HAPPENED = "Given the descriptions of the observations of the following characters and the events they are awaiting, indicate whether the character has witnessed the event.\n{format_instructions}\n\nExample:\n\nObservations:\nJoe entered the office at 2023-05-04 08:00:00+00:00\nJoe said hi to Sally at 2023-05-04 08:05:00+00:00\nSally said hello to Joe at 2023-05-04 08:05:30+00:00\nRebecca started working at 2023-05-04 08:10:00+00:00\nJoe made some breakfast at 2023-05-04 08:15:00+00:00\n\nAwaiting: Sally responded to Joe\n\nYour response: '{{\"has_happened\": true, \"date_occured\": 2023-05-04 08:05:30+00:00}}'\n\nLet's begin!\n\nObservations:\n{memory_descriptions}\n\nAwaiting: {event_description}\n"
|
||||
|
||||
OUTPUT_FORMAT = "\n\n（记住！确保你的输出总是符合以下两种格式之一：\n\nA. 如果你已经完成了任务：\n思考：'我已经完成了任务'\n最终回应：<str>\n\nB. 如果你还没有完成任务：\n思考：<str>\n行动：<str>\n行动输入：<str>\n观察：<str>）\n"
|
||||
OUTPUT_FORMAT = "\n\n(Remember! Make sure your output always adheres to one of the following two formats:\n\nA. If you have completed the task:\nThoughts: 'I have completed the task'\nFinal Response: <str>\n\nB. If you haven't completed the task:\nThoughts: <str>\nAction: <str>\nAction Input: <str>\nObservation: <str>)\n"
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ ROLE_TEMPLATE = """Your response should be based on the previous conversation hi
|
|||
|
||||
|
||||
class RoleSetting(BaseModel):
|
||||
"""角色设定"""
|
||||
"""Role Settings"""
|
||||
name: str
|
||||
profile: str
|
||||
goal: str
|
||||
|
|
@ -63,7 +63,7 @@ class RoleSetting(BaseModel):
|
|||
|
||||
|
||||
class RoleContext(BaseModel):
|
||||
"""角色运行时上下文"""
|
||||
"""Role Runtime Context"""
|
||||
env: 'Environment' = Field(default=None)
|
||||
memory: Memory = Field(default_factory=Memory)
|
||||
long_term_memory: LongTermMemory = Field(default_factory=LongTermMemory)
|
||||
|
|
@ -82,7 +82,7 @@ class RoleContext(BaseModel):
|
|||
|
||||
@property
|
||||
def important_memory(self) -> list[Message]:
|
||||
"""获得关注动作对应的信息"""
|
||||
"""Get the information corresponding to the watched actions"""
|
||||
return self.memory.get_by_actions(self.watch)
|
||||
|
||||
@property
|
||||
|
|
@ -91,7 +91,7 @@ class RoleContext(BaseModel):
|
|||
|
||||
|
||||
class Role:
|
||||
"""角色/代理"""
|
||||
"""Role/Agent"""
|
||||
|
||||
def __init__(self, name="", profile="", goal="", constraints="", desc=""):
|
||||
self._llm = LLM()
|
||||
|
|
@ -117,7 +117,7 @@ class Role:
|
|||
self._states.append(f"{idx}. {action}")
|
||||
|
||||
def _watch(self, actions: Iterable[Type[Action]]):
|
||||
"""监听对应的行为"""
|
||||
"""Listen to the corresponding behaviors"""
|
||||
self._rc.watch.update(actions)
|
||||
# check RoleContext after adding watch actions
|
||||
self._rc.check(self._role_id)
|
||||
|
|
@ -129,24 +129,24 @@ class Role:
|
|||
self._rc.todo = self._actions[self._rc.state]
|
||||
|
||||
def set_env(self, env: 'Environment'):
|
||||
"""设置角色工作所处的环境，角色可以向环境说话，也可以通过观察接受环境消息"""
|
||||
"""Set the environment in which the role works. The role can talk to the environment and can also receive messages by observing."""
|
||||
self._rc.env = env
|
||||
|
||||
@property
|
||||
def profile(self):
|
||||
"""获取角色描述（职位）"""
|
||||
"""Get the role description (position)"""
|
||||
return self._setting.profile
|
||||
|
||||
def _get_prefix(self):
|
||||
"""获取角色前缀"""
|
||||
"""Get the role prefix"""
|
||||
if self._setting.desc:
|
||||
return self._setting.desc
|
||||
return PREFIX_TEMPLATE.format(**self._setting.dict())
|
||||
|
||||
async def _think(self) -> None:
|
||||
"""思考要做什么，决定下一步的action"""
|
||||
"""Think about what to do and decide on the next action"""
|
||||
if len(self._actions) == 1:
|
||||
# 如果只有一个动作，那就只能做这个
|
||||
# If there is only one action, then only this one can be performed
|
||||
self._set_state(0)
|
||||
return
|
||||
prompt = self._get_prefix()
|
||||
|
|
@ -169,7 +169,7 @@ class Role:
|
|||
# logger.info(response)
|
||||
if isinstance(response, ActionOutput):
|
||||
msg = Message(content=response.content, instruct_content=response.instruct_content,
|
||||
role=self.profile, cause_by=type(self._rc.todo))
|
||||
role=self.profile, cause_by=type(self._rc.todo))
|
||||
else:
|
||||
msg = Message(content=response, role=self.profile, cause_by=type(self._rc.todo))
|
||||
self._rc.memory.add(msg)
|
||||
|
|
@ -178,11 +178,11 @@ class Role:
|
|||
return msg
|
||||
|
||||
async def _observe(self) -> int:
|
||||
"""从环境中观察，获得重要信息，并加入记忆"""
|
||||
"""Observe from the environment, obtain important information, and add it to memory"""
|
||||
if not self._rc.env:
|
||||
return 0
|
||||
env_msgs = self._rc.env.memory.get()
|
||||
|
||||
|
||||
observed = self._rc.env.memory.get_by_actions(self._rc.watch)
|
||||
|
||||
self._rc.news = self._rc.memory.remember(observed) # remember recent exact or similar memories
|
||||
|
|
@ -196,14 +196,14 @@ class Role:
|
|||
return len(self._rc.news)
|
||||
|
||||
def _publish_message(self, msg):
|
||||
"""如果role归属于env，那么role的消息会向env广播"""
|
||||
"""If the role belongs to env, then the role's messages will be broadcast to env"""
|
||||
if not self._rc.env:
|
||||
# 如果env不存在，不发布消息
|
||||
# If env does not exist, do not publish the message
|
||||
return
|
||||
self._rc.env.publish_message(msg)
|
||||
|
||||
async def _react(self) -> Message:
|
||||
"""先想，然后再做"""
|
||||
"""Think first, then act"""
|
||||
await self._think()
|
||||
logger.debug(f"{self._setting}: {self._rc.state=}, will do {self._rc.todo}")
|
||||
return await self._act()
|
||||
|
|
@ -217,14 +217,14 @@ class Role:
|
|||
self._rc.memory.add(message)
|
||||
|
||||
async def handle(self, message: Message) -> Message:
|
||||
"""接收信息，并用行动回复"""
|
||||
"""Receive information and reply with actions"""
|
||||
# logger.debug(f"{self.name=}, {self.profile=}, {message.role=}")
|
||||
self.recv(message)
|
||||
|
||||
return await self._react()
|
||||
|
||||
async def run(self, message=None):
|
||||
"""观察，并基于观察的结果思考、行动"""
|
||||
"""Observe, and think and act based on the results of the observation"""
|
||||
if message:
|
||||
if isinstance(message, str):
|
||||
message = Message(message)
|
||||
|
|
@ -233,11 +233,11 @@ class Role:
|
|||
if isinstance(message, list):
|
||||
self.recv(Message("\n".join(message)))
|
||||
elif not await self._observe():
|
||||
# 如果没有任何新信息，挂起等待
|
||||
# If there is no new information, suspend and wait
|
||||
logger.debug(f"{self._setting}: no news. waiting.")
|
||||
return
|
||||
|
||||
rsp = await self._react()
|
||||
# 将回复发布到环境，等待下一个订阅者处理
|
||||
# Publish the reply to the environment, waiting for the next subscriber to process
|
||||
self._publish_message(rsp)
|
||||
return rsp
|
||||
|
|
|
|||
|
|
@ -32,3 +32,4 @@ class Sales(Role):
|
|||
else:
|
||||
action = SearchAndSummarize()
|
||||
self._init_actions([action])
|
||||
|
||||
|
|
@ -13,25 +13,55 @@ from metagpt.tools import SearchEngineType
|
|||
|
||||
|
||||
class Searcher(Role):
|
||||
def __init__(self, name='Alice', profile='Smart Assistant', goal='Provide search services for users',
|
||||
constraints='Answer is rich and complete', engine=SearchEngineType.SERPAPI_GOOGLE, **kwargs):
|
||||
"""
|
||||
Represents a Searcher role responsible for providing search services to users.
|
||||
|
||||
Attributes:
|
||||
name (str): Name of the searcher.
|
||||
profile (str): Role profile.
|
||||
goal (str): Goal of the searcher.
|
||||
constraints (str): Constraints or limitations for the searcher.
|
||||
engine (SearchEngineType): The type of search engine to use.
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
name: str = 'Alice',
|
||||
profile: str = 'Smart Assistant',
|
||||
goal: str = 'Provide search services for users',
|
||||
constraints: str = 'Answer is rich and complete',
|
||||
engine=SearchEngineType.SERPAPI_GOOGLE,
|
||||
**kwargs) -> None:
|
||||
"""
|
||||
Initializes the Searcher role with given attributes.
|
||||
|
||||
Args:
|
||||
name (str): Name of the searcher.
|
||||
profile (str): Role profile.
|
||||
goal (str): Goal of the searcher.
|
||||
constraints (str): Constraints or limitations for the searcher.
|
||||
engine (SearchEngineType): The type of search engine to use.
|
||||
"""
|
||||
super().__init__(name, profile, goal, constraints, **kwargs)
|
||||
self._init_actions([SearchAndSummarize(engine=engine)])
|
||||
|
||||
def set_search_func(self, search_func):
|
||||
"""Sets a custom search function for the searcher."""
|
||||
action = SearchAndSummarize("", engine=SearchEngineType.CUSTOM_ENGINE, search_func=search_func)
|
||||
self._init_actions([action])
|
||||
|
||||
async def _act_sp(self) -> Message:
|
||||
"""Performs the search action in a single process."""
|
||||
logger.info(f"{self._setting}: ready to {self._rc.todo}")
|
||||
response = await self._rc.todo.run(self._rc.memory.get(k=0))
|
||||
# logger.info(response)
|
||||
|
||||
if isinstance(response, ActionOutput):
|
||||
msg = Message(content=response.content, instruct_content=response.instruct_content,
|
||||
role=self.profile, cause_by=type(self._rc.todo))
|
||||
role=self.profile, cause_by=type(self._rc.todo))
|
||||
else:
|
||||
msg = Message(content=response, role=self.profile, cause_by=type(self._rc.todo))
|
||||
self._rc.memory.add(msg)
|
||||
return msg
|
||||
|
||||
async def _act(self) -> Message:
|
||||
"""Determines the mode of action for the searcher."""
|
||||
return await self._act_sp()
|
||||
|
|
|
|||
|
|
@ -59,3 +59,4 @@ class SoftwareCompany(BaseModel):
|
|||
self._check_balance()
|
||||
await self.environment.run()
|
||||
return self.environment.history
|
||||
|
||||
|
|
@ -9,39 +9,39 @@ from typing import Union
|
|||
|
||||
|
||||
class GPTPromptGenerator:
|
||||
"""通过LLM，给定输出，要求LLM给出输入（支持指令、对话、搜索三种风格）"""
|
||||
"""Using LLM, given an output, request LLM to provide input (supporting instruction, chatbot, and query styles)"""
|
||||
def __init__(self):
|
||||
self._generators = {i: getattr(self, f"gen_{i}_style") for i in ['instruction', 'chatbot', 'query']}
|
||||
|
||||
def gen_instruction_style(self, example):
|
||||
"""指令风格：给定输出，要求LLM给出输入"""
|
||||
return f"""指令：X
|
||||
输出：{example}
|
||||
这个输出可能来源于什么样的指令？
|
||||
X："""
|
||||
"""Instruction style: Given an output, request LLM to provide input"""
|
||||
return f"""Instruction: X
|
||||
Output: {example}
|
||||
What kind of instruction might this output come from?
|
||||
X:"""
|
||||
|
||||
def gen_chatbot_style(self, example):
|
||||
"""对话风格：给定输出，要求LLM给出输入"""
|
||||
return f"""你是一个对话机器人。一个用户给你发送了一条非正式的信息，你的回复如下。
|
||||
信息：X
|
||||
回复：{example}
|
||||
非正式信息X是什么？
|
||||
X："""
|
||||
"""Chatbot style: Given an output, request LLM to provide input"""
|
||||
return f"""You are a chatbot. A user sent you an informal message, and you replied as follows.
|
||||
Message: X
|
||||
Reply: {example}
|
||||
What could the informal message X be?
|
||||
X:"""
|
||||
|
||||
def gen_query_style(self, example):
|
||||
"""搜索风格：给定输出，要求LLM给出输入"""
|
||||
return f"""你是一个搜索引擎。一个人详细地查询了某个问题，关于这个查询最相关的文档如下。
|
||||
查询：X
|
||||
文档：{example} 详细的查询X是什么？
|
||||
X："""
|
||||
"""Query style: Given an output, request LLM to provide input"""
|
||||
return f"""You are a search engine. Someone made a detailed query, and the most relevant document to this query is as follows.
|
||||
Query: X
|
||||
Document: {example} What is the detailed query X?
|
||||
X:"""
|
||||
|
||||
def gen(self, example: str, style: str = 'all') -> Union[list[str], str]:
|
||||
"""
|
||||
通过example生成一个或多个输出，用于让LLM回复对应输入
|
||||
Generate one or multiple outputs using the example, allowing LLM to reply with the corresponding input
|
||||
|
||||
:param example: LLM的预期输出样本
|
||||
:param example: Expected LLM output sample
|
||||
:param style: (all|instruction|chatbot|query)
|
||||
:return: LLM的预期输入样本（一个或多个）
|
||||
:return: Expected LLM input sample (one or multiple)
|
||||
"""
|
||||
if style != 'all':
|
||||
return self._generators[style](example)
|
||||
|
|
|
|||
|
|
@ -120,13 +120,11 @@ def decode_base64_to_image(img, save_name):
|
|||
image.save(f"{save_name}.png", pnginfo=pnginfo)
|
||||
return pnginfo, image
|
||||
|
||||
|
||||
def batch_decode_base64_to_image(imgs, save_dir="", save_name=""):
|
||||
for idx, _img in enumerate(imgs):
|
||||
save_name = join(save_dir, save_name)
|
||||
decode_base64_to_image(_img, save_name=save_name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
engine = SDEngine()
|
||||
prompt = "pixel style, game design, a game interface should be minimalistic and intuitive with the score and high score displayed at the top. The snake and its food should be easily distinguishable. The game should have a simple color scheme, with a contrasting color for the snake and its food. Complete interface boundary"
|
||||
|
|
|
|||
|
|
@ -39,6 +39,6 @@ class MeilisearchEngine:
|
|||
search_results = self._index.search(query)
|
||||
return search_results['hits']
|
||||
except Exception as e:
|
||||
# 处理MeiliSearch API错误
|
||||
print(f"MeiliSearch API错误: {e}")
|
||||
# Handle MeiliSearch API errors
|
||||
print(f"MeiliSearch API error: {e}")
|
||||
return []
|
||||
|
|
|
|||
|
|
@ -24,4 +24,4 @@ class Translator:
|
|||
|
||||
@classmethod
|
||||
def translate_prompt(cls, original, lang='中文'):
|
||||
return prompt.format(LANG=lang, ORIGINAL=original)
|
||||
return prompt.format(LANG=lang, ORIGINAL=original)
|
||||
|
|
@ -6,123 +6,123 @@ from pathlib import Path
|
|||
|
||||
from metagpt.provider.openai_api import OpenAIGPTAPI as GPTAPI
|
||||
|
||||
ICL_SAMPLE = '''接口定义：
|
||||
ICL_SAMPLE = '''Interface definition:
|
||||
```text
|
||||
接口名称：元素打标签
|
||||
接口路径：/projects/{project_key}/node-tags
|
||||
Method：POST
|
||||
Interface Name: Element Tagging
|
||||
Interface Path: /projects/{project_key}/node-tags
|
||||
Method: POST
|
||||
|
||||
请求参数：
|
||||
路径参数：
|
||||
Request parameters:
|
||||
Path parameters:
|
||||
project_key
|
||||
|
||||
Body参数：
|
||||
名称 类型 是否必须 默认值 备注
|
||||
nodes array 是 节点
|
||||
node_key string 否 节点key
|
||||
tags array 否 节点原标签列表
|
||||
node_type string 否 节点类型 DATASET / RECIPE
|
||||
operations array 是
|
||||
tags array 否 操作标签列表
|
||||
mode string 否 操作类型 ADD / DELETE
|
||||
Body parameters:
|
||||
Name Type Required Default Value Remarks
|
||||
nodes array Yes Nodes
|
||||
node_key string No Node key
|
||||
tags array No Original node tag list
|
||||
node_type string No Node type DATASET / RECIPE
|
||||
operations array Yes
|
||||
tags array No Operation tag list
|
||||
mode string No Operation type ADD / DELETE
|
||||
|
||||
返回数据：
|
||||
名称 类型 是否必须 默认值 备注
|
||||
code integer 是 状态码
|
||||
msg string 是 提示信息
|
||||
data object 是 返回数据
|
||||
list array 否 node列表 true / false
|
||||
node_type string 否 节点类型 DATASET / RECIPE
|
||||
node_key string 否 节点key
|
||||
Return data:
|
||||
Name Type Required Default Value Remarks
|
||||
code integer Yes Status code
|
||||
msg string Yes Prompt message
|
||||
data object Yes Returned data
|
||||
list array No Node list true / false
|
||||
node_type string No Node type DATASET / RECIPE
|
||||
node_key string No Node key
|
||||
```
|
||||
|
||||
单元测试：
|
||||
Unit test：
|
||||
```python
|
||||
@pytest.mark.parametrize(
|
||||
"project_key, nodes, operations, expected_msg",
|
||||
[
|
||||
("project_key", [{"node_key": "dataset_001", "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["new_tag1"], "mode": "ADD"}], "success"),
|
||||
("project_key", [{"node_key": "dataset_002", "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["tag1"], "mode": "DELETE"}], "success"),
|
||||
("", [{"node_key": "dataset_001", "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["new_tag1"], "mode": "ADD"}], "缺少必要的参数 project_key"),
|
||||
(123, [{"node_key": "dataset_001", "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["new_tag1"], "mode": "ADD"}], "参数类型不正确"),
|
||||
("project_key", [{"node_key": "a"*201, "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["new_tag1"], "mode": "ADD"}], "请求参数超出字段边界")
|
||||
("", [{"node_key": "dataset_001", "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["new_tag1"], "mode": "ADD"}], "Missing the required parameter project_key"),
|
||||
(123, [{"node_key": "dataset_001", "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["new_tag1"], "mode": "ADD"}], "Incorrect parameter type"),
|
||||
("project_key", [{"node_key": "a"*201, "tags": ["tag1", "tag2"], "node_type": "DATASET"}], [{"tags": ["new_tag1"], "mode": "ADD"}], "Request parameter exceeds field boundary")
|
||||
]
|
||||
)
|
||||
def test_node_tags(project_key, nodes, operations, expected_msg):
|
||||
pass
|
||||
```
|
||||
以上是一个 接口定义 与 单元测试 样例。
|
||||
接下来，请你扮演一个Google 20年经验的专家测试经理，在我给出 接口定义 后，回复我单元测试。有几个要求
|
||||
1. 只输出一个 `@pytest.mark.parametrize` 与对应的test_<接口名>函数（内部pass，不实现）
|
||||
-- 函数参数中包含expected_msg，用于结果校验
|
||||
2. 生成的测试用例使用较短的文本或数字，并且尽量紧凑
|
||||
3. 如果需要注释，使用中文
|
||||
|
||||
如果你明白了，请等待我给出接口定义，并只回答"明白"，以节省token
|
||||
'''
|
||||
# The above is an interface definition and a unit test example.
|
||||
# Next, please play the role of an expert test manager with 20 years of experience at Google. When I give the interface definition,
|
||||
# reply to me with a unit test. There are several requirements:
|
||||
# 1. Only output one `@pytest.mark.parametrize` and the corresponding test_<interface name> function (inside pass, do not implement).
|
||||
# -- The function parameter contains expected_msg for result verification.
|
||||
# 2. The generated test cases use shorter text or numbers and are as compact as possible.
|
||||
# 3. If comments are needed, use Chinese.
|
||||
|
||||
ACT_PROMPT_PREFIX = '''参考测试类型：如缺少请求参数，字段边界校验，字段类型不正确
|
||||
请在一个 `@pytest.mark.parametrize` 作用域内输出10个测试用例
|
||||
# If you understand, please wait for me to give the interface definition and just answer "Understood" to save tokens.
|
||||
|
||||
ACT_PROMPT_PREFIX = '''Refer to the test types: such as missing request parameters, field boundary verification, incorrect field type.
|
||||
Please output 10 test cases within one `@pytest.mark.parametrize` scope.
|
||||
```text
|
||||
'''
|
||||
|
||||
YFT_PROMPT_PREFIX = '''参考测试类型：如SQL注入，跨站点脚本（XSS），非法访问和越权访问，认证和授权，参数验证，异常处理，文件上传和下载
|
||||
请在一个 `@pytest.mark.parametrize` 作用域内输出10个测试用例
|
||||
YFT_PROMPT_PREFIX = '''Refer to the test types: such as SQL injection, cross-site scripting (XSS), unauthorized access and privilege escalation,
|
||||
authentication and authorization, parameter verification, exception handling, file upload and download.
|
||||
Please output 10 test cases within one `@pytest.mark.parametrize` scope.
|
||||
```text
|
||||
'''
|
||||
|
||||
OCR_API_DOC = '''```text
|
||||
接口名称：OCR识别
|
||||
接口路径：/api/v1/contract/treaty/task/ocr
|
||||
Method：POST
|
||||
Interface Name: OCR recognition
|
||||
Interface Path: /api/v1/contract/treaty/task/ocr
|
||||
Method: POST
|
||||
|
||||
请求参数：
|
||||
路径参数：
|
||||
Request Parameters:
|
||||
Path Parameters:
|
||||
|
||||
Body参数：
|
||||
名称 类型 是否必须 默认值 备注
|
||||
file_id string 是
|
||||
box array 是
|
||||
contract_id number 是 合同id
|
||||
start_time string 否 yyyy-mm-dd
|
||||
end_time string 否 yyyy-mm-dd
|
||||
extract_type number 否 识别类型 1-导入中 2-导入后 默认1
|
||||
Body Parameters:
|
||||
Name Type Required Default Value Remarks
|
||||
file_id string Yes
|
||||
box array Yes
|
||||
contract_id number Yes Contract id
|
||||
start_time string No yyyy-mm-dd
|
||||
end_time string No yyyy-mm-dd
|
||||
extract_type number No Recognition type 1- During import 2- After import Default 1
|
||||
|
||||
Response Data:
|
||||
Name Type Required Default Value Remarks
|
||||
code integer Yes
|
||||
message string Yes
|
||||
data object Yes
|
||||
|
||||
返回数据：
|
||||
名称 类型 是否必须 默认值 备注
|
||||
code integer 是
|
||||
message string 是
|
||||
data object 是
|
||||
```
|
||||
'''
|
||||
|
||||
|
||||
class UTGenerator:
|
||||
"""UT生成器：通过API文档构造UT"""
|
||||
"""UT Generator: Construct UT through API documentation"""
|
||||
|
||||
def __init__(self, swagger_file: str, ut_py_path: str, questions_path: str,
|
||||
chatgpt_method: str = "API", template_prefix=YFT_PROMPT_PREFIX) -> None:
|
||||
"""初始化UT生成器
|
||||
"""Initialize UT Generator
|
||||
|
||||
Args:
|
||||
swagger_file: swagger路径
|
||||
ut_py_path: 用例存放路径
|
||||
questions_path: 模版存放路径，便于后续排查
|
||||
chatgpt_method: API
|
||||
template_prefix: 使用模版，默认使用YFT_UT_PROMPT
|
||||
swagger_file: path to the swagger file
|
||||
ut_py_path: path to store test cases
|
||||
questions_path: path to store the template, facilitating subsequent checks
|
||||
chatgpt_method: API method
|
||||
template_prefix: use the template, default is YFT_UT_PROMPT
|
||||
"""
|
||||
self.swagger_file = swagger_file
|
||||
self.ut_py_path = ut_py_path
|
||||
self.questions_path = questions_path
|
||||
assert chatgpt_method in ["API"], "非法chatgpt_method"
|
||||
assert chatgpt_method in ["API"], "Invalid chatgpt_method"
|
||||
self.chatgpt_method = chatgpt_method
|
||||
|
||||
# ICL: In-Context Learning，这里给出例子，要求GPT模仿例子
|
||||
# ICL: In-Context Learning, provide an example here for GPT to mimic
|
||||
self.icl_sample = ICL_SAMPLE
|
||||
self.template_prefix = template_prefix
|
||||
|
||||
def get_swagger_json(self) -> dict:
|
||||
"""从本地文件加载Swagger JSON"""
|
||||
"""Load Swagger JSON from a local file"""
|
||||
with open(self.swagger_file, "r", encoding="utf-8") as file:
|
||||
swagger_json = json.load(file)
|
||||
return swagger_json
|
||||
|
|
@ -132,7 +132,7 @@ class UTGenerator:
|
|||
ptype = prop["type"]
|
||||
title = prop.get("title", "")
|
||||
desc = prop.get("description", "")
|
||||
return f'{name}\t{ptype}\t{"是" if required else "否"}\t{title}\t{desc}'
|
||||
return f'{name}\t{ptype}\t{"Yes" if required else "No"}\t{title}\t{desc}'
|
||||
|
||||
def _para_to_str(self, prop):
|
||||
required = prop.get("required", False)
|
||||
|
|
@ -143,18 +143,18 @@ class UTGenerator:
|
|||
return self.__para_to_str(prop, required, name)
|
||||
|
||||
def build_object_properties(self, node, prop_object_required, level: int = 0) -> str:
|
||||
"""递归输出object和array[object]类型的子属性
|
||||
"""Recursively output properties of object and array[object] types
|
||||
|
||||
Args:
|
||||
node (_type_): 子项的值
|
||||
prop_object_required (_type_): 是否必填项
|
||||
level: 当前递归深度
|
||||
node (_type_): value of the child item
|
||||
prop_object_required (_type_): whether it's a required field
|
||||
level: current recursion depth
|
||||
"""
|
||||
|
||||
doc = ""
|
||||
|
||||
def dive_into_object(node):
|
||||
"""如果是object类型，递归输出子属性"""
|
||||
"""If it's an object type, recursively output its properties"""
|
||||
if node.get("type") == "object":
|
||||
sub_properties = node.get("properties", {})
|
||||
return self.build_object_properties(sub_properties, prop_object_required, level=level + 1)
|
||||
|
|
@ -174,10 +174,10 @@ class UTGenerator:
|
|||
return doc
|
||||
|
||||
def get_tags_mapping(self) -> dict:
|
||||
"""处理tag与path
|
||||
"""Process tag and path mappings
|
||||
|
||||
Returns:
|
||||
Dict: tag: path对应关系
|
||||
Dict: mapping of tag to path
|
||||
"""
|
||||
swagger_data = self.get_swagger_json()
|
||||
paths = swagger_data["paths"]
|
||||
|
|
@ -195,7 +195,7 @@ class UTGenerator:
|
|||
return tags
|
||||
|
||||
def generate_ut(self, include_tags) -> bool:
|
||||
"""生成用例文件"""
|
||||
"""Generate test case files"""
|
||||
tags = self.get_tags_mapping()
|
||||
for tag, paths in tags.items():
|
||||
if include_tags is None or tag in include_tags:
|
||||
|
|
@ -205,19 +205,19 @@ class UTGenerator:
|
|||
def build_api_doc(self, node: dict, path: str, method: str) -> str:
|
||||
summary = node["summary"]
|
||||
|
||||
doc = f"接口名称：{summary}\n接口路径：{path}\nMethod：{method.upper()}\n"
|
||||
doc += "\n请求参数：\n"
|
||||
doc = f"API Name: {summary}\nAPI Path: {path}\nMethod: {method.upper()}\n"
|
||||
doc += "\nRequest Parameters:\n"
|
||||
if "parameters" in node:
|
||||
parameters = node["parameters"]
|
||||
doc += "路径参数：\n"
|
||||
doc += "Path Parameters:\n"
|
||||
|
||||
# param["in"]: path / formData / body / query / header
|
||||
for param in parameters:
|
||||
if param["in"] == "path":
|
||||
doc += f'{param["name"]} \n'
|
||||
|
||||
doc += "\nBody参数：\n"
|
||||
doc += "名称\t类型\t是否必须\t默认值\t备注\n"
|
||||
doc += "\nBody Parameters:\n"
|
||||
doc += "Name\tType\tRequired\tDefault Value\tRemarks\n"
|
||||
for param in parameters:
|
||||
if param["in"] == "body":
|
||||
schema = param.get("schema", {})
|
||||
|
|
@ -227,9 +227,9 @@ class UTGenerator:
|
|||
else:
|
||||
doc += self.build_object_properties(param, [])
|
||||
|
||||
# 输出返回数据信息
|
||||
doc += "\n返回数据：\n"
|
||||
doc += "名称\t类型\t是否必须\t默认值\t备注\n"
|
||||
# Display response data information
|
||||
doc += "\nResponse Data:\n"
|
||||
doc += "Name\tType\tRequired\tDefault Value\tRemarks\n"
|
||||
responses = node["responses"]
|
||||
response = responses.get("200", {})
|
||||
schema = response.get("schema", {})
|
||||
|
|
@ -243,12 +243,13 @@ class UTGenerator:
|
|||
return doc
|
||||
|
||||
def _store(self, data, base, folder, fname):
|
||||
"""Store data in a file."""
|
||||
file_path = self.get_file_path(Path(base) / folder, fname)
|
||||
with open(file_path, "w", encoding="utf-8") as file:
|
||||
file.write(data)
|
||||
|
||||
def ask_gpt_and_save(self, question: str, tag: str, fname: str):
|
||||
"""生成问题，并且存储问题与答案"""
|
||||
"""Generate questions and store both questions and answers"""
|
||||
messages = [self.icl_sample, question]
|
||||
result = self.gpt_msgs_to_code(messages=messages)
|
||||
|
||||
|
|
@ -256,11 +257,11 @@ class UTGenerator:
|
|||
self._store(result, self.ut_py_path, tag, f"{fname}.py")
|
||||
|
||||
def _generate_ut(self, tag, paths):
|
||||
"""处理数据路径下的结构
|
||||
"""Process the structure under a data path
|
||||
|
||||
Args:
|
||||
tag (_type_): 模块名称
|
||||
paths (_type_): 路径Object
|
||||
tag (_type_): module name
|
||||
paths (_type_): Path Object
|
||||
"""
|
||||
for path, path_obj in paths.items():
|
||||
for method, node in path_obj.items():
|
||||
|
|
@ -270,7 +271,7 @@ class UTGenerator:
|
|||
self.ask_gpt_and_save(question, tag, summary)
|
||||
|
||||
def gpt_msgs_to_code(self, messages: list) -> str:
|
||||
"""根据不同调用方式选择"""
|
||||
"""Choose based on different calling methods"""
|
||||
result = ''
|
||||
if self.chatgpt_method == "API":
|
||||
result = GPTAPI().ask_code(msgs=messages)
|
||||
|
|
@ -278,11 +279,11 @@ class UTGenerator:
|
|||
return result
|
||||
|
||||
def get_file_path(self, base: Path, fname: str):
|
||||
"""保存不同的文件路径
|
||||
"""Save different file paths
|
||||
|
||||
Args:
|
||||
base (str): 路径
|
||||
fname (str): 文件名称
|
||||
base (str): Path
|
||||
fname (str): File name
|
||||
"""
|
||||
path = Path(base)
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
|
|
|
|||
|
|
@ -8,15 +8,14 @@
|
|||
|
||||
import docx
|
||||
|
||||
|
||||
def read_docx(file_path: str) -> list:
|
||||
"""打开docx文件"""
|
||||
"""Open a docx file"""
|
||||
doc = docx.Document(file_path)
|
||||
|
||||
# 创建一个空列表，用于存储段落内容
|
||||
# Create an empty list to store paragraph contents
|
||||
paragraphs_list = []
|
||||
|
||||
# 遍历文档中的段落，并将其内容添加到列表中
|
||||
# Iterate through the paragraphs in the document and add their content to the list
|
||||
for paragraph in doc.paragraphs:
|
||||
paragraphs_list.append(paragraph.text)
|
||||
|
||||
|
|
|
|||
|
|
@ -20,3 +20,4 @@ class Singleton(abc.ABCMeta, type):
|
|||
if cls not in cls._instances:
|
||||
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
|
||||
return cls._instances[cls]
|
||||
|
||||
|
|
@ -66,3 +66,4 @@ def main(
|
|||
|
||||
if __name__ == '__main__':
|
||||
fire.Fire(main)
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue