svjack/Code_Act_Agent
1
1from typing import List, Optional, Tuple, Dict2History = List[Tuple[str, str]]3Messages = List[Dict[str, str]]4 5import enum6from dataclasses import dataclass7from typing import List, Dict, Any, Optional, Tuple8from collections import defaultdict9 10 11@dataclass(frozen=True)12class Action:13 value: str # LM returned string for now14 use_tool: bool # if use_tool == False -> propose answer15 error: Optional[str] = None16 17def lm_output_to_action(lm_output: str) -> Action:18 propose_solution = bool("<solution>" in lm_output)19 return Action(lm_output, not propose_solution)20 21from typing import Mapping22import re23import signal24from contextlib import contextmanager25from IPython.core.interactiveshell import InteractiveShell26from IPython.utils import io27from typing import Any28 29from abc import ABC, abstractmethod30from typing import Any31 32 33class Tool(ABC):34 """Abstract class for a tool."""35 36 name: str37 signature: str38 description: str39 40 @abstractmethod41 def __call__(self, *args: Any, **kwds: Any) -> str:42 """Execute the tool with the given args and return the output."""43 # execute tool with abitrary args44 pass45 46 def reset(self) -> None:47 """Reset the tool to its initial state."""48 pass49 50 51class PythonREPL(Tool):52 """A tool for running python code in a REPL."""53 54 name = "PythonREPL"55 # This PythonREPL is not used by the environment; It is THE ENVIRONMENT.56 signature = "NOT_USED"57 description = "NOT_USED"58 59 def __init__(60 self,61 user_ns: Mapping[str, Any],62 timeout: int = 30,63 ) -> None:64 super().__init__()65 self.user_ns = user_ns66 self.timeout = timeout67 self.reset()68 69 @contextmanager70 def time_limit(self, seconds):71 def signal_handler(signum, frame):72 raise TimeoutError(f"Timed out after {seconds} seconds.")73 74 signal.signal(signal.SIGALRM, signal_handler)75 signal.alarm(seconds)76 try:77 yield78 finally:79 signal.alarm(0) # Disable the alarm80 81 def reset(self) -> None:82 InteractiveShell.clear_instance()83 self.shell = InteractiveShell.instance(84 # NOTE: shallow copy is needed to avoid85 # shell modifying the original user_ns dict86 user_ns=dict(self.user_ns),87 colors="NoColor",88 )89 90 def __call__(self, query: str) -> str:91 """Use the tool and return observation"""92 with io.capture_output() as captured:93 _ = self.shell.run_cell(query, store_history=True)94 output = captured.stdout95 96 if output == "":97 output = "[Executed Successfully with No Output]"98 99 # replace potentially sensitive filepath100 # e.g., File /mint/mint/tools/python_tool.py:30, in PythonREPL.time_limit.<locals>.signal_handler(signum, frame)101 # with File <filepath>:30, in PythonREPL.time_limit.<locals>.signal_handler(signum, frame)102 # use re103 output = re.sub(104 # r"File (/mint/)mint/tools/python_tool.py:(\d+)",105 r"File (.*)mint/tools/python_tool.py:(\d+)",106 r"File <hidden_filepath>:\1",107 output,108 )109 if len(output) > 2000:110 output = output[:2000] + "...\n[Output Truncated]"111 112 return output113 114class ParseError(Exception):115 pass116 117def parse_action(action: Action) -> Tuple[str, Dict[str, Any]]:118 """Define the parsing logic."""119 lm_output = "\n" + action.value + "\n"120 output = {}121 try:122 if not action.use_tool:123 answer = "\n".join(124 [125 i.strip()126 for i in re.findall(127 r"<solution>(.*?)</solution>", lm_output, re.DOTALL128 )129 ]130 )131 if answer == "":132 raise ParseError("No answer found.")133 output["answer"] = answer134 else:135 env_input = "\n".join(136 [137 i.strip()138 for i in re.findall(139 r"<execute>(.*?)</execute>", lm_output, re.DOTALL140 )141 ]142 )143 if env_input == "":144 raise ParseError("No code found.")145 output["env_input"] = env_input146 except Exception as e:147 raise ParseError(e)148 return output149 150python_repl = PythonREPL(151 user_ns={},152 )153 154import gradio as gr155import llama_cpp156import llama_cpp.llama_tokenizer157import torch158 159if torch.cuda.is_available():160 CodeActAgent_llm = llama_cpp.Llama.from_pretrained(161 repo_id="xingyaoww/CodeActAgent-Mistral-7b-v0.1.q8_0.gguf",162 filename="*q8_0.gguf",163 verbose=False,164 n_gpu_layers = -1,165 n_ctx = 3060166 )167else:168 CodeActAgent_llm = llama_cpp.Llama.from_pretrained(169 repo_id="xingyaoww/CodeActAgent-Mistral-7b-v0.1.q8_0.gguf",170 filename="*q8_0.gguf",171 verbose=False,172 #n_gpu_layers = -1,173 n_ctx = 3060174 )175 176system_prompt = '''177You are a helpful assistant assigned with the task of problem-solving. To achieve this, you will be using an interactive coding environment equipped with a variety of tool functions to assist you throughout the process.178 179At each turn, you should first provide your step-by-step thinking for solving the task. Your thought process should be enclosed using "<thought>" tag, for example: <thought> I need to print "Hello World!" </thought>.180 181After that, you have two options:182 1831) Interact with a Python programming environment and receive the corresponding output. Your code should be enclosed using "<execute>" tag, for example: <execute> print("Hello World!") </execute>.1842) Directly provide a solution that adheres to the required format for the given task. Your solution should be enclosed using "<solution>" tag, for example: The answer is <solution> A </solution>.185 186You have {max_total_steps} chances to interact with the environment or propose a solution. You can only propose a solution {max_propose_solution} times.187'''.format(188 **{189 "max_total_steps": 5,190 "max_propose_solution": 2,191 }192)193 194 195def exe_to_md(str_):196 req = str_.replace("<execute>" ,"```python").replace("</execute>" ,"```").replace("<solution>" ,"```python").replace("</solution>" ,"```")197 if "<thought>" in req and "def " in req:198 req = req.replace("<thought>" ,"```python").replace("</thought>" ,"```")199 return req200 201def md_to_exe(str_):202 return str_.replace("```python", "<execute>").replace("```", "</execute>")203 204def clear_session() -> History:205 return '', []206 207def modify_system_session(system: str) -> str:208 if system is None or len(system) == 0:209 system = default_system210 return system, system, []211 212def history_to_messages(history: History, system: str) -> Messages:213 messages = [{'role': "system", 'content': system}]214 for h in history:215 messages.append({'role': "user", 'content': h[0]})216 if h[1] != "😊":217 messages.append({'role': "assistant", 'content':218 md_to_exe(h[1])219 })220 return messages221 222def messages_to_history(messages: Messages) -> Tuple[str, History]:223 assert messages[0]['role'] == "system"224 system = messages[0]['content']225 history = []226 import numpy as np227 import pandas as pd228 from copy import deepcopy229 messages = deepcopy(messages)230 if messages[-1]["role"] == "user":231 messages += [{"role": "assistant", "content": "😊"}]232 233 messages_ = []234 for ele in messages[1:]:235 if not messages_:236 messages_.append(ele)237 else:238 if messages_[-1]["role"] == ele["role"]:239 continue240 else:241 messages_.append(ele)242 243 history = pd.DataFrame(np.asarray(messages_).reshape([-1, 2]).tolist()).applymap(244 lambda x: x["content"]245 ).applymap(246 exe_to_md247 ).values.tolist()248 return system, history249 250def model_chat(query: Optional[str], history: Optional[History], system: str251) -> Tuple[str, str, History]:252 if query is None:253 query = ''254 if history is None:255 history = []256 messages = history_to_messages(history, system)257 if query:258 messages.append({'role': "user", 'content': query})259 260 response = CodeActAgent_llm.create_chat_completion(261 messages=messages,262 stream=True,263 top_p = 0.9,264 temperature = 0.01265 )266 267 from IPython.display import clear_output268 lm_output = ""269 for chunk in response:270 delta = chunk["choices"][0]["delta"]271 if "content" not in delta:272 continue273 lm_output += delta["content"]274 275 lm_output = lm_output.replace("<solution>", "<execute>").replace("</solution>", "</execute>")276 277 if "<execute>" in lm_output:278 action_out = lm_output_to_action(lm_output)279 parsed = parse_action(action_out)280 env_input = parsed["env_input"]281 obs = python_repl(env_input).strip()282 obs = '''283 Observation:284 {}285 '''.format(obs).strip()286 287 system, history = messages_to_history(messages + [288 {'role': "assistant",289 'content': exe_to_md(lm_output)},290 {291 'role': "user",292 "content": obs293 }294 ])295 elif "<thought>" in lm_output:296 system, history = messages_to_history(messages + [297 {'role': "assistant",298 'content': exe_to_md(lm_output)},299 ])300 else:301 system, history = messages_to_history(messages + [302 {'role': "assistant",303 'content': exe_to_md(lm_output)},304 ])305 return "", history, system306 307 308with gr.Blocks() as demo:309 gr.Markdown("""<center><font size=8>CodeActAgent Mistral 7B Bot 🤖</center>""")310 311 with gr.Row():312 with gr.Column(scale=3):313 system_input = gr.Textbox(value=system_prompt, lines=1, label='System', visible = False)314 with gr.Column(scale=1):315 modify_system = gr.Button("🛠️ Set system prompt and clear history", scale=2, visible = False)316 system_state = gr.Textbox(value=system_prompt, visible=False)317 chatbot = gr.Chatbot(label='CodeActAgent-Mistral-7b-v0.1')318 textbox = gr.Textbox(lines=2, label='Input')319 320 with gr.Row():321 clear_history = gr.Button("🧹 Clear History")322 sumbit = gr.Button("🚀 Send")323 324 sumbit.click(model_chat,325 inputs=[textbox, chatbot, system_state],326 outputs=[textbox, chatbot, system_input],327 concurrency_limit = 100)328 clear_history.click(fn=clear_session,329 inputs=[],330 outputs=[textbox, chatbot])331 modify_system.click(fn=modify_system_session,332 inputs=[system_input],333 outputs=[system_state, system_input, chatbot])334 335 gr.Examples(336 [337 "teach me how to use numpy.",338 "Give me a python function give the divide of number it self 10 times.",339 '''340 Plot box plot with pandas and save it to local.341 '''.strip(),342 343 '''344 Write a python code about, download image to local from url, the format as :345 url = f'https://image.pollinations.ai/prompt/{prompt}'346 where prompt as the input of download function.347 '''.strip(),348 "Use this function download a image of bee.",349 350 '''351 Draw a picture teach me what linear regression is.352 '''.strip(),353 "Use more points and draw the image with the line fitted.",354 355 '''356 Write a piece of Python code to simulate the financial transaction process and draw a financial images chart by lineplot of Poisson process.357 '''.strip(),358 #"Add monotonic increasing trend on it.",359 "Add a Trigonometric function loop on it.",360 ],361 inputs = textbox,362 label = "Task Prompt: \n(Used to give the function or task defination on the head)",363 )364 365 gr.Examples(366 [367 '''368 Give me the function defination. 💡369 '''.strip(),370 371 '''372 Correct it. ☹️❌373 '''.strip(),374 375 '''376 Save the output as image 🖼️ to local. ⏬377 '''.strip(),378 379 '''380 Good Job 😊381 '''.strip(),382 ],383 inputs = textbox,384 label = "Action Prompt: \n(Used to specify downstream actions taken by LLM, such as modifying errors, saving running results locally, saying you did a good job, etc.)",385 )386 387demo.queue(api_open=False)388demo.launch(max_threads=30, share = False)