使用 Autogen 0.4 构建我自己的概念验证代理

Autogen 0.4 — 使用代理创建和执行 Python 代码

我的目标是看看我是否可以让代理团队完成我的一些工作，并且我可以将工作委托给代理。目前这只是一个实验。需要更多的工作来创建一个可在生产规模使用的稳定版本。

概述

• 创建一个由代理执行各种任务的协调
• 首先创建一个可以由代理执行的 Python 脚本
• 然后执行脚本并存储输出
• 基于之前的输出创建预测 Python 代码
• 执行预测代码并存储输出
• 使用 autogen 0.4.1 稳定版本进行测试
• 不是生产就绪代码，仅将其视为概念证明（POC）

需求

• autogen 0.4.1
• Azure 订阅
• Azure Open AI 资源
• python 3.12 或更高版本

设置

• 首先创建一个 Azure Open AI 资源，并获取 API 密钥和端点
• 创建一个 python 环境并安装所需的包
• 安装 autogen 0.4.1
• 安装其他必要的包

pip install yfinance matplotlib scikit-learn

• 我还没有弄清楚如何使用 autogen 代理自动安装包
• 我现在正在创建一个控制台应用程序
• 现在导入必要的库并设置环境变量

from collections import defaultdict
import datetime
from pathlib import Path
import tempfile
import time
from typing import Sequence
import PyPDF2
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_agentchat.messages import AgentEvent, ChatMessage
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import AzureOpenAIChatCompletionClient
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import MagenticOneGroupChat
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core.code_executor import CodeBlock
from autogen_core import CancellationToken
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_agentchat.ui import Console
from autogen.code_utils import extract_code
from autogen.code_utils import create_virtual_env
import asyncio
import os
import venv
from bs4 import BeautifulSoup
from openai import AzureOpenAI
from dotenv import load_dotenv
import streamlit as st
from PIL import Image
import base64
import requests
import io

## 加载 .env 文件
load_dotenv()

• .env 文件用于环境变量的创建
• 现在设置 Azure Open AI 客户端以及 autogen 客户端

client = AzureOpenAI(
  azure_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT_VISION"), 
  api_key=os.getenv("AZURE_OPENAI_KEY_VISION"),  
  api_version="2024-10-21"
)

model_name = "gpt-4o-2"

model_client = AzureOpenAIChatCompletionClient(model="gpt-4o", 
                                               azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"), 
                                               api_key=os.getenv("AZURE_OPENAI_API_KEY"), 
                                               api_version="2024-10-21")

• 我关注的不是使用 Open AI 版本，而是 Azure Open AI

• 这是包含以下过程的主文件

• 流程如下

• 首先要求 LLM 创建一段 Python 代码以获取特斯拉过去 6 个月的股票价格，保存为 stocks.csv 然后打印

• 我正在打印输出以检查代码是否正确

• 现在从输出中提取代码

• 执行上述步骤创建的代码。这是为了下载数据集以用于基于数据科学的预测

• 现在使用 LLM 创建一段 Python 代码，以使用 sklearn 预测收盘价，使用 stocks.csv 文件作为输入

• 从输出中提取代码

• 执行使用 sklearn 创建的预测代码并打印输出

• 还将预测结果保存到 stockspred.csv

• 两个 csv 文件应位于编码文件夹中

• 首先输出代码，然后输出代码执行的结果，获取数据并保存

• 现在创建机器学习代码

async def main():
    st.set_page_config(page_title="CodeAgent", page_icon=":robot:")
    st.title("CodeAgent :robot:")
    st.write("一个可以阅读和编写代码以解决编程问题的 AI 代理。")
    work_dir = Path("coding")
    work_dir.mkdir(exist_ok=True)

    code_create = AssistantAgent(
        "CreateCodeAgent",
        description="为给定任务创建代码。",
        #tools=[bing_search_and_summarize],
        model_client=model_client,
        system_message="""
        为给定任务创建代码。以 Python 编写代码，仅将代码作为输出响应。
        还要添加运行代码所需的必要包。
        如果代码需要任何数据，请创建一个 json 文件并在代码中添加 json 文件的链接。
        如果有数据，请将输出保存为名为 stocks.csv 的 csv 文件。
        """,        
    )

    # 使用 GPTAssistantAgent 生成代码
    #code_create.initiate_chat(model_client, message="写一个打印 'Hello, World!' 的 Python 函数")
    text_mention_termination = TextMentionTermination("TERMINATE")
    max_messages_termination = MaxMessageTermination(max_messages=25)
    termination = text_mention_termination | max_messages_termination
    team = MagenticOneGroupChat([code_create], 
                                 model_client=model_client,
                                 termination_condition=termination, max_turns=1)
    
    # 提取生成的代码
    result = await Console(team.run_stream(task="写一个 Python 函数以聊天过去 6 个月的特斯拉股票价格并以表格形式打印。"))
    last_message = result.messages
    code_blocks = extract_code(last_message[-1].content)

    shell_commands = "pip install yfinance matplotlib"
    python_code = ""
    if code_blocks:
        print("提取的代码块:", code_blocks)
        #shell_commands = next(block[1] for block in code_blocks if block[0] == 'sh' or block[0] == 'bash' or block[0] == 'shell')
        #print(shell_commands)
        python_code = next(block[1] for block in code_blocks if block[0] == 'python')
        #print('仅 Python 代码: ', python_code)

    
    try:
        venv_dir = ".venv"
        venv_context = create_virtual_env(venv_dir)

        executor = LocalCommandLineCodeExecutor(virtual_env_context=venv_context, work_dir=work_dir)
        shresult = await executor.execute_code_blocks(
                code_blocks=[
                    CodeBlock(language="shell", code=shell_commands),
                ],
                cancellation_token=CancellationToken(),
            )
        print(shresult.output.strip())
    except Exception as e:
        pass

    local_executor = LocalCommandLineCodeExecutor(work_dir=work_dir)
    result = await local_executor.execute_code_blocks(
            code_blocks=[
                CodeBlock(language="python", code=python_code),
            ],
            cancellation_token=CancellationToken(),
        )
    #print(result)
    print(result.output.strip())

    code_create = AssistantAgent(
        "CreateCodeAgent",
        description="为给定任务创建代码。",
        #tools=[bing_search_and_summarize],
        model_client=model_client,
        system_message="""
        为 stocks.csv 创建代码以预测收盘价，使用 lstm。
        还要添加运行代码所需的必要包。
        如果代码需要任何数据，请创建一个 json 文件并在代码中添加 json 文件的链接。
        如果有数据，请将输出保存为名为 stockspred.csv 的 csv 文件。
        """,        
    )

    # 使用 GPTAssistantAgent 生成代码
    #code_create.initiate_chat(model_client, message="写一个打印 'Hello, World!' 的 Python 函数")
    text_mention_termination = TextMentionTermination("TERMINATE")
    max_messages_termination = MaxMessageTermination(max_messages=25)
    termination = text_mention_termination | max_messages_termination
    team = MagenticOneGroupChat([code_create], 
                                 model_client=model_client,
                                 termination_condition=termination, max_turns=1)
    

    # 提取生成的代码
    result = await Console(team.run_stream(task="写一段 Python 代码以使用 sklearn 预测 stocks.csv 的收盘价并以表格形式打印。同时保存为 stockspred.csv"))
    last_message = result.messages
    code_blocks = extract_code(last_message[-1].content)

    shell_commands = "pip install yfinance matplotlib scikit-learn"
    python_code = ""
    if code_blocks:
        print("提取的代码块:", code_blocks)
        #shell_commands = next(block[1] for block in code_blocks if block[0] == 'sh' or block[0] == 'bash' or block[0] == 'shell')
        #print(shell_commands)
        python_code = next(block[1] for block in code_blocks if block[0] == 'python')

    print('预测的 Python 代码: ', python_code)

    local_executor = LocalCommandLineCodeExecutor(work_dir=work_dir)
    result = await local_executor.execute_code_blocks(
            code_blocks=[
                CodeBlock(language="python", code=python_code),
            ],
            cancellation_token=CancellationToken(),
        )
    #print(result)
    print('预测: ', result.output.strip())

• 现在运行主函数

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

• 运行代码并检查输出
• 编码文件夹中应该有 stocks.csv 和 stockspred.csv
• 编码文件夹是存储所有代码和工件的地方。

原始文章 — Samples2025/AgentsAI/codeagent.md at main · balakreshnan/Samples2025