Hanah

# どの役割でもインポート可能で、初期化し、開始メッセージで実行し、完了！
from metagpt.roles.product_manager import ProductManager
prompt = f"""
# 役割：ソフトウェア開発チーム

## 背景 :

私はソフトウェア開発チームです。
今、html、js、vue3、element-plusを使って問題解決プログラムを開発します。
問題解決は、人々が問題に関連する知識をより深く理解するのに役立ちます。

## プロフィール:
- 著者: 黎伟
- バージョン: 0.1
- 言語: 中国語
- 説明: 私はソフトウェア開発チームです。

## 目標:
- html、js、vue3、element-plusを使って問題解決プログラムの開発要件文書を作成します。

## 制約:
1. 最終的に納品されるプログラムはhtml単一ファイルであり、他のファイルは不要です。
2. 問題の形式は、少なくとも2つの判断問題、2つの選択問題、2つの空欄問題を含む必要があります。
3. 問題の内容は人工知能のエージェントの基本理論に関連している必要があります。
4. 問題解決プログラムは、少なくとも10のサンプル問題を提供する必要があります。
5. 問題はリスト形式でhtmlファイルのscript部分に書き込む必要があります。
6. vue3、element-plusはcdn形式でhtmlのheader部分にインポートします。

## スキル:
1. 強力なjs言語開発能力を持つ
2. vue3、element-plusの使用に精通している
3. 人工知能のエージェントの基本理論を良く理解している
4. タイポグラフィの美学を持ち、番号、インデント、区切り線、改行などを利用して情報のレイアウトを美化することができる


上記の要件に基づいて、問題解決プログラムの開発要件文書を完成させてください。
"""
async def main():
    role = ProductManager()
    result = await role.run(prompt)
    
await main()

from metagpt.actions import Action

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = SimpleWriteCode.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        pattern = r"```python(.*)```"
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text

import re
import os
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger

class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode])

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo  # todo will be SimpleWriteCode()

        msg = self.get_memories(k=1)[0]  # find the most recent messages
        code_text = await todo.run(msg.content)
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

async def main():
    msg = "write a function that calculates the sum of a list"
    role = SimpleCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
    return result
    
        
rtn = await main()

2025-01-26 18:14:18.317 | INFO     | __main__:main:4 - write a function that calculates the sum of a list
2025-01-26 18:14:18.319 | INFO     | __main__:_act:16 - Alice(SimpleCoder): to do SimpleWriteCode(SimpleWriteCode)
```python
def sum_list(numbers):
    return sum(numbers)

# テストケース
test_case_1 = [1, 2, 3, 4, 5]
test_case_2 = [10, 20, 30, 40, 50]

# テストケース1
print(sum_list(test_case_1))  # 期待される出力: 15

# テストケース2
print(sum_list(test_case_2))  # 期待される出力: 150

2025-01-26 18:14:25.372 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 18:14:25.374 | INFO     | __main__:main:6 - SimpleCoder: 
def sum_list(numbers):
    return sum(numbers)

# テストケース
test_case_1 = [1, 2, 3, 4, 5]
test_case_2 = [10, 20, 30, 40, 50]

# テストケース1
print(sum_list(test_case_1))  # 期待される出力: 15

# テストケース2
print(sum_list(test_case_2))  # 期待される出力: 150

# SimpleWriteCode このクラスは前のセクションとまったく同じです

from metagpt.actions import Action

class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction} and provide two runnnable test cases.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = SimpleWriteCode.parse_code(rsp)

        return code_text

    @staticmethod
    def parse_code(rsp):
        pattern = r"```python(.*)```"
        match = re.search(pattern, rsp, re.DOTALL)
        code_text = match.group(1) if match else rsp
        return code_text

# 本節ではSimpleRunCodeというクラスを追加します
class SimpleRunCode(Action):
    name: str = "SimpleRunCode"

    async def run(self, code_text: str):
        result = subprocess.run(["python", "-c", code_text], capture_output=True, text=True)
        code_result = result.stdout
        logger.info(f"{code_result=}")
        return code_result

import re
import os
import subprocess
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.logs import logger
class RunnableCoder(Role):
    name: str = "Alice"
    profile: str = "RunnableCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteCode, SimpleRunCode])
        self._set_react_mode(react_mode="by_order")

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        # By choosing the Action by order under the hood
        # todo will be first SimpleWriteCode() then SimpleRunCode()
        todo = self.rc.todo

        msg = self.get_memories(k=1)[0]  # find the most k recent messages
        result = await todo.run(msg.content)

        msg = Message(content=result, role=self.profile, cause_by=type(todo))
        self.rc.memory.add(msg)
        return msg

async def main():
    msg = "write a function that calculates the sum of a list"
    role = RunnableCoder()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
    return result

rtn = await main()
print(rtn)

def calculate_sum(numbers):
    return sum(numbers)

# テストケース
def test_case_1():
    assert calculate_sum([1, 2, 3, 4, 5]) == 15

def test_case_2():
    assert calculate_sum([10, -5, 0, 7, 3]) == 15

from metagpt.actions import Action
from typing import Dict, Union
import ast

def extract_struct(text: str, data_type: Union[type(list), type(dict)]) -> Union[list, dict]:
    """指定されたタイプの構造（辞書またはリスト）を与えられたテキストから抽出し、解析します。
    テキストにはリストまたは辞書のみが含まれ、ネストされた構造を持つ場合があります。

    Args:
        text: 構造を含むテキスト（辞書またはリスト）。
        data_type: 抽出するデータタイプ、"list"または"dict"。

    Returns:
        - 抽出と解析が成功した場合、対応するデータ構造（リストまたは辞書）を返します。
        - 抽出が失敗した場合や解析中にエラーが発生した場合は、例外をスローします。
    例:
        >>> text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
        >>> result_list = OutputParser.extract_struct(text, "list")
        >>> print(result_list)
        >>> # 出力: [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}]

        >>> text = 'xxx {"x": 1, "y": {"a": 2, "b": {"c": 3}}} xxx'
        >>> result_dict = OutputParser.extract_struct(text, "dict")
        >>> print(result_dict)
        >>> # 出力: {"x": 1, "y": {"a": 2, "b": {"c": 3}}}
    """
    # 最初の "[" または "{" と最後の "]" または "}" を見つける
    start_index = text.find("[" if data_type is list else "{")
    end_index = text.rfind("]" if data_type is list else "}")

    if start_index != -1 and end_index != -1:
        # 構造部分を抽出する
        structure_text = text[start_index : end_index + 1]

        try:
            # テキストをPythonデータ型に変換しようとする
            result = ast.literal_eval(structure_text)

            # 結果が指定されたデータタイプと一致することを確認する
            if isinstance(result, list) or isinstance(result, dict):
                return result

            raise ValueError(f"The extracted structure is not a {data_type}.")

        except (ValueError, SyntaxError) as e:
            raise Exception(f"Error while extracting and parsing the {data_type}: {e}")
    else:
        logger.error(f"No {data_type} found in the text.")
        return [] if data_type is list else {}

class WriteDirectory(Action):
    """チュートリアルディレクトリを書くためのアクションクラス。

    Args:
        name: アクションの名前。
        language: 出力する言語、デフォルトは"Chinese"。
        
        チュートリアルディレクトリを書くためのアクションクラス。
        パラメータ：
        name：アクションの名前。
        language：出力の言語、デフォルトは"Chinese"。
    """

    name: str = "WriteDirectory"
    language: str = "Chinese"

    async def run(self, topic: str, *args, **kwargs) -> Dict:
        """トピックに従ってチュートリアルディレクトリを生成するアクションを実行します。

        Args:
            topic: チュートリアルのトピック。

        Returns:
            チュートリアルディレクトリ情報、{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}を含む。
        トピックに従ってチュートリアルディレクトリを生成するアクションを実行します。
            パラメータ：
            topic：チュートリアルのトピック。
            戻り値：
            チュートリアルディレクトリ情報、{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}を含む。
        """
        COMMON_PROMPT = """
        あなたは今、インターネットの分野で経験豊富な技術専門家です。 
        私たちはあなたに「{topic}」というトピックの技術チュートリアルを書いてもらいたいです。
        """

        DIRECTORY_PROMPT = COMMON_PROMPT + """
        このチュートリアルの具体的な目次を提供してください。以下の要件に厳密に従ってください：
        1. 出力は指定された言語、{language}に厳密に従う必要があります。
        2. 回答は厳密に辞書形式で、{{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}のようにしてください。
        3. 目次はできるだけ具体的かつ十分であるべきで、一次および二次の目次を含むべきです。二次の目次は配列にします。
        4. 余分なスペースや改行がないようにしてください。
        5. 各目次のタイトルには実用的な意味があるべきです。
        """
        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
        resp = await self._aask(prompt=prompt)
        return extract_struct(resp, dict)

text = 'xxx [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}] xxx'
result_list = extract_struct(text, list)
print(result_list)
# 出力: [1, 2, ["a", "b", [3, 4]], {"x": 5, "y": [6, 7]}]

text = 'xxx {"x": 1, "y": {"a": 2, "b": {"c": 3}}} xxx'
result_dict = extract_struct(text, dict)
print(result_dict)
# 出力: {"x": 1, "y": {"a": 2, "b": {"c": 3}}}

from metagpt.const import TUTORIAL_PATH
TUTORIAL_PATH
# WindowsPath('d:/Code/agent_from_scratch/data/tutorial_docx')

from datetime import datetime
from typing import Dict
import asyncio
from metagpt.actions.write_tutorial import WriteDirectory, WriteContent
from metagpt.const import TUTORIAL_PATH
from metagpt.logs import logger
from metagpt.roles.role import Role, RoleReactMode
from metagpt.schema import Message
from metagpt.utils.file import File

from typing import Dict

from metagpt.actions import Action

class WriteDirectory(Action):
    """チュートリアルディレクトリを書くためのアクションクラス。

    Args:
        name: アクションの名前。
        language: 出力する言語、デフォルトは"Chinese"。
    """

    name: str = "WriteDirectory"
    language: str = "Chinese"

    async def run(self, topic: str, *args, **kwargs) -> Dict:
        """トピックに従ってチュートリアルディレクトリを生成するアクションを実行します。

        Args:
            topic: チュートリアルのトピック。

        Returns:
            チュートリアルディレクトリ情報、{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}を含む。
        """
        COMMON_PROMPT = """
        あなたは今、インターネットの分野で経験豊富な技術専門家です。 
        私たちはあなたに「{topic}」というトピックの技術チュートリアルを書いてもらいたいです。
        """

        DIRECTORY_PROMPT = COMMON_PROMPT + """
        このチュートリアルの具体的な目次を提供してください。以下の要件に厳密に従ってください：
        1. 出力は指定された言語、{language}に厳密に従う必要があります。
        2. 回答は厳密に辞書形式で、{{"title": "xxx", "directory": [{{"dir 1": ["sub dir 1", "sub dir 2"]}}, {{"dir 2": ["sub dir 3", "sub dir 4"]}}]}}のようにしてください。
        3. 目次はできるだけ具体的かつ十分であるべきで、一次および二次の目次を含むべきです。二次の目次は配列にします。
        4. 余分なスペースや改行がないようにしてください。
        5. 各目次のタイトルには実用的な意味があるべきです。
        """
        prompt = DIRECTORY_PROMPT.format(topic=topic, language=self.language)
        resp = await self._aask(prompt=prompt)
        return extract_struct(resp, dict)

class WriteContent(Action):
    """チュートリアルコンテンツを書くためのアクションクラス。

    Args:
        name: アクションの名前。
        directory: 書くコンテンツ。
        language: 出力する言語、デフォルトは"Chinese"。
    """

    name: str = "WriteContent"
    directory: dict = dict()
    language: str = "Chinese"

    async def run(self, topic: str, *args, **kwargs) -> str:
        """ディレクトリとトピックに従って文書コンテンツを書くアクションを実行します。

        Args:
            topic: チュートリアルのトピック。

        Returns:
            書かれたチュートリアルコンテンツ。
        """
        COMMON_PROMPT = """
        あなたは今、インターネットの分野で経験豊富な技術専門家です。 
        私たちはあなたに「{topic}」というトピックの技術チュートリアルを書いてもらいたいです。
        """
        CONTENT_PROMPT = COMMON_PROMPT + """
        今、私はあなたにトピックのモジュールディレクトリタイトルを与えます。 
        このタイトルの詳細な原則コンテンツを詳細に出力してください。 
        コード例がある場合は、標準のコード仕様に従って提供してください。 
        コード例がない場合は、必要ありません。

        トピックのモジュールディレクトリタイトルは次のとおりです：
        {directory}

        出力を次の要件に厳密に制限してください：
        1. レイアウトのためにMarkdown構文形式に従ってください。
        2. コード例がある場合は、標準の構文仕様に従い、ドキュメント注釈を付け、コードブロックで表示してください。
        3. 出力は指定された言語、{language}に厳密に従う必要があります。
        4. 冗長な出力は避けてください。結論の発言を含めないでください。
        5. トピック"{topic}"を出力しないという厳格な要件。
        """
        prompt = CONTENT_PROMPT.format(
            topic=topic, language=self.language, directory=self.directory)
        return await self._aask(prompt=prompt)

class TutorialAssistant(Role):
    """チュートリアルアシスタント、1文を入力してマークアップ形式のチュートリアル文書を生成します。

    Args:
        name: 役割の名前。
        profile: 役割のプロフィール説明。
        goal: 役割の目標。
        constraints: 役割の制約または要件。
        language: チュートリアル文書が生成される言語。
    """

    name: str = "Stitch"
    profile: str = "Tutorial Assistant"
    goal: str = "Generate tutorial documents"
    constraints: str = "Strictly follow Markdown's syntax, with neat and standardized layout"
    language: str = "Chinese"

    topic: str = ""
    main_title: str = ""
    total_content: str = ""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WriteDirectory(language=self.language)])
        self._set_react_mode(react_mode=RoleReactMode.REACT.value)

    async def _think(self) -> None:
        """役割が取るべき次のアクションを決定します。"""
        logger.info(self.rc.state)
        logger.info(self,)
        if self.rc.todo is None:
            self._set_state(0)
            return

        if self.rc.state + 1 < len(self.states):
            self._set_state(self.rc.state + 1)
        else:
            self.rc.todo = None

    async def _handle_directory(self, titles: Dict) -> Message:
        """チュートリアル文書のディレクトリを処理します。

        Args:
            titles: タイトルとディレクトリ構造を含む辞書、
                    たとえば{"title": "xxx", "directory": [{"dir 1": ["sub dir 1", "sub dir 2"]}]}

        Returns:
            ディレクトリに関する情報を含むメッセージ。
        """
        self.main_title = titles.get("title")
        directory = f"{self.main_title}\n"
        self.total_content += f"# {self.main_title}"
        actions = list()
        for first_dir in titles.get("directory"):
            actions.append(WriteContent(
                language=self.language, directory=first_dir))
            key = list(first_dir.keys())[0]
            directory += f"- {key}\n"
            for second_dir in first_dir[key]:
                directory += f"  - {second_dir}\n"
        self.set_actions(actions)
        self.rc.todo = None
        return Message(content=directory)

    async def _act(self) -> Message:
        """役割によって決定されたアクションを実行します。

        Returns:
            アクションの結果を含むメッセージ。
        """
        todo = self.rc.todo
        if type(todo) is WriteDirectory:
            msg = self.rc.memory.get(k=1)[0]
            self.topic = msg.content
            resp = await todo.run(topic=self.topic)
            logger.info(resp)
            return await self._handle_directory(resp)
        resp = await todo.run(topic=self.topic)
        logger.info(resp)
        if self.total_content != "":
            self.total_content += "\n\n\n"
        self.total_content += resp
        return Message(content=resp, role=self.profile)

    async def _react(self) -> Message:
        """アシスタントの思考とアクションを実行します。

        Returns:
            アシスタントのアクションの最終結果を含むメッセージ。
        """
        while True:
            await self._think()
            if self.rc.todo is None:
                break
            msg = await self._act()
        root_path = TUTORIAL_PATH / datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        await File.write(root_path, f"{self.main_title}.md", self.total_content.encode('utf-8'))
        return msg

async def main():
    msg = "AI Agent開発教程"
    role = TutorialAssistant()
    logger.info(msg)
    result = await role.run(msg)
    logger.info(result)
    return role.total_content

rtn = await main()

from bs4 import BeautifulSoup

with open("github-trending-raw.html") as f:
    html = f.read()

soup = BeautifulSoup(html, "html.parser")
for i in soup.find_all(True):
    for name in list(i.attrs):
        if i[name] and name not in ["class"]:
            del i[name]

for i in soup.find_all(["svg", "img", "video", "audio"]):
    i.decompose()

with open("github-trending-slim.html", "w") as f:
    f.write(str(soup))

import aiohttp
import asyncio
from bs4 import BeautifulSoup

def fetch_html(url):
    with open(url, encoding="utf-8") as f:
        html = f.read()
    return html

async def parse_github_trending(html):
    soup = BeautifulSoup(html, 'html.parser')

    repositories = []

    for article in soup.select('article.Box-row'):
        repo_info = {}
        
        repo_info['name'] = article.select_one('h2 a').text.strip()
        repo_info['url'] = article.select_one('h2 a')['href'].strip()

        # Description
        description_element = article.select_one('p')
        repo_info['description'] = description_element.text.strip() if description_element else None

        # Language
        language_element = article.select_one('span[itemprop="programmingLanguage"]')
        repo_info['language'] = language_element.text.strip() if language_element else None

        # Stars and Forks
        stars_element = article.select('a.Link--muted')[0]
        forks_element = article.select('a.Link--muted')[1]
        repo_info['stars'] = stars_element.text.strip()
        repo_info['forks'] = forks_element.text.strip()

        # Today's Stars
        today_stars_element = article.select_one('span.d-inline-block.float-sm-right')
        repo_info['today_stars'] = today_stars_element.text.strip() if today_stars_element else None

        repositories.append(repo_info)

    return repositories

async def main():
    url = 'github-trending-raw.html'
    html = fetch_html(url)
    repositories = await parse_github_trending(html)

    for repo in repositories:
        print(f"Name: {repo['name']}")
        print(f"URL: https://github.com{repo['url']}")
        print(f"Description: {repo['description']}")
        print(f"Language: {repo['language']}")
        print(f"Stars: {repo['stars']}")
        print(f"Forks: {repo['forks']}")
        print(f"Today's Stars: {repo['today_stars']}")
        print()
        
await main()

import os
from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Optional
from aiocron import crontab
from pydantic import BaseModel, Field
from pytz import BaseTzInfo

from metagpt.actions.action import Action
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message

# fix SubscriptionRunner not fully defined
from metagpt.environment import Environment as _  # noqa: F401

class SubscriptionRunner(BaseModel):
    """A simple wrapper to manage subscription tasks for different roles using asyncio.
    Example:
        >>> import asyncio
        >>> from metagpt.subscription import SubscriptionRunner
        >>> from metagpt.roles import Searcher
        >>> from metagpt.schema import Message
        >>> async def trigger():
        ...     while True:
        ...         yield Message("the latest news about OpenAI")
        ...         await asyncio.sleep(3600 * 24)
        >>> async def callback(msg: Message):
        ...     print(msg.content)
        >>> async def main():
        ...     pb = SubscriptionRunner()
        ...     await pb.subscribe(Searcher(), trigger(), callback)
        ...     await pb.run()
        >>> asyncio.run(main())
    """

    tasks: Dict[Role, asyncio.Task] = Field(default_factory=dict)

    class Config:
        arbitrary_types_allowed = True

    async def subscribe(
        self,
        role: Role,
        trigger: AsyncGenerator[Message, None],
        callback: Callable[
            [
                Message,
            ],
            Awaitable[None],
        ],
    ):
        """Subscribes a role to a trigger and sets up a callback to be called with the role's response.
        Args:
            role: The role to subscribe.
            trigger: An asynchronous generator that yields Messages to be processed by the role.
            callback: An asynchronous function to be called with the response from the role.
        """
        loop = asyncio.get_running_loop()

        async def _start_role():
            async for msg in trigger:
                resp = await role.run(msg)
                await callback(resp)

        self.tasks[role] = loop.create_task(_start_role(), name=f"Subscription-{role}")

    async def unsubscribe(self, role: Role):
        """Unsubscribes a role from its trigger and cancels the associated task.
        Args:
            role: The role to unsubscribe.
        """
        task = self.tasks.pop(role)
        task.cancel()

    async def run(self, raise_exception: bool = True):
        """Runs all subscribed tasks and handles their completion or exception.
        Args:
            raise_exception: _description_. Defaults to True.
        Raises:
            task.exception: _description_
        """
        i=0
        while True:
            i+=1
            for role, task in self.tasks.items():
                i=0
                if task.done():
                    if task.exception():
                        if raise_exception:
                            raise task.exception()
                        logger.opt(exception=task.exception()).error(
                            f"Task {task.get_name()} run error"
                        )
                    else:
                        logger.warning(
                            f"Task {task.get_name()} has completed. "
                            "If this is unexpected behavior, please check the trigger function."
                        )
                    self.tasks.pop(role)
                    break
            else:
                await asyncio.sleep(1)
            if i>0:
                break

TRENDING_ANALYSIS_PROMPT = """# Requirements
You are a GitHub Trending Analyst, aiming to provide users with insightful and personalized recommendations based on the latest
GitHub Trends. Based on the context, fill in the following missing information, generate engaging and informative titles, 
ensuring users discover repositories aligned with their interests.

# The title about Today's GitHub Trending
## Today's Trends: Uncover the Hottest GitHub Projects Today! Explore the trending programming languages and discover key domains capturing developers' attention. From ** to **, witness the top projects like never before.
## The Trends Categories: Dive into Today's GitHub Trending Domains! Explore featured projects in domains such as ** and **. Get a quick overview of each project, including programming languages, stars, and more.
## Highlights of the List: Spotlight noteworthy projects on GitHub Trending, including new tools, innovative projects, and rapidly gaining popularity, focusing on delivering distinctive and attention-grabbing content for users.
---
# Format Example


# [Title]

## Today's Trends
Today, ** and ** continue to dominate as the most popular programming languages. Key areas of interest include **, ** and **.
The top popular projects are Project1 and Project2.

## The Trends Categories
1. Generative AI
    - [Project1](https://github/xx/project1): [detail of the project, such as star total and today, language, ...]
    - [Project2](https://github/xx/project2): ...
...

## Highlights of the List
1. [Project1](https://github/xx/project1): [provide specific reasons why this project is recommended].
...

---
# Github Trending
{trending}
"""


class CrawlOSSTrending(Action):
    async def run(self, url: str = "https://github.com/trending"):
        async with aiohttp.ClientSession() as client:
            async with client.get(url, proxy=CONFIG.global_proxy) as response:
                response.raise_for_status()
                html = await response.text()

        soup = BeautifulSoup(html, "html.parser")

        repositories = []

        for article in soup.select("article.Box-row"):
            repo_info = {}

            repo_info["name"] = (
                article.select_one("h2 a")
                .text.strip()
                .replace("\n", "")
                .replace(" ", "")
            )
            repo_info["url"] = (
                "https://github.com" + article.select_one("h2 a")["href"].strip()
            )

            # Description
            description_element = article.select_one("p")
            repo_info["description"] = (
                description_element.text.strip() if description_element else None
            )

            # Language
            language_element = article.select_one(
                'span[itemprop="programmingLanguage"]'
            )
            repo_info["language"] = (
                language_element.text.strip() if language_element else None
            )

            # Stars and Forks
            stars_element = article.select("a.Link--muted")[0]
            forks_element = article.select("a.Link--muted")[1]
            repo_info["stars"] = stars_element.text.strip()
            repo_info["forks"] = forks_element.text.strip()

            # Today's Stars
            today_stars_element = article.select_one(
                "span.d-inline-block.float-sm-right"
            )
            repo_info["today_stars"] = (
                today_stars_element.text.strip() if today_stars_element else None
            )

            repositories.append(repo_info)

        return repositories


class AnalysisOSSTrending(Action):
    async def run(self, trending: Any):
        return await self._aask(TRENDING_ANALYSIS_PROMPT.format(trending=trending))

# Role实现
# 对于V0.7 以上的版本，需要把老版本的
# self._init_actions 改为self.set_actions
class OssWatcher(Role):
    def __init__(
        self,
        name="Codey",
        profile="OssWatcher",
        goal="Generate an insightful GitHub Trending analysis report.",
        constraints="Only analyze based on the provided GitHub Trending data.",
    ):
        super().__init__(name=name, profile=profile, goal=goal, constraints=constraints)
        self.set_actions([CrawlOSSTrending, AnalysisOSSTrending])
        self._set_react_mode(react_mode="by_order")

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        # By choosing the Action by order under the hood
        # todo will be first SimpleWriteCode() then SimpleRunCode()
        todo = self.rc.todo

        msg = self.get_memories(k=1)[0]  # find the most k recent messages
        result = await todo.run(msg.content)

        msg = Message(content=str(result), role=self.profile, cause_by=type(todo))
        self.rc.memory.add(msg)
        return msg

async def wxpusher_callback(msg: Message):
    print(msg.content)


async def trigger():
    # 这里设置了只触发五次，也可以用while True 永远执行下去
    for i in range(5):
        yield Message("the latest news about OpenAI")
        await asyncio.sleep(5)
        #  每隔五秒钟执行一次。
        # 也可以设置为每隔3600 * 24 秒执行一次
    

# 运行入口，
async def main():
    callbacks = []
    if not callbacks:
        async def _print(msg: Message):
            print(msg.content)
        callbacks.append(_print)

    # callback
    async def callback(msg):
        await asyncio.gather(*(call(msg) for call in callbacks))

    runner = SubscriptionRunner()
    await runner.subscribe(OssWatcher(), trigger(), callback)
    await runner.run()

await main()

import requests

def get_repo_info(repo_name):
    url = f"https://github.com/search?q=stars:{repo_name}&type=Repositories"
    response = requests.get(url)
    if response.status_code == 200:
        repos = []
        for repo in response.json():
            if "name" in repo and "url" in repo:
                name = repo["name"]
                url = repo["url"]
                stars = int(repo["stargazers_count"])
                forks = int(repo["forks_count"])  # Assuming the number of forks is available
                repos.append({"name": name, "url": url, "stars": stars, "forks": forks})
        return repos
    else:
        print("Failed to fetch data. Status code:", response.status_code)

repos = get_repo_info('language')
for repo in repos:
    if 'language' not in repo or repo['language'] == None:
        del repo['language']
print(repos)

import requests

def get_repo_info(repo_name):
    url = f"https://github.com/search?q=stars:{repo_name}&type=Repositories"
    response = requests.get(url)
    if response.status_code == 200:
        repos = []
        for repo in response.json():
            if "name" in repo and "url" in repo:
                name = repo["name"]
                url = repo["url"]
                stars = int(repo["stargazers_count"])
                forks = int(repo["forks_count"])  # Assuming the number of forks is available
                repos.append({"name": name, "url": url, "stars": stars, "forks": forks})
        return repos
    else:
        print("Failed to fetch data. Status code:", response.status_code)

repos = get_repo_info('language')
for repo in repos:
    if 'language' not in repo or repo['language'] == None:
        del repo['language']
print(repos)

import re
from metagpt.actions import Action, UserRequirement

# コードを書くアクションを構築します

def parse_code(rsp):
    pattern = r"```python(.*)```"
    match = re.search(pattern, rsp, re.DOTALL)
    code_text = match.group(1) if match else rsp
    return code_text


class SimpleWriteCode(Action):
    PROMPT_TEMPLATE: str = """
    Write a python function that can {instruction}.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """
    name: str = "SimpleWriteCode"

    async def run(self, instruction: str):
        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)

        rsp = await self._aask(prompt)

        code_text = parse_code(rsp)

        return code_text

# テストケースを書くアクションを構築します
class SimpleWriteTest(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Write {k} unit tests using pytest for the given function, assuming you have imported it.
    Return ```python your_code_here ```with NO other texts,
    your code:
    """

    name: str = "SimpleWriteTest"

    async def run(self, context: str, k: int = 3):
        prompt = self.PROMPT_TEMPLATE.format(context=context, k=k)

        rsp = await self._aask(prompt)

        code_text = parse_code(rsp)

        return code_text

# コードをレビューするアクションを構築します
class SimpleWriteReview(Action):
    PROMPT_TEMPLATE: str = """
    Context: {context}
    Review the test cases and provide one critical comments:
    """

    name: str = "SimpleWriteReview"

    async def run(self, context: str):
        prompt = self.PROMPT_TEMPLATE.format(context=context)

        rsp = await self._aask(prompt)

        return rsp
                

class SimpleCoder(Role):
    name: str = "Alice"
    profile: str = "SimpleCoder"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._watch([UserRequirement])
        self.set_actions([SimpleWriteCode])

class SimpleTester(Role):
    name: str = "Bob"
    profile: str = "SimpleTester"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteTest])
        self._watch([SimpleWriteCode])
        # self._watch([SimpleWriteCode, SimpleWriteReview])  # feel free to try this too

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})")
        todo = self.rc.todo

        # context = self.get_memories(k=1)[0].content # use the most recent memory as context
        context = self.get_memories()  # use all memories as context

        code_text = await todo.run(context, k=5)  # specify arguments
        msg = Message(content=code_text, role=self.profile, cause_by=type(todo))

        return msg

# 同様のプロセスでSimpleReviewerを定義します：
class SimpleReviewer(Role):
    name: str = "Charlie"
    profile: str = "SimpleReviewer"

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([SimpleWriteReview])
        self._watch([SimpleWriteTest])

import asyncio
from metagpt.team import Team

async def main(
    idea: str = "write a function that calculates the product of a list",
    investment: float = 3.0,
    n_round: int = 5,
):
    logger.info(idea)

    team = Team()
    team.hire(
        [
            SimpleCoder(),
            SimpleTester(),
            SimpleReviewer(),
        ]
    )

    team.invest(investment=investment)
    team.run_project(idea)
    await team.run(n_round=n_round)
await main()

def product_of_list(lst):
    result = 1
    for num in lst:
        result *= num
    return result

def test_product_of_list_empty():
    assert product_of_list([]) == 1

def test_product_of_list_single_element():
    assert product_of_list([5]) == 5

def test_product_of_list_positive_numbers():
    assert product_of_list([1, 2, 3, 4]) == 24

def test_product_of_list_negative_numbers():
    assert product_of_list([-1, -2, -3, -4]) == -24

def test_product_of_list_mixed_numbers():
    assert product_of_list([-1, 2, -3, 4]) ==
2025-01-26 19:38:09.372 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.
2025-01-26 19:38:09.376 | INFO     | metagpt.roles.role:_act:391 - Charlie(SimpleReviewer): to do SimpleWriteReview(SimpleWriteReview)
 24

import asyncio
import platform
from typing import Any

import fire

from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
import asyncio
from metagpt.actions import Action,UserRequirement
from metagpt.schema import Message

class SpeakAloud(Action):
    """アクション：ディベートで大声で話す（口論）"""

    PROMPT_TEMPLATE: str = """
    ## 背景
    あなたは{name}です。あなたは{opponent_name}とのディベートにいます。
    ## ディベートの履歴
    前のラウンド：
    {context}
    ## あなたのターン
    今、あなたのターンです。相手の最新の主張に密接に応じ、自分の立場を述べ、主張を守り、相手の主張を攻撃してください。
    強力で感情的な応答を80語で作成し、{name}の修辞と視点で主張します。
    """

    def __init__(self, name="SpeakAloud", context=None, llm=None):
        super().__init__(name, context, llm)

    async def run(self, context: str, name: str, opponent_name: str):

        prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)

        rsp = await self._aask(prompt)

        return rsp

class Debator(Role):
    def __init__(
        self,
        name: str,
        profile: str,
        opponent_name: str,
        **kwargs,
    ):
        super().__init__(name=name, profile=profile, **kwargs)
        self.set_actions([SpeakAloud])
        self._watch([UserRequirement, SpeakAloud])
        self.name = name
        self.opponent_name = opponent_name

async def _observe(self) -> int:
        await super()._observe()
        # 自分に送られたメッセージを受け入れ、前のラウンドからの自分のメッセージを無視します
        self.rc.news = [msg for msg in self.rc.news if msg.send_to == self.name]
        return len(self.rc.news)

async def _act(self) -> Message:
    logger.info(f"{self._setting}: ready to {self.rc.todo}")
    todo = self.rc.todo # SpeakAloudのインスタンス

    memories = self.get_memories()
    context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)

    rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)

    msg = Message(
        content=rsp,
        role=self.profile,
        cause_by=todo,
        sent_from=self.name,
        send_to=self.opponent_name,
    )
    self.rc.memory.add(msg)
    return msg

async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
    """バイデン-トランプディベートを実行し、彼らの友好的な対話を観察します :) """
    Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
    Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")
    team = Team()
    team.hire([Biden, Trump])
    team.invest(investment)
    team.run_project(idea, send_to="Biden")  # ディベートのトピックをバイデンに送信し、彼が最初に発言するようにします
    await team.run(n_round=n_round)


import typer
app = typer.Typer()

@app.command()
def main(
    idea: str = typer.Argument(..., help="Economic Policy: Discuss strategies and plans related to taxation, employment, fiscal budgeting, and economic growth."),
    investment: float = typer.Option(default=3.0, help="Dollar amount to invest in the AI company."),
    n_round: int = typer.Option(default=5, help="Number of rounds for the simulation."),
):
    """
    :param idea: Debate topic, such as "Topic: The U.S. should commit more in climate change fighting"
                 or "Trump: Climate change is a hoax"
    :param investment: contribute a certain dollar amount to watch the debate
    :param n_round: maximum rounds of the debate
    :return:
    """
    if platform.system() == "Windows":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(debate(idea, investment, n_round))

if __name__ == '__main__':
    app()# run as python debate.py --idea="TOPIC" --investment=3.0 --n_round=5

import asyncio
import platform
from typing import Any

import fire

from metagpt.actions import Action, UserRequirement
from metagpt.logs import logger
from metagpt.roles import Role
from metagpt.schema import Message
from metagpt.team import Team
import asyncio
from metagpt.actions import Action,UserRequirement
from metagpt.schema import Message
class SpeakAloud(Action):
    """アクション：ディベートで大声で話す（口論）"""

    PROMPT_TEMPLATE: str = """
    ## 背景
    あなたは{name}です。あなたは{opponent_name}とのディベートにいます。
    ## ディベートの履歴
    前のラウンド：
    {context}
    ## あなたのターン
    今、あなたのターンです。相手の最新の主張に密接に応じ、自分の立場を述べ、主張を守り、相手の主張を攻撃してください。
    強力で感情的な応答を80語で作成し、{name}の修辞と視点で主張します。
    """

    def __init__(self, name="SpeakAloud", context=None, llm=None):
        super().__init__(name=name, context=context, llm=llm)

    async def run(self, context: str, name: str, opponent_name: str):

        prompt = self.PROMPT_TEMPLATE.format(context=context, name=name, opponent_name=opponent_name)

        rsp = await self._aask(prompt)

        return rsp
    
class Debator(Role):
    def __init__(
        self,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.set_actions([SpeakAloud])
        self._watch([UserRequirement, SpeakAloud])


    async def _observe(self) -> int:
        await super()._observe()
        # 自分に送られたメッセージを受け入れ、前のラウンドからの自分のメッセージを無視します
        self.rc.news = [msg for msg in self.rc.news if self.name in msg.send_to]
        return len(self.rc.news)

    async def _act(self) -> Message:
        logger.info(f"{self._setting}: ready to {self.rc.todo}")
        todo = self.rc.todo # SpeakAloudのインスタンス

        memories = self.get_memories()
        context = "\n".join(f"{msg.sent_from}: {msg.content}" for msg in memories)

        rsp = await todo.run(context=context, name=self.name, opponent_name=self.opponent_name)

        msg = Message(
            content=rsp,
            role=self.profile,
            cause_by=todo,
            sent_from=self.name,
            send_to=self.opponent_name,
        )

        return msg
async def debate(idea: str, investment: float = 3.0, n_round: int = 5):
    """バイデン-トランプディベートを実行し、彼らの友好的な対話を観察します :) """
    Biden = Debator(name="Biden", profile="Democrat", opponent_name="Trump")
    Trump = Debator(name="Trump", profile="Republican", opponent_name="Biden")
    team = Team()
    team.hire([Biden, Trump])
    team.invest(investment)
    team.run_project(idea, send_to="Biden")  # ディベートのトピックをバイデンに送信し、彼が最初に発言するようにします
    await team.run(n_round=n_round)


import typer
app = typer.Typer()

@app.command()
def main(
    idea: str = typer.Argument(..., help="Economic Policy: Discuss strategies and plans related to taxation, employment, fiscal budgeting, and economic growth."),
    investment: float = typer.Option(default=3.0, help="Dollar amount to invest in the AI company."),
    n_round: int = typer.Option(default=5, help="Number of rounds for the simulation."),
):
    """
    :param idea: