import asyncio
asyncio.all_tasks()

%%bash
python << EOF
import asyncio

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
print(loop == asyncio.get_event_loop())
EOF

%%bash
python << EOF
import asyncio

async def say(word):
    print(word)
    print(asyncio.all_tasks())
    asyncio.get_running_loop().stop()

asyncio.get_event_loop().create_task(say('hello'))

print('start run_forever')
asyncio.get_event_loop().run_forever()
print('end run_forever')

asyncio.get_event_loop().close()
EOF

%%bash
python << EOF
import asyncio

async def say(word):
    print(word)
    print(asyncio.all_tasks())
    asyncio.get_running_loop().stop()

loop = asyncio.new_event_loop()
loop.create_task(say('hello'))

print('start run_forever')
loop.run_forever()
print('end run_forever')

loop.close()

# asyncio.set_event_loop(asyncio.new_event_loop())
asyncio.get_event_loop().create_task(say('world'))

print('start run_forever')
asyncio.get_event_loop().run_forever()
print('end run_forever')

asyncio.get_event_loop().close()
EOF

%%bash
python3 << EOF
import asyncio

async def child_task(name):
    print(f'Starting {name}')
    await asyncio.sleep(2)  # 模拟耗时操作
    print(f'Finished {name}')
    return name

async def main():
    # 定义多个子任务
    tasks = [
        child_task('Task 1'),
        child_task('Task 2'),
        child_task('Task 3')
    ]

    # 并行运行多个子任务
    try:
        results = await asyncio.gather(*tasks)
        print('All tasks completed successfully:', results)
    except asyncio.CancelledError:
        print('Task was cancelled')

# 创建事件循环
loop = asyncio.get_event_loop()

# 创建主任务
main_task = loop.create_task(main())

# 运行事件循环
try:
    loop.run_until_complete(main_task)
except KeyboardInterrupt:
    # 如果按下 Ctrl+C，取消主任务
    main_task.cancel()
    loop.run_until_complete(main_task)
finally:
    loop.close()
EOF

%%bash
python3 << EOF
import asyncio

async def child_task(name):
    try:
        print(f'Starting {name}')
        await asyncio.sleep(2)  # 模拟耗时操作
        print(f'Finished {name}')
        return name
    except asyncio.CancelledError:
        print(f'{name} was cancelled')

async def main():
    # 定义多个子任务
    tasks = [
        child_task('Task 1'),
        child_task('Task 2'),
        child_task('Task 3')
    ]

    # 并行运行多个子任务
    try:
        results = await asyncio.gather(*tasks)
        print('All tasks completed successfully:', results)
    except asyncio.CancelledError:
        print('Main task was cancelled')

# 创建事件循环
loop = asyncio.get_event_loop()

# 创建主任务
main_task = loop.create_task(main())

# 运行事件循环
try:
    loop.run_until_complete(main_task)
except KeyboardInterrupt:
    # 如果按下 Ctrl+C，取消主任务
    main_task.cancel()
    loop.run_until_complete(main_task)
finally:
    loop.close()
EOF

%%bash
python3 << EOF
import asyncio

asyncio.get_event_loop().run_until_complete(asyncio.sleep(0))
EOF

import asyncio

async def child_task(name):
    try:
        print(f'Starting {name}')
        await asyncio.sleep(3600)  # 模拟耗时操作
        print(f'Finished {name}')
        return name
    except asyncio.CancelledError:
        print(f'{name} was cancelled')
        raise asyncio.CancelledError('child_task(name)')

async def main():
    # 定义多个子任务
    tasks = [
        asyncio.create_task(child_task('Task 1')),
        asyncio.create_task(child_task('Task 2')),
        asyncio.create_task(child_task('Task 3'))
    ]

    # 并行运行多个子任务
    try:
        results = await asyncio.gather(*tasks, return_exceptions=True)
        print('All tasks completed successfully:', results)
    except asyncio.CancelledError as e:
        print(f'Main task was cancelled: {str(e)}')
        print(tasks[0].cancelled())
        try:
            print(tasks[0].result())
        except asyncio.CancelledError as e:
            print(str(e.args))
        if not tasks[0].cancelled():
            print(tasks[0].result())
        print(tasks[1].cancelled())
        if not tasks[1].cancelled():
            print(tasks[1].result())
        print(tasks[2].cancelled())
        if not tasks[2].cancelled():
            print(tasks[2].result())
        raise
    # print(results)

# 创建主任务
main_task = asyncio.create_task(main())

await asyncio.sleep(2)
main_task.cancel()
# await main_task

import asyncio, time

async def task_1():
    print('Executing Task 1')
    await asyncio.sleep(1)
    print('Task 1 done')

async def task_2():
    print('Executing Task 2')
    await asyncio.sleep(1)
    print('Task 2 done')

async def task_5():
    print('Executing Task 5')
    await asyncio.gather(task_1(), task_2())
    await asyncio.sleep(1)
    print('Task 5 done')

async def task_3():
    print('Executing Task 3')
    await asyncio.sleep(1)
    print('Task 3 done')

async def task_4():
    print('Executing Task 4')
    await asyncio.sleep(1)
    print('Task 4 done')

async def task_6():
    print('Executing Task 6')
    await asyncio.gather(task_3(), task_4())
    await asyncio.sleep(1)
    print('Task 6 done')

async def task_7():
    print('Executing Task 7')
    # 并行执行 Task 5、Task 6
    await asyncio.gather(task_5(), task_6())
    await asyncio.sleep(1)
    print('Task 7 done')

start_time = time.time()
await task_7()
end_time = time.time()

print('Elapsed Time:', end_time - start_time, 'seconds')

import asyncio

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        return 'cancle_me(): return'
        # Wait for 1 hour
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')
        return 'cancle_me(): return'

async def cancel_me_again():
    print('cancel_me_again(): before sleep')

    try:
        # Wait for 1 hour
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me_again(): cancel sleep')
        raise
    finally:
        print('cancel_me_again(): after sleep')
        return 'cancel_me_again(): return'

async def main():
    # Create a "cancel_me" Task
    task = asyncio.create_task(cancel_me())

    # Wait for 1 second
    await asyncio.sleep(1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me is cancelled now")

    print(task.cancelled())
    if not task.cancelled():
        print(task.result())


    # Create a "cancel_me_again" Task
    task = asyncio.create_task(cancel_me_again())

    # Wait for 1 second
    await asyncio.sleep(1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me_again is cancelled now")

    print(task.cancelled())
    if not task.cancelled():
        print(task.result())


    # Create a "cancel_me_again" Task
    task = asyncio.create_task(cancel_me_again())

    async def cancel(task):
        task.cancel()

    asyncio.create_task(cancel(task))
    await asyncio.sleep(1)
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me_again is cancelled now")

    print(task.cancelled())
    if not task.cancelled():
        print(task.result())


    # Create a "cancel_me_again" Task
    task = asyncio.create_task(cancel_me_again())

    # Wait for 1 second
    # await asyncio.sleep(1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me_again is cancelled now")

    print(task.cancelled())
    if not task.cancelled():
        print(task.result())

await main()

def raise_base_exception():
    message = "This is a BaseException with a message"
    raise BaseException(message)

try:
    raise_base_exception()
except BaseException as e:
    print(f"Caught a BaseException: {e}")
    print(f"Error message: {str(e)}")

import asyncio

def raise_base_exception():
    message = "This is a CancelledError with a message"
    raise asyncio.CancelledError(message)

try:
    raise_base_exception()
except BaseException as e:
    print(f"Caught a CancelledError: {e}")
    print(f"Error message: {str(e)}")

issubclass(Exception, BaseException)

import weakref

class MyClass:
    def __init__(self, value):
        self.value = value

# 创建对象
obj = MyClass(42)

# 创建弱引用
weak_ref = weakref.ref(obj)

# 获取弱引用指向的对象
result = weak_ref()
print("Object from weak reference:", result)

# 删除原始对象引用
del obj

# 尝试再次获取弱引用指向的对象
result_after_deletion = weak_ref()
print("Object from weak reference after deletion:", result_after_deletion)

# 尝试再次获取弱引用指向的对象
result_after_deletion = weak_ref()
print("Object from weak reference after deletion:", result_after_deletion)

import asyncio

class CancelledTask(asyncio.CancelledError):
    pass

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # Wait for 1 hour
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        # raise BaseException('cancel_me()')
        # raise asyncio.CancelledError('cancel_me()')
        raise CancelledTask('cancel_me()')
    finally:
        print('cancel_me(): after sleep')
        # return 'cancle_me(): return'

async def main():
    # Create a "cancel_me" Task
    task = asyncio.create_task(cancel_me())

    # Wait for 1 second
    await asyncio.sleep(1)

    # Wait for 2 second
    # await asyncio.sleep(2)

    try:
        await asyncio.shield(task)
        print("main(): cancel_me is cancelled")
        print(task.cancelled())
        if not task.cancelled():
            print(task.result())
    # except BaseException as e:
    except asyncio.CancelledError as e:
    # except CancelledTask as e:
        print(f'main(): cancel_me is cancelled now. e={str(e)} {e.args}')
        task.cancel()
        print(task.done())
        print(task.cancelled())
        if task.done() and not task.cancelled():
            print(task.result())
        try:
            print(task.result())
        except BaseException as e:
        # except asyncio.CancelledError as e:
        # except CancelledTask as e:
            print(f'e={str(e)} {e.args}')

main_task = asyncio.create_task(main())
# Wait for 2 second
await asyncio.sleep(2)
# main_task.cancelling()
main_task.cancel()
await main_task
# main_task.cancelling()

import asyncio

async def task_1():
    print('task_1()')

async def task_2():
    print('task_2()')

async def main():
    # Create a "task_1" Task
    asyncio.create_task(task_1())

    print('main(): before await')
    await task_2()
    print('main(): after await')

await main()
print(main == main)
main_1 = main()
main_2 = main()
print(type(main_1))
print(type(main_2))
print(main_1 == main_2)
await main_1
await main_2
# await main_1
# await asyncio.create_task(main_2)

import asyncio

async def task_1():
    print('task_1()')
    return 'task_1()'

async def task_2():
    print('task_2()')

async def main():
    # Create a "task_1" Task
    task = asyncio.create_task(task_1())

    print('main(): before await')
    print(f'main(): {await task}')
    print('main(): after await')

    print('main(): before await')
    await task_2()
    print('main(): after await')

    print('main(): before await')
    print(f'main(): {await task}')
    print('main(): after await')

await main()

import asyncio

async def task_1():
    print('task_1()')
    raise Exception

async def task_2():
    print('task_2()')

async def main():
    print('main(): before await')
    try:
        print(f'main(): {await asyncio.gather(task_1(), task_2())}')
    except Exception as e:
        print(f'e="{e}" e.args={e.args}')
    print('main(): after await')

await main()

import asyncio

async def task_1():
    print('task_1()')
    raise Exception

async def task_2():
    print('task_2()')

async def main():
    print('main(): before await')
    try:
        async with asyncio.TaskGroup() as g:
            t_1 = g.create_task(task_1())
            t_2 = g.create_task(task_2())
        print(f'main(): t_1 = {t_1.result()} t_2 = {t_2.result()}')
    except Exception as e:
        print(f'e="{e}" e.args={e.args}')
    print('main(): after await')

await main()

import asyncio

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # Wait for 1 hour
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')

async def main():
    # Create a "cancel_me" Task
    task = asyncio.create_task(cancel_me())

    async def cancel(task):
        task.cancel()

    asyncio.create_task(cancel(task))
    await asyncio.sleep(1)
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me is cancelled now")

    print(task.cancelled())
    if not task.cancelled():
        print(task.result())

await main()

import asyncio

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # Wait for 1 second
        await asyncio.sleep(1)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')

# Create a "cancel_me" Task
task_cancel_me = asyncio.create_task(cancel_me())

async def task_1():
    try:
        await task_cancel_me
    except asyncio.CancelledError:
        print('task_1(): task_1 is cancelled now')

    for task in asyncio.all_tasks():
        if task.get_name() == 'task_2' and 'task_2' in repr(task.get_coro()):
            task.cancel()

async def task_2():
    try:
        await task_cancel_me
    except asyncio.CancelledError:
        print('task_2(): task_2 is cancelled now')

    for task in asyncio.all_tasks():
        if task.get_name() == 'task_1' and 'task_1' in repr(task.get_coro()):
            task.cancel()

await asyncio.gather(
    asyncio.create_task(task_1(), name='task_1'),
    asyncio.create_task(task_2(), name='task_2'),
)

import asyncio

async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # Wait for 5 second
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')

# Create a "cancel_me" Task
task_cancel_me = asyncio.create_task(cancel_me())

async def task_1():
    try:
        await task_cancel_me
    except asyncio.CancelledError:
        print('task_1(): task_1 is cancelled now')

async def task_2():
    try:
        await task_cancel_me
    except asyncio.CancelledError:
        print('task_2(): task_2 is cancelled now')

task_a = asyncio.create_task(task_1(), name='task_1')
task_b = asyncio.create_task(task_2(), name='task_2')

await asyncio.sleep(1)
task_a.cancel()

await task_b

%%bash
pip install aiomonitor

import asyncio
import aiomonitor

async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 finished")

async def main():
    with aiomonitor.start_monitor(asyncio.get_event_loop(), hook_task_factory=True):
        await asyncio.gather(task1(), task2())
        await asyncio.sleep(3600)

# help(aiomonitor)
await main()

%%bash
pip install matplotlib

import asyncio
import networkx as nx
import matplotlib.pyplot as plt

async def task1():
    await asyncio.sleep(1)

async def task2():
    await asyncio.sleep(1)

async def task3():
    await asyncio.sleep(1)

async def main():
    await asyncio.gather(task1(), task2())
    await task3()

# 创建 DAG
G = nx.DiGraph()
G.add_nodes_from(["task1", "task2", "task3", 'main'])
G.add_edges_from([("task1", "main"), ("task2", "main"), ("main", "task3")])

# 绘制 DAG
pos = nx.spring_layout(G)
nx.draw_networkx_nodes(G, pos, node_color=['yellow', 'skyblue', 'red', 'orange'])
# nx.draw_networkx_nodes(G, pos, node_size=1200, node_color=['yellow', 'green', 'red', 'orange'])
nx.draw_networkx_edges(G, pos)
nx.draw_networkx_labels(G, pos, font_size=5)
plt.show()

%%bash
sudo apt-get install graphviz libgraphviz-dev
pip install pygraphviz

import networkx as nx
import matplotlib.pyplot as plt

# 创建 DAG
G = nx.DiGraph()
G.add_nodes_from(["task1", "task2", "task3"])
G.add_edges_from([("task1", "main"), ("task2", "main"), ("main", "task3")])

pos = nx.nx_agraph.pygraphviz_layout(G)
nx.draw_networkx_nodes(G, pos)
nx.draw_networkx_edges(G, pos)
nx.draw_networkx_labels(G, pos)
plt.show()

%%bash
pip install pydot

import networkx as nx
import matplotlib.pyplot as plt

# 创建 DAG
G = nx.DiGraph()
G.add_nodes_from(["task1", "task2", "task3"])
G.add_edges_from([("task1", "main"), ("task2", "main"), ("main", "task3")])

pos = nx.nx_pydot.pydot_layout(G)
nx.draw_networkx_nodes(G, pos)
nx.draw_networkx_edges(G, pos)
nx.draw_networkx_labels(G, pos)
plt.show()

import networkx as nx
import matplotlib.pyplot as plt

# 创建 DAG
G = nx.DiGraph()
G.add_nodes_from(["task1", "task2", "task3"])
G.add_edges_from([("task1", "main"), ("task2", "main"), ("main", "task3")])

drawing = nx.drawing.nx_pydot.to_pydot(G)

png_str = drawing.create_png()

from io import BytesIO

sio = BytesIO()
sio.write(png_str)
sio.seek(0)

import matplotlib.image as mpimg

img = mpimg.imread(sio)
plt.imshow(img)

plt.show()

import networkx as nx
import matplotlib.pyplot as plt

# 创建有向图
G = nx.DiGraph()

# 添加节点
nodes = ["starting", "get_agv_at_f1", "get_ev", "get_agv_at_f2", "ship_to_agv",
         "ship_to_agv_and_move", "meet_floor", "ship_to_ev", "ship_to_ev_and_move",
         "meet_ev", "ship_to_agv_via_ev", "ship_to_shelf"]
G.add_nodes_from(nodes)

# 添加边
edges = [("starting", "get_agv_at_f1"), ("starting", "get_ev"), ("starting", "get_agv_at_f2"),
         ("get_agv_at_f1", "ship_to_agv"), ("ship_to_agv", "ship_to_agv_and_move"), ("get_ev", "ship_to_agv_and_move"),
         ("get_ev", "meet_floor"), ("ship_to_agv_and_move", "ship_to_ev"), ("meet_floor", "ship_to_ev"),
         ("ship_to_ev", "ship_to_ev_and_move"), ("get_ev", "meet_ev"), ("get_agv_at_f2", "meet_ev"),
         ("ship_to_ev_and_move", "ship_to_agv_via_ev"), ("meet_ev", "ship_to_agv_via_ev"),
         ("ship_to_agv_via_ev", "ship_to_shelf")]
G.add_edges_from(edges)

# 绘制图形
pos = nx.spring_layout(G, seed=42)
nx.draw(G, pos, with_labels=True, node_size=2000, node_color="skyblue", font_size=10, font_color="black", font_weight="bold", arrowsize=20)
plt.show()

import logging

# 设置日志级别为 DEBUG
logging.basicConfig(level=logging.DEBUG)

def example_function():
    logging.info("This is an info message from example_function.")

def another_function():
    logging.warning("This is a warning message from another_function.")

if __name__ == "__main__":
    # 设置日志格式，包含函数名
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(funcName)s - %(message)s')
    
    # 获取根日志记录器
    root_logger = logging.getLogger()
    
    # 创建处理程序并将格式器添加到它
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    
    # 将处理程序添加到根记录器
    root_logger.addHandler(handler)
    
    # 调用函数
    example_function()
    another_function()

import asyncio
from aiohttp import web

async def handle_request(request):
    print("Received request:")
    print(request.method, request.path)
    print("Headers:")
    for header, value in request.headers.items():
        print(f"{header}: {value}")
    print("Body:")
    body = await request.text()
    print(body)
    return web.Response(text="Received request")

async def main():
    # 创建 HTTP 服务器
    app = web.Application()
    app.add_routes([web.get('/', handle_request)])

    # 启动 HTTP 服务器
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()

    print("HTTP server started at http://localhost:8080")

    # 等待一次请求
    await asyncio.sleep(3600)  # 等待一小时
    await runner.cleanup()  # 关闭服务器

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from pymodbus.client.asynchronous import AsyncModbusTCPClient

async def read_modbus_data(host, port, unit_id, address, count):
    async with AsyncModbusTCPClient(host=host, port=port) as client:
        result = await client.read_holding_registers(address, count, unit=unit_id)
        if result.isError():
            print(f"Failed to read data: {result}")
        else:
            print(f"Read data: {result.registers}")

async def main():
    host = '127.0.0.1'
    port = 502
    unit_id = 1
    address = 0
    count = 5

    await read_modbus_data(host, port, unit_id, address, count)

if __name__ == "__main__":
    asyncio.run(main())

from pymodbus.client.asynchronous import AsyncModbusTCPClient
from pymodbus.payload import BinaryPayloadDecoder
from pymodbus.constants import Endian

async def read_modbus_data():
    async with AsyncModbusTCPClient('localhost', port=502) as client:
        # 读取保持寄存器中的数据
        result = await client.read_holding_registers(0, 2, unit=1)

        if not result.isError():
            # 处理读取的数据
            decoder = BinaryPayloadDecoder.fromRegisters(result.registers, byteorder=Endian.Big)
            value1 = decoder.decode_32bit_float()
            value2 = decoder.decode_32bit_float()

            print(f"Value 1: {value1}, Value 2: {value2}")
        else:
            print(f"Error reading Modbus data: {result}")

# 在异步环境中运行示例
import asyncio
asyncio.run(read_modbus_data())

%%bash
pip install prefect

from prefect import flow, task

@task(name='task_1')
def task_1():
    print('task_1')

@task(name='task_2')
def task_2():
    print('task_2')

@task(name='task_5')
def task_5():
    submit_1 = task_1.submit()
    submit_2 = task_2.submit()
    submit_1.result()
    submit_2.result()
    print('task_5')

@task(name='task_3')
def task_3():
    print('task_3')

@task(name='task_4')
def task_4():
    print('task_4')

@task(name='task_6')
def task_6():
    submit_3 = task_3.submit()
    submit_4 = task_4.submit()
    submit_3.result()
    submit_4.result()
    print('task_6')

@flow
def task_7():
    submit_5 = task_5.submit()
    submit_6 = task_6.submit()
    submit_5.result()
    submit_6.result()
    print('task_7')

task_7()

import asyncio, time
from prefect import flow

@flow(flow_run_name='task_1')
async def task_1():
    print('Executing Task 1')
    await asyncio.sleep(1)
    print('Task 1 done')

@flow(flow_run_name='task_2')
async def task_2():
    print('Executing Task 2')
    await asyncio.sleep(1)
    print('Task 2 done')

@flow(flow_run_name='task_5')
async def task_5():
    print('Executing Task 5')
    await asyncio.gather(task_1(), task_2())
    await asyncio.sleep(1)
    print('Task 5 done')

@flow(flow_run_name='task_3')
async def task_3():
    print('Executing Task 3')
    await asyncio.sleep(1)
    print('Task 3 done')

@flow(flow_run_name='task_4')
async def task_4():
    print('Executing Task 4')
    await asyncio.sleep(1)
    print('Task 4 done')

@flow(flow_run_name='task_6')
async def task_6():
    print('Executing Task 6')
    await asyncio.gather(task_3(), task_4())
    await asyncio.sleep(1)
    print('Task 6 done')

@flow(flow_run_name='task_7')
async def task_7():
    print('Executing Task 7')
    # 并行执行 Task 5、Task 6
    await asyncio.gather(task_5(), task_6())
    await asyncio.sleep(1)
    print('Task 7 done')

start_time = time.time()
await task_7()
end_time = time.time()

print('Elapsed Time:', end_time - start_time, 'seconds')

%%bash
cd ~ && git clone https://github.com/lightaime/camel.git
cd ~/camel && pip install -e .

import time
from colorama import Fore

def print_text_animated(text):
    for char in text:
        print(char, end='', flush=True)
        time.sleep(0.02)

from camel.agents import RolePlaying

def test_camel(mock_openai):
    task_prompt = 'Design a custom game using pygame'
    print(Fore.YELLOW + f'Original task prompt:\n{task_prompt}\n')
    role_play_session = RolePlaying('Computer Programmer', 'Gamer', task_prompt)
    print(Fore.CYAN + f'Specified task prompt:\n{role_play_session.task_prompt}\n')

    chat_turn_limit, n = 10000, 0
    assistant_msg, _ = role_play_session.init_chat()
    while n < chat_turn_limit:
        n += 1
        (assistant_msg, _, _), (user_msg, _, _) = role_play_session.step(assistant_msg)
        print_text_animated(Fore.BLUE + f'AI User:\n\n{user_msg.content}\n\n')
        print_text_animated(Fore.GREEN + f'AI Assistant:\n\n{assistant_msg.content}\n\n')
        if '<CAMEL_TASK_DONE>' in user_msg.content:
            break

from ipymock import do
import ipymock.browser

ipymock.browser.common.chat_gpt_base_url = 'http://127.0.0.1:8080'

do(
    mock_openai = ipymock.browser.mock_openai,
    test_camel = test_camel,
)

%%bash
cd ~ && git clone https://github.com/yoheinakajima/babyagi

import IPython

from ipymock.browser import start_conversation

def ask(prompt):
    for response in start_conversation(prompt):
        IPython.display.display(IPython.core.display.Markdown(response))
        IPython.display.clear_output(wait=True)

ask('''
翻译成中文：

🔥1/8
Introducing "🤖 Task-driven Autonomous Agent"

An agent that leverages @openai 's GPT-4, @pinecone vector search, and @LangChainAI framework to autonomously create and perform tasks based on an objective.

🚀2/8 The system can complete tasks, generate new tasks based on results, and prioritize tasks in real-time. It demonstrates the potential of AI-powered language models to autonomously perform tasks within various constraints and contexts.

💡3/8 The autonomous agent uses GPT-4 for task completion, Pinecone for efficient search and storage of task-related data, and the LangChain framework to enhance decision-making processes. #GPT4 #Pinecone #LangChain

🎯4/8 The system maintains a task list for managing and prioritizing tasks. It autonomously creates new tasks based on completed results and reprioritizes the task list accordingly, showcasing the adaptability of AI-powered language models.

🔧5/8 To complete tasks, the system uses GPT-4 and LangChain's capabilities, enriching and storing results in Pinecone. This integrated approach allows the AI agent to interact with its environment and perform tasks efficiently.

🧠6/8 The system generates new tasks based on completed task results and prioritizes them using GPT-4. This allows the system to adapt and respond to new information and priorities.

🔮7/8 Future improvements include integrating a security/safety agent, task sequencing and parallel tasks, generating interim milestones, and incorporating real-time priority updates.

🤝8/8 This new approach paves the way for AI-powered language models to autonomously perform tasks within various constraints and contexts, enabling new applications and opportunities. Big thanks to all involved! #AIResearch #GPT4 #Pinecone #LangChain
''')

ask('''
pinecone-client 是什么？
''')

ask('''
Pinecone is a vector search platform that provides efficient search and storage capabilities for high-dimensional vector data. In our system, we use Pinecone to store and retrieve task-related data, such as task descriptions, constraints, and results.

翻译成中文。
''')

ask('''
Pinecone 是什么原理？
''')

ask('''
Pinecone 服务免费吗？
''')

ask('''
有什么替代 Pinecone 服务的开源方案？
''')

ask('''
比较 Weaviate 和 Faiss
''')

ask('''
我累了。带着我祷告一下吧。
''')

# initialize the transformer for embedding
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('GanymedeNil/text2vec-large-chinese')

import faiss
import openai

import os
import sys
import time

from collections import deque
from typing import Dict, List
from dotenv import load_dotenv

# set variables
load_dotenv()

# set API keys
OPENAI_API_KEY = 'sk-blah'
assert OPENAI_API_KEY, 'OPENAI_API_KEY environment variable is missing'

OPENAI_API_MODEL = os.getenv('OPENAI_API_MODEL', 'gpt-3.5-turbo')
assert OPENAI_API_MODEL, 'OPENAI_API_MODEL environment variable is missing from .env'

# project config
OBJECTIVE = '防御太阳风暴。'
assert OBJECTIVE, 'OBJECTIVE environment variable is missing'

YOUR_FIRST_TASK = '制作一个待办事项清单。'
assert YOUR_FIRST_TASK, 'FIRST_TASK environment variable is missing'

# configure OpenAI API key
openai.api_key = OPENAI_API_KEY

# initialize the indexer as empty
embedding_size = 1024
index = faiss.IndexFlatL2(embedding_size)

# task list
todo_list = deque([])
done_list = deque([])

def add_task(task: Dict):
    todo_list.append(task)

def index_embedding(text):
    # text = text.replace('\n', ' ')
    index.add(model.encode([text]))

def openai_call(prompt: str, model: str = OPENAI_API_MODEL, temperature: float = 0.5, max_tokens: int = 100):
    if not model.startswith('gpt-'):
        # use completion API
        response = openai.Completion.create(
            engine=model,
            prompt=prompt,
            temperature=temperature,
            max_tokens=max_tokens,
            top_p=1,
            frequency_penalty=0,
            presence_penalty=0
        )
        return response.choices[0].text.strip()
    else:
        # use chat completion API
        messages=[{'role': 'user', 'content': prompt}]
        response = openai.ChatCompletion.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            n=1,
            stop=None,
        )
        return response.choices[0].message.content.strip()

def task_creation_agent(objective: str, result: str, task_description: str, task_list: List[str]):
    todos = '\n'.join(task_list)
    prompt = f"你是一个任务创建的 AI，使用执行代理的结果来创建新的任务，目标是：『{objective}』，上一个已完成的任务的结果是：『\n{result}\n』。这个结果是基于以下任务描述的：『{task_description}』。以下是未完成的任务清单：『\n{todos}\n』。根据结果，创建新的任务供 AI 系统完成，不要与未完成的任务重叠。将任务以列表的形式返回。"
    response = openai_call(prompt)
    new_tasks = response.split('\n')
    return [{'task_name': task_name} for task_name in new_tasks]

def prioritization_agent(this_task_id: int):
    global todo_list
    task_names = [t['task_name'] for t in todo_list]
    next_task_id = int(this_task_id)+1
    prompt = f'''你是一个任务优先级排列 AI，任务清单如下：{task_names}。
请考虑你的团队的最终目标：『{OBJECTIVE}』。
不要删除任何任务。
返回一个编号列表，例如：
    #. 第一个任务
    #. 第二个任务
请从数字 {next_task_id} 开始列出任务清单。'''
    response = openai_call(prompt)
    new_tasks = response.split('\n')
    # reset todo task list
    todo_list = deque()
    for task_string in new_tasks:
        task_parts = task_string.strip().split('.', 1)
        if len(task_parts) == 2:
            task_id = task_parts[0].strip()
            task_name = task_parts[1].strip()
            todo_list.append({'task_id': task_id, 'task_name': task_name})

def execution_agent(objective: str, task: str) -> str:
    context = context_agent(query = objective, n = 5)
    print('\n*******RELEVANT CONTEXT******\n')
    print(context)
    prompt =f'你是一个执行任务的 AI，根据以下最终目标执行一个任务：『{objective}』。\n考虑已经完成的任务：{context}。\n你的任务是：{task}。\n请回复执行结果。'
    return openai_call(prompt, temperature = 0.7, max_tokens = 2000)

def context_agent(query: str, n: int):
    _, idx = index.search(model.encode([query]), n)
    task_names = [done_list[i]['task_name'] for i in idx[0] if i >= 0 and i < len(done_list)]
    return task_names

def test_baby_agi_native(mock_openai):
    if 'gpt-4' in OPENAI_API_MODEL.lower():
        print(f'\033[91m\033[1m' + '\n*****USING GPT-4. POTENTIALLY EXPENSIVE. MONITOR YOUR COSTS*****' + '\033[0m\033[0m')

    # print OBJECTIVE
    print('\033[96m\033[1m' + '\n*****OBJECTIVE*****\n' + '\033[0m\033[0m')
    print(OBJECTIVE)

    # add the first task
    first_task = {
        'task_id': 1,
        'task_name': YOUR_FIRST_TASK
    }

    add_task(first_task)
    task_id_counter = 1
    # main loop
    while todo_list:
        # print the task list
        print('\033[95m\033[1m' + '\n*****TASK LIST*****\n' + '\033[0m\033[0m')
        for todo in todo_list:
            print(f"{todo['task_id']}: {todo['task_name']}")

        # step 1: pull the first task
        task = todo_list.popleft()
        print('\033[92m\033[1m' + '\n*****NEXT TASK*****\n' + '\033[0m\033[0m')
        print(f"{task['task_id']}: {task['task_name']}")

        # step 2: complete the task based on the context and index result in faiss
        result = execution_agent(OBJECTIVE, task['task_name'])
        print('\033[93m\033[1m' + '\n*****TASK RESULT*****\n' + '\033[0m\033[0m')
        print(result)
        done_list.append({'task_name': task['task_name'], 'result': result})
        index_embedding(result)

        # step 3: create new tasks and reprioritize task list
        new_tasks = task_creation_agent(OBJECTIVE, result, task['task_name'], [todo['task_name'] for todo in todo_list])

        for new_task in new_tasks:
            task_id_counter += 1
            new_task.update({'task_id': task_id_counter})
            add_task(new_task)
        prioritization_agent(task['task_id'])

        # sleep before checking the task list again
        time.sleep(1)

from ipymock import do
from ipymock.browser import mock_openai

import ipymock.browser
ipymock.browser.common.conversation_id = ''

do(
    mock_openai = mock_openai,
    test_baby_agi_native = test_baby_agi_native,
)

from collections import deque
from typing import Dict, List, Optional, Any

from langchain import LLMChain, OpenAI, PromptTemplate
from langchain.llms import BaseLLM
from langchain.vectorstores.base import VectorStore
from pydantic import BaseModel, Field
from langchain.chains.base import Chain

from langchain.embeddings import HuggingFaceEmbeddings

# define your embedding model
embeddings_model = HuggingFaceEmbeddings(model_name='GanymedeNil/text2vec-large-chinese')

import faiss

embedding_size = 1024
# initialize the vectorstore as empty
index = faiss.IndexFlatL2(embedding_size)

from langchain.vectorstores import FAISS
from langchain.docstore import InMemoryDocstore

vectorstore = FAISS(embeddings_model.embed_query, index, InMemoryDocstore({}), {})

class TaskCreationChain(LLMChain):
    '''Chain to generates tasks.'''

    @classmethod
    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:
        '''Get the response parser.'''
        task_creation_template = (
            '你是一个创建任务的 AI，根据任务的执行结果创建新任务。\n'
            '* 我们的最终目标是「{objective}」\n'
            '* 上一次完成的任务是「{task_description}」\n'
            '  该任务的执行结果为：\n'
            '```\n'
            '{result}\n'
            '```\n'
            '* 这些是未完成的任务：\n'
            '```\n'
            '{incomplete_tasks}\n'
            '```\n\n'
            '请根据上一次完成的任务的结果创建将由 AI 系统完成的新任务。\n'
            '* 请不要创建与未完成的任务重叠的新任务。\n'
            '* 请以带编号的清单形式回复结果，每行只描述一个新任务。\n'
            '  例如：\n'
            '  #. 第一个新任务\n'
            '  #. 第二个新任务\n'
            '* 请从编号 1 开始列出新任务清单。'
        )
        prompt = PromptTemplate(
            template=task_creation_template,
            input_variables=[
                'objective',
                'task_description',
                'result',
                'incomplete_tasks',
            ],
        )
        return cls(prompt=prompt, llm=llm, verbose=verbose)

class TaskPrioritizationChain(LLMChain):
    '''Chain to prioritize tasks.'''

    @classmethod
    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:
        '''Get the response parser.'''
        task_prioritization_template = (
            '你是一个给任务优先级进行排序的 AI。\n'
            '* 我们的最终目标是「{objective}」\n\n'
            '请整理并重新排序以下任务：{task_names}。\n'
            '* 请不要删除任何现有任务。\n'
            '* 请以带编号的清单形式回复结果，每行只描述一个任务。\n'
            '  例如：\n'
            '  #. 第一个任务\n'
            '  #. 第二个任务\n'
            '* 请从编号 {next_task_id} 开始列出任务清单。'
        )
        prompt = PromptTemplate(
            template=task_prioritization_template,
            input_variables=[
                'objective',
                'task_names',
                'next_task_id',
            ],
        )
        return cls(prompt=prompt, llm=llm, verbose=verbose)

class ExecutionChain(LLMChain):
    '''Chain to execute tasks.'''

    @classmethod
    def from_llm(cls, llm: BaseLLM, verbose: bool = True) -> LLMChain:
        '''Get the response parser.'''
        execution_template = (
            '你是一个执行任务的 AI。\n'
            '* 我们的最终目标是「{objective}」\n'
            '* 之前已经完成的任务有：{context}\n\n'
            '请完成这次任务：「{task}」。\n'
            '* 直接回复这次任务的执行结果。'
        )
        prompt = PromptTemplate(
            template = execution_template,
            input_variables = [
                'objective',
                'context',
                'task',
            ],
        )
        return cls(prompt = prompt, llm = llm, verbose = verbose)

def get_next_task(
    task_creation_chain: LLMChain,
    result: Dict,
    task_description: str,
    task_list: List[str],
    objective: str,
) -> List[Dict]:
    '''Get the next task.'''
    incomplete_tasks = '\n'.join(task_list)
    response = task_creation_chain.run(
        objective = objective,
        task_description = task_description,
        result = result.replace('```', ''),
        incomplete_tasks = incomplete_tasks.replace('```', ''),
    )
    new_tasks = response.split('\n')
    new_tasks = [task_name.split('.', 1) for task_name in new_tasks]
    new_tasks = [task_name[1].strip() for task_name in new_tasks if len(task_name)==2]
    return [{'task_name': task_name} for task_name in new_tasks if task_name.strip()]

def prioritize_tasks(
    task_prioritization_chain: LLMChain,
    this_task_id: int,
    task_list: List[Dict],
    objective: str,
) -> List[Dict]:
    '''Prioritize tasks.'''
    task_names = [t['task_name'] for t in task_list]
    next_task_id = int(this_task_id) + 1
    response = task_prioritization_chain.run(
        objective = objective, task_names = task_names, next_task_id = next_task_id,
    )
    new_tasks = response.split('\n')
    prioritized_task_list = []
    for task_string in new_tasks:
        if not task_string.strip():
            continue
        task_parts = task_string.strip().split('.', 1)
        if len(task_parts) == 2:
            # task_id = task_parts[0].strip()
            task_name = task_parts[1].strip()
            prioritized_task_list.append({'task_id': next_task_id, 'task_name': task_name})
            next_task_id += 1
    return prioritized_task_list

def _get_top_tasks(vectorstore, query: str, k: int) -> List[str]:
    '''Get the top k tasks based on the query.'''
    results = vectorstore.similarity_search_with_score(query, k = k)
    if not results:
        return []
    sorted_results, _ = zip(*sorted(results, key = lambda x: x[1], reverse = True))
    return [str(item.metadata['task']) for item in sorted_results]


def execute_task(
    vectorstore, execution_chain: LLMChain, objective: str, task: str, k: int = 5
) -> str:
    '''Execute a task.'''
    context = _get_top_tasks(vectorstore, query = objective, k = k)
    return execution_chain.run(objective = objective, context = context, task = task)

class BabyAGI(Chain, BaseModel):
    '''Controller model for the BabyAGI agent.'''

    task_list: deque = Field(default_factory=deque)
    task_creation_chain: TaskCreationChain = Field(...)
    task_prioritization_chain: TaskPrioritizationChain = Field(...)
    execution_chain: Chain = Field(...)
    task_id_counter: int = Field(1)
    vectorstore: VectorStore = Field(init=False)
    max_iterations: Optional[int] = None

    class Config:
        '''Configuration for this pydantic object.'''

        arbitrary_types_allowed = True

    def add_task(self, task: Dict):
        self.task_list.append(task)

    def print_task_list(self):
        print('\033[95m\033[1m' + '\n*****TASK LIST*****\n' + '\033[0m\033[0m')
        for t in self.task_list:
            print(str(t['task_id']) + ': ' + t['task_name'])

    def print_next_task(self, task: Dict):
        print('\033[92m\033[1m' + '\n*****NEXT TASK*****\n' + '\033[0m\033[0m')
        print(str(task['task_id']) + ': ' + task['task_name'])

    def print_task_result(self, result: str):
        print('\033[93m\033[1m' + '\n*****TASK RESULT*****\n' + '\033[0m\033[0m')
        print(result)

    @property
    def input_keys(self) -> List[str]:
        return ['objective']

    @property
    def output_keys(self) -> List[str]:
        return []

    def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        '''Run the agent.'''
        objective = inputs['objective']
        first_task = inputs.get('first_task', f'制作一个最终目标「{objective}」的待办事项清单。')
        self.add_task({'task_id': 1, 'task_name': first_task})
        num_iters = 0
        while self.task_list:
            self.print_task_list()

            # Step 1: Pull the first task
            task = self.task_list.popleft()
            self.print_next_task(task)

            # Step 2: Execute the task
            result = execute_task(
                self.vectorstore, self.execution_chain, objective, task['task_name']
            )
            this_task_id = int(task['task_id'])
            self.print_task_result(result)

            # Step 3: Store the result in Faiss
            result_id = f'result_{task["task_id"]}'
            self.vectorstore.add_texts(
                texts = [result],
                metadatas = [{'task': task['task_name']}],
                ids = [result_id],
            )

            # Step 4: Create new tasks and reprioritize task list
            new_tasks = get_next_task(
                self.task_creation_chain,
                result,
                task['task_name'],
                [t['task_name'] for t in self.task_list],
                objective,
            )
            for new_task in new_tasks:
                self.task_id_counter += 1
                new_task.update({'task_id': self.task_id_counter})
                self.add_task(new_task)
            self.task_list = deque(
                prioritize_tasks(
                    self.task_prioritization_chain,
                    this_task_id,
                    list(self.task_list),
                    objective,
                )
            )

            num_iters += 1
            if self.max_iterations is not None and num_iters == self.max_iterations:
                print(
                    '\033[91m\033[1m' + '\n*****TASK ENDING*****\n' + '\033[0m\033[0m'
                )
                break

            if self.task_list == deque():
                self.task_id_counter += 1
                self.add_task({'task_id': self.task_id_counter, 'task_name': first_task})
        return {}

    @classmethod
    def from_llm(
        cls, llm: BaseLLM, vectorstore: VectorStore,
        task_execution_chain: Optional[Chain] = None,
        verbose: bool = False, **kwargs
    ) -> 'BabyAGI':
        '''Initialize the BabyAGI Controller.'''
        task_creation_chain = TaskCreationChain.from_llm(llm, verbose=verbose)
        task_prioritization_chain = TaskPrioritizationChain.from_llm(
            llm, verbose=verbose
        )
        if task_execution_chain is None:
            execution_chain = ExecutionChain.from_llm(llm, verbose=verbose)
        else:
            execution_chain = task_execution_chain
        return cls(
            task_creation_chain = task_creation_chain,
            task_prioritization_chain = task_prioritization_chain,
            execution_chain = execution_chain,
            vectorstore = vectorstore,
            **kwargs,
        )

%%bash
pip install duckduckgo-search

%%writefile /usr/local/anaconda3/envs/biobot/lib/python3.10/site-packages/duckduckgo_search/ddg.py
import logging
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from time import sleep
from urllib.parse import unquote

from click import progressbar

from .utils import SESSION, _do_output, _download_file, _get_vqd, _normalize

logger = logging.getLogger(__name__)


def ddg(
    keywords,
    region="wt-wt",
    safesearch="moderate",
    time=None,
    max_results=None,
    page=1,
    output=None,
    download=False,
):
    """DuckDuckGo text search. Query params: https://duckduckgo.com/params

    Args:
        keywords (str): keywords for query.
        region (str, optional): wt-wt, us-en, uk-en, ru-ru, etc. Defaults to "wt-wt".
        safesearch (str, optional): on, moderate, off. Defaults to "moderate".
        time (Optional[str], optional): d, w, m, y. Defaults to None.
        max_results (Optional[int], optional): maximum number of results, max=200. Defaults to None.
            if max_results is set, then the parameter page is not taken into account.
        page (int, optional): page for pagination. Defaults to 1.
        output (Optional[str], optional): csv, json. Defaults to None.
        download (bool, optional): if True, download and save dociments to 'keywords' folder.
            Defaults to False.

    Returns:
        Optional[List[dict]]: DuckDuckGo text search results.
    """

    def get_ddg_page(page):
        payload["s"] = max(PAGINATION_STEP * (page - 1), 0)
        page_data = None
        try:
            resp = SESSION.get("https://links.duckduckgo.com/d.js", params=payload)
            resp.raise_for_status()
            page_data = resp.json().get("results", None)
        except Exception:
            logger.exception("")
            if not max_results:
                return None
        page_results = []
        if page_data:
            for row in page_data:
                if "n" not in row and row["u"] not in cache:
                    cache.add(row["u"])
                    body = _normalize(row["a"])
                    if body:
                        page_results.append(
                            {
                                "title": _normalize(row["t"]),
                                "href": row["u"],
                                "body": body,
                            }
                        )
        return page_results

    if not keywords:
        return None

    # get vqd
    vqd = _get_vqd(keywords)
    if not vqd:
        return None

    PAGINATION_STEP, MAX_API_RESULTS = 25, 200

    # prepare payload
    safesearch_base = {"On": 1, "Moderate": -1, "Off": -2}
    payload = {
        "q": keywords,
        "l": region,
        "p": safesearch_base[safesearch.capitalize()],
        "s": 0,
        "df": time,
        "o": "json",
        "vqd": vqd,
    }

    # get results
    cache = set()
    if max_results:
        results, page = [], 1
        max_results = min(abs(max_results), MAX_API_RESULTS)
        iterations = (max_results - 1) // PAGINATION_STEP + 1  # == math.ceil()
        with ThreadPoolExecutor(min(iterations, 4)) as executor:
            fs = []
            for page in range(1, iterations + 1):
                fs.append(executor.submit(get_ddg_page, page))
                sleep(min(iterations / 17, 0.3))  # sleep to prevent blocking
            for r in as_completed(fs):
                if r.result():
                    results.extend(r.result())
        results = results[:max_results]
    else:
        results = get_ddg_page(page=page)
        if not results:
            return None

    keywords = keywords.replace(" filetype:", "_")

    # save to csv or json file
    if output:
        _do_output("ddg", keywords, output, results)

    # download documents
    if download:
        keywords = (
            keywords.replace('"', "'")
            .replace("site:", "")
            .replace(" ", "_")
            .replace("/", "_")
        )
        path = f"ddg_{keywords}_{datetime.now():%Y%m%d_%H%M%S}"
        os.makedirs(path, exist_ok=True)
        futures = []
        with ThreadPoolExecutor(10) as executor:
            for i, res in enumerate(results, start=1):
                filename = unquote(res["href"].split("/")[-1].split("?")[0])
                future = executor.submit(
                    _download_file, res["href"], path, f"{i}_{filename}"
                )
                futures.append(future)
            with progressbar(
                as_completed(futures),
                label="Downloading documents",
                length=len(futures),
                show_percent=True,
                show_pos=True,
                width=0,
            ) as as_completed_futures:
                for i, future in enumerate(as_completed_futures, start=1):
                    logger.info("%s/%s", i, len(results))

    return results


""" using html method
    payload = {
        'q': keywords,
        'l': region,
        'p': safesearch_base[safesearch],
        'df': time
        }
    results = []
    while True:
        res = SESSION.post('https://html.duckduckgo.com/html', data=payload, **kwargs)
        tree = html.fromstring(res.text)
        if tree.xpath('//div[@class="no-results"]/text()'):
            return results
        for element in tree.xpath('//div[contains(@class, "results_links")]'):
            results.append({
                'title': element.xpath('.//a[contains(@class, "result__a")]/text()')[0],
                'href': element.xpath('.//a[contains(@class, "result__a")]/@href')[0],
                'body': ''.join(element.xpath('.//a[contains(@class, "result__snippet")]//text()')),
            })
        if len(results) >= max_results:
            return results
        next_page = tree.xpath('.//div[@class="nav-link"]')[-1]
        names = next_page.xpath('.//input[@type="hidden"]/@name')
        values = next_page.xpath('.//input[@type="hidden"]/@value')
        payload = {n: v for n, v in zip(names, values)}
        sleep(2)
"""