| import google.generativeai as genai |
| import importlib.metadata |
| import requests |
| from packaging import version as pkg_version |
| import io |
| import sys |
| import os |
|
|
| |
| sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') |
|
|
| |
|
|
| API_KEY = os.getenv("GOOGLE_GENAI_API_KEY", "") |
| genai.configure(api_key=API_KEY) |
|
|
| |
| def log_output(text, filename="log.txt"): |
| with open(filename, "a", encoding="utf-8") as f: |
| f.write(text + "\n") |
|
|
| |
| try: |
| lib_version = importlib.metadata.version("google-generativeai") |
| print(f"讙专住转 google-generativeai: {lib_version}") |
| if pkg_version.parse(lib_version) < pkg_version.parse("0.7.2"): |
| print("砖讚专讙 讗转 讛住驻专讬讬讛: pip install --upgrade google-generativeai") |
| exit(1) |
| except importlib.metadata.PackageNotFoundError: |
| print("讛住驻专讬讬讛 google-generativeai 诇讗 诪讜转拽谞转.") |
| exit(1) |
|
|
| |
| try: |
| model = genai.GenerativeModel(model_name="gemini-2.5-flash") |
| print("诪讜讚诇 谞讟注谉 讘讛爪诇讞讛") |
| except Exception as e: |
| print(f"砖讙讬讗讛 讘讟注讬谞转 讛诪讜讚诇: {e}") |
| exit(1) |
|
|
| |
| try: |
| chat = model.start_chat() |
| print("爪'讗讟 讛转讞讬诇") |
| except Exception as e: |
| print(f"砖讙讬讗讛 讘讛转讞诇转 爪'讗讟: {e}") |
| exit(1) |
|
|
| |
| chat.send_message(""" |
| 讗转讛 AI Agent 讞讻诐 砖诪讘爪注 诪砖讬诪讜转 砖诇讘 讗讞专 砖诇讘. |
| 讻讗砖专 讬砖 爪讜专讱, 讗转讛 讬讻讜诇 诇讛砖转诪砖 讘住讬诪讜谉 @@PYTHON@@ 诇讛专爪转 拽讜讚 驻讬讬转讜谉 |
| 讗讜 讘住讬诪讜谉 @@URL@@ 讻讚讬 诇讛讘讬讗 转讜讻谉 诪讚驻讬 讗讬谞讟专谞讟. |
| 谞讬转谉 诇讛砖转诪砖 讘-@@THOUGHTS@@ 讻讚讬 诇转讻谞谉 诪讞砖讘讜转 砖诇讘 讗讞专讬 砖诇讘. |
| 讛诪砖讱 注讚 砖讬砖 转砖讜讘讛 住讜驻讬转 砖诪讜爪讙转 诇诪砖转诪砖. |
| """) |
|
|
| |
| def execute_code(code): |
| try: |
| exec_globals = {} |
| buffer = io.StringIO() |
| old_stdout = sys.stdout |
| sys.stdout = buffer |
|
|
| result = None |
| try: |
| |
| result = eval(code, exec_globals) |
| except: |
| |
| exec(code, exec_globals) |
| result = exec_globals.get("result", None) |
|
|
| output = buffer.getvalue().strip() |
| sys.stdout = old_stdout |
|
|
| |
| if result is not None: |
| if output: |
| final = f"{output}\n{result}" |
| else: |
| final = str(result) |
| else: |
| final = output or "No result" |
|
|
| |
| log_output(f"========== PYTHON CODE ==========\n{code}\n========== OUTPUT ==========\n{final}\n", "LOG.TXT") |
| return final |
|
|
| except Exception as e: |
| err = f"砖讙讬讗转 拽讜讚: {str(e)}" |
| log_output(f"========== PYTHON CODE ==========\n{code}\n========== ERROR ==========\n{err}\n", "LOG.TXT") |
| return err |
|
|
| def get_web_page(url): |
| """拽讘诇转 转讜讻谉 诪讚祝 讗讬谞讟专谞讟""" |
| try: |
| headers = {"User-Agent": "Mozilla/5.0"} |
| r = requests.get(url, headers=headers, timeout=5) |
| r.raise_for_status() |
| html_snippet = r.text[:5000] |
| log_output(f"@@URL@@ {url}\n{html_snippet}") |
| return html_snippet |
| except Exception as e: |
| err = f"砖讙讬讗讛 讘拽讘诇转 讛讚祝: {str(e)}" |
| log_output(f"@@URL@@ {url}\n{err}") |
| return err |
|
|
| LOG_FILE = "log.txt" |
| THOUGHTS_FILE = "thoughts_log.txt" |
|
|
| def handle_special_requests(text): |
| import re |
|
|
| |
| url_pattern = re.compile(r"@@URL@@\s*(.*?)\s*(?:@@|$)", re.DOTALL) |
| for match in url_pattern.finditer(text): |
| url = match.group(1).strip() |
| result = get_web_page(url) |
| text = text.replace(match.group(0), result) |
| log_output(f"\n========== URL REQUEST ==========\n{url}\n========== OUTPUT ==========\n{result}\n", LOG_FILE) |
|
|
| |
| python_pattern = re.compile(r"@@PYTHON@@\s*(.*?)\s*(?:@@|$)", re.DOTALL) |
| for match in python_pattern.finditer(text): |
| code = match.group(1).strip() |
| result = execute_code(code) |
| text = text.replace(match.group(0), result) |
| log_output(f"\n========== PYTHON CODE ==========\n{code}\n========== OUTPUT ==========\n{result}\n", LOG_FILE) |
|
|
| |
| thoughts_pattern = re.compile(r"@@THOUGHTS@@\s*(.*?)\s*(?:@@|$)", re.DOTALL) |
| for match in thoughts_pattern.finditer(text): |
| thoughts = match.group(1).strip() |
| |
| log_output(f"\n========== THOUGHTS ==========\n{thoughts}\n", THOUGHTS_FILE) |
| |
| text = text.replace(match.group(0), f"(诪讞砖讘讜转 AI 诪转讻谞谞讜转 砖诇讘 讗讞专讬 砖诇讘)\n{thoughts}") |
|
|
| return text |
|
|
| def process_message(user_input): |
| """砖诇讬讞转 讘拽砖讛 诇诪讜讚诇 讜讟讬驻讜诇 讘讻诇 住讬诪讜谉""" |
| response = chat.send_message(user_input) |
| output = getattr(response, "text", "") |
| while True: |
| new_output = handle_special_requests(output) |
| if new_output == output: |
| break |
| response = chat.send_message(new_output) |
| output = getattr(response, "text", "") |
| return output |
|
|
| def main(): |
| print("AI Agent 诪讜讻谉! 讛拽诇讚 诪砖讬诪讛 (讗讜 'exit' 诇住讬讜诐).") |
| while True: |
| try: |
| user_input = input("\n诪砖讬诪讛: ") |
| if user_input.lower() == "exit": |
| print("讬讜爪讗...") |
| break |
| answer = process_message(user_input) |
| print(f"\nAgent:\n{answer}\n") |
| except KeyboardInterrupt: |
| print("\n讬讜爪讗...") |
| break |
| except Exception as e: |
| print(f"砖讙讬讗讛 讻诇诇讬转: {str(e)}\n") |
|
|
| if __name__ == "__main__": |
| main() |
|
|
