#1.python文件执行。创建docker环境执行.py文件（篇幅所限，仅摘取部分代码） def execute_python_file(file): """Execute a Python file in a Docker container and return the output"""
print (f"Executing file '{file}' in workspace '{WORKSPACE_FOLDER}'")
if not file.endswith(".py"): return "Error: Invalid file type. Only .py files are allowed."
file_path = os.path.join(WORKSPACE_FOLDER, file)
if not os.path.isfile(file_path): return f"Error: File '{file}' does not exist."
try: client = docker.from_env()
image_name = 'python:3.10' try: client.images.get(image_name) print(f"Image '{image_name}' found locally") except docker.errors.ImageNotFound: print(f"Image '{image_name}' not found locally, pulling from Docker Hub") # Use the low-level API to stream the pull response low_level_client = docker.APIClient() （剩下的代码部分省略）
#2. shell命令行执行 def execute_shell(command_line):
current_dir = os.getcwd()
if not WORKSPACE_FOLDER in current_dir: # Change dir into workspace if necessary work_dir = os.path.join(os.getcwd(), WORKSPACE_FOLDER) os.chdir(work_dir)
print (f"Executing command '{command_line}' in working directory '{os.getcwd()}'")
result = subprocess.run(command_line, capture_output=True, shell=True) output = f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
# Change back to whatever the prior working dir was
#1.前面的大致是用urllib库进行爬虫，并用BeautifulSoup进行解析 def summarize_text(text, question): """Summarize text using the LLM model""" if not text: return "Error: No text to summarize"
#1. Pinecone 调用向量数据库查询方法（./memory/pinecone.py) def get_relevant(self, data, num_relevant=5): """ Returns all the data in the memory that is relevant to the given data. :param data: The data to compare to. :param num_relevant: The number of relevant data to return. Defaults to 5 """ query_embedding = get_ada_embedding(data) results = self.index.query(query_embedding, top_k=num_relevant, include_metadata=True) sorted_results = sorted(results.matches, key=lambda x: x.score) return [str(item['metadata']["raw_text"]) for item in sorted_results]
#2. Local 使用numpy求向量内积并找出最小k个（./memory/local.py) def get_relevant(self, text: str, k: int) -> List[Any]: """" matrix-vector mult to find score-for-each-row-of-matrix get indices for top-k winning scores return texts for those indices Args: text: str k: int
Returns: List[str] """ embedding = get_ada_embedding(text) scores = np.dot(self.data.embeddings, embedding) top_k_indices = np.argsort(scores)[-k:][::-1] return [self.data.texts[i] for i in top_k_indices]
#3. Redis 使用Redis自带的向量搜索VectorField方法查找（./memory/redis.py) def get_relevant( self, data: str, num_relevant: int = 5 ) -> Optional[List[Any]]: """ Returns all the data in the memory that is relevant to the given data. Args: data: The data to compare to. num_relevant: The number of relevant data to return.
Returns: A list of the most relevant data. """ query_embedding = get_ada_embedding(data) base_query = f"*=>[KNN {num_relevant} @embedding $vector AS vector_score]" query = Query(base_query).return_fields( "data", "vector_score" ).sort_by("vector_score").dialect(2) query_vector = np.array(query_embedding).astype(np.float32).tobytes()
try: results = self.redis.ft(f"{self.cfg.memory_index}").search( query, query_params={"vector": query_vector} ) except Exception as e: print("Error calling Redis search: ", e) return None return [result.data for result in results.docs]
if response.status_code == 200: with mutex_lock: with open("speech.mpeg", "wb") as f: f.write(response.content) playsound("speech.mpeg", True) os.remove("speech.mpeg") return True else: print("Request failed with status code:", response.status_code) print("Response content:", response.content) return False #3. 执行方法 def say_text(text, voice_index=0):
def speak(): if not cfg.elevenlabs_api_key: if cfg.use_mac_os_tts == 'True': macos_tts_speech(text, voice_index) else: gtts_speech(text) else: success = eleven_labs_speech(text, voice_index) if not success: gtts_speech(text)
