2025颠覆认知的AI工具！小白也能用AutoGen零代码搭建“公司研究分析师”！

🤯还在为繁琐的财务分析头疼？还在苦苦搜集各种公司数据？今天，给你带来一个颠覆认知的AI神器——AutoGen！它能让你在短短几分钟内，零代码搭建一个专属的“公司研究分析师”智能体，自动完成公司财务报告，效率提升1000%！1. 传统公司研究 VS AutoGen智能体：效率天壤之别！

你是否经历过这样的场景：

*   😨手动搜集信息： 在各大网站、论坛、财报中翻找信息，耗费大量时间精力，还容易遗漏关键数据。
*   😨数据处理困难： 面对海量数据，Excel技能不够用，数据清洗、整理、分析更是让人抓狂。
*   😨报告撰写头疼： 财务分析报告专业性强，格式要求高，熬夜加班是常态。

现在，有了AutoGen，这一切都将成为过去！

2. 什么是AutoGen？为什么它能火爆全网？

AutoGen，简单来说，就是一个能让你“零代码”创建AI智能体（Agent）的超级工具！它就像一个“乐高积木”，你可以通过简单的配置，将不同的AI模型（如GPT-4、Gemini等）和工具（如搜索引擎、数据分析工具等）组合起来，创造出各种各样的智能体，帮你自动完成任务！

AutoGen之所以能火爆全网，因为它：

*   🔥零门槛： 不需要任何编程基础，小白也能轻松上手！
*   🔥高效： 自动执行任务，省时省力，效率提升1000%！
*   🔥强大： 可以连接各种AI模型和工具，功能无限扩展！
*   🔥灵活： 可以根据自己的需求定制智能体，满足个性化需求！

3. 手把手教你用AutoGen搭建“公司研究分析师”！

今天，我们就来教你如何用AutoGen搭建一个“公司研究分析师”智能体，自动完成公司财务报告！

核心代码揭秘（别怕，我们会一步步解释）：

from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelFamily
from autogen_core.models import UserMessage
import os
from dotenv import load_dotenv


load_dotenv()
import asyncio # 导入 asyncio 库
# 从环境变量中读取api_key


api_key1 = os.getenv('GEMINI_API_KEY')
openai_client = OpenAIChatCompletionClient(
    model="gemini-2.0-pro-exp-02-05",
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
    api_key=api_key1,
    model_info={
        "vision": False,
        "function_calling": True,
        "json_output": False,
        "family": ModelFamily.R1,
    },
)
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.ui import Console
from autogen_core.tools import FunctionTool
from autogen_ext.models.openai import OpenAIChatCompletionClient
#!pip install yfinance matplotlib pytz numpy pandas python-dotenv requests bs4


def google_search(query: str, num_results: int = 2, max_chars: int = 500) -> list:  # type: ignore[type-arg]
    import os
    import time

    import requests
    from bs4 import BeautifulSoup
    from dotenv import load_dotenv

    load_dotenv()

    api_key = os.getenv("GOOGLE_API_KEY")
    search_engine_id = os.getenv("SEARCH_ENGINE_ID")

    if not api_key or not search_engine_id:
        raise ValueError("API key or Search Engine ID not found in environment variables")

    url = "https://customsearch.googleapis.com/customsearch/v1"
    params = {"key": str(api_key), "cx": str(search_engine_id), "q": str(query), "num": str(num_results)}

    response = requests.get(url, params=params)

    if response.status_code != 200:
        print(response.json())
        raise Exception(f"Error in API request: {response.status_code}")

    results = response.json().get("items", [])

    def get_page_content(url: str) -> str:
        try:
            response = requests.get(url, timeout=10)
            soup = BeautifulSoup(response.content, "html.parser")
            text = soup.get_text(separator=" ", strip=True)
            words = text.split()
            content = ""
            for word in words:
                if len(content) + len(word) + 1 > max_chars:
                    break
                content += " " + word
            return content.strip()
        except Exception as e:
            print(f"Error fetching {url}: {str(e)}")
            return ""

    enriched_results = []
    for item in results:
        body = get_page_content(item["link"])
        enriched_results.append(
            {"title": item["title"], "link": item["link"], "snippet": item["snippet"], "body": body}
        )
        time.sleep(1)  # Be respectful to the servers

    return enriched_results


def analyze_stock(ticker: str) -> dict:  # type: ignore[type-arg]
    import os
    from datetime import datetime, timedelta
    import matplotlib
    matplotlib.use('agg')
    import matplotlib.pyplot as plt
    import numpy as np
    import pandas as pd
    import yfinance as yf
    from pytz import timezone  # type: ignore

    stock = yf.Ticker(ticker)

    # Get historical data (1 year of data to ensure we have enough for 200-day MA)
    end_date = datetime.now(timezone("UTC"))
    start_date = end_date - timedelta(days=365)
    hist = stock.history(start=start_date, end=end_date)

    # Ensure we have data
    if hist.empty:
        return {"error": "No historical data available for the specified ticker."}

    # Compute basic statistics and additional metrics
    current_price = stock.info.get("currentPrice", hist["Close"].iloc[-1])
    year_high = stock.info.get("fiftyTwoWeekHigh", hist["High"].max())
    year_low = stock.info.get("fiftyTwoWeekLow", hist["Low"].min())

    # Calculate 50-day and 200-day moving averages
    ma_50 = hist["Close"].rolling(window=50).mean().iloc[-1]
    ma_200 = hist["Close"].rolling(window=200).mean().iloc[-1]

    # Calculate YTD price change and percent change
    ytd_start = datetime(end_date.year, 1, 1, tzinfo=timezone("UTC"))
    ytd_data = hist.loc[ytd_start:]  # type: ignore[misc]
    if not ytd_data.empty:
        price_change = ytd_data["Close"].iloc[-1] - ytd_data["Close"].iloc[0]
        percent_change = (price_change / ytd_data["Close"].iloc[0]) * 100
    else:
        price_change = percent_change = np.nan

    # Determine trend
    if pd.notna(ma_50) and pd.notna(ma_200):
        if ma_50 > ma_200:
            trend = "Upward"
        elif ma_50 < ma_200:
            trend = "Downward"
        else:
            trend = "Neutral"
    else:
        trend = "Insufficient data for trend analysis"

    # Calculate volatility (standard deviation of daily returns)
    daily_returns = hist["Close"].pct_change().dropna()
    volatility = daily_returns.std() * np.sqrt(252)  # Annualized volatility

    # Create result dictionary
    result = {
        "ticker": ticker,
        "current_price": current_price,
        "52_week_high": year_high,
        "52_week_low": year_low,
        "50_day_ma": ma_50,
        "200_day_ma": ma_200,
        "ytd_price_change": price_change,
        "ytd_percent_change": percent_change,
        "trend": trend,
        "volatility": volatility,
    }

    # Convert numpy types to Python native types for better JSON serialization
    for key, value in result.items():
        if isinstance(value, np.generic):
            result[key] = value.item()

    # Generate plot
    plt.figure(figsize=(12, 6))
    plt.plot(hist.index, hist["Close"], label="Close Price")
    plt.plot(hist.index, hist["Close"].rolling(window=50).mean(), label="50-day MA")
    plt.plot(hist.index, hist["Close"].rolling(window=200).mean(), label="200-day MA")
    plt.title(f"{ticker} Stock Price (Past Year)")
    plt.xlabel("Date")
    plt.ylabel("Price ($)")
    plt.legend()
    plt.grid(True)

    # Save plot to file
    os.makedirs("coding", exist_ok=True)
    plot_file_path = f"coding/{ticker}_stockprice.png"
    plt.savefig(plot_file_path)
    print(f"Plot saved as {plot_file_path}")
    result["plot_file_path"] = plot_file_path

    return result

async def main(): # 定义一个异步函数 main
    google_search_tool = FunctionTool(
        google_search, description="Search Google for information, returns results with a snippet and body content"
    )
    stock_analysis_tool = FunctionTool(analyze_stock, description="Analyze stock data and generate a plot")
    search_agent = AssistantAgent(
        name="Google_Search_Agent",
        model_client=openai_client,
        tools=[google_search_tool],
        description="Search Google for information, returns top 2 results with a snippet and body content",
        system_message="You are a helpful AI assistant. Solve tasks using your tools.",
    )

    stock_analysis_agent = AssistantAgent(
        name="Stock_Analysis_Agent",
        model_client=openai_client,
        tools=[stock_analysis_tool],
        description="Analyze stock data and generate a plot",
        system_message="Perform data analysis.",
    )

    report_agent = AssistantAgent(
        name="Report_Agent",
        model_client=openai_client,
        description="Generate a report based the search and results of stock analysis",
        system_message="You are a helpful assistant that can generate a comprehensive report on a given topic based on search and stock analysis. When you done with generating the report, reply with TERMINATE.",
    )
    team = RoundRobinGroupChat([stock_analysis_agent, search_agent, report_agent], max_turns=6)
    try:
        stream = team.run_stream(task="使用中文撰写美国航空公司的财务报告")
        await Console(stream)
    except Exception as e:
        print(f"发生错误: {e}")
        print("请检查网络连接、API 密钥以及 One API 服务是否正常运行。")

asyncio.run(main()) # 使用 asyncio.run() 运行 main 函数

步骤拆解：

1.  🔑配置环境：
    *   安装必要的库（代码中有`#!pip install ...`，跟着做就行）。
    *   获取API密钥（如Google Gemini API密钥，代码中已给出获取方式）。

2.  🛠️创建工具：
    *   `google_search`函数：用于从Google搜索信息。
    *   `analyze_stock`函数：用于分析股票数据并生成图表。

3.  🤖创建智能体：
    *   `search_agent`：负责搜索信息的智能体。
    *   `stock_analysis_agent`：负责分析股票数据的智能体。
    *   `report_agent`：负责生成报告的智能体。

4.  🤝组建团队：
    *   `team`：将三个智能体组成一个团队，让它们协同工作。

5.  🚀运行任务：
    *   `team.run_stream(task="使用中文撰写美国航空公司的财务报告")`：启动任务，让智能体团队自动完成财务报告！

4. 见证奇迹：AutoGen自动生成的财务报告！

运行代码后，AutoGen会自动完成以下任务：

*   🔍搜索信息： 搜索美国航空公司的相关新闻、财报等信息。
*   📈分析数据： 分析美国航空公司的股票数据，计算各种指标，并生成图表。
*   ✍️撰写报告： 将搜索到的信息和分析结果整合成一份完整的财务报告！

 

5. 不止于此：AutoGen的无限可能！

这只是AutoGen的冰山一角！你还可以用它来创建：

*   📧自动回复邮件的智能助理
*   🛍️帮你比价的购物助手
*   📝自动写代码的编程助手
*   ...以及更多你能想到的任何智能体！

6. 立即行动：开启你的AutoGen之旅！

*   👉关注我，获取更多AutoGen教程和案例！
*   👉评论区留言，分享你想用AutoGen做什么！
*   👉点赞、收藏、分享，让更多人了解AutoGen的强大！⚠️风险提示：

*   本文仅为技术分享，不构成任何投资建议。
*   请确保你的API密钥安全，避免泄露。
*   在使用AutoGen时，请遵守相关法律法规和平台规定。

🔥彩蛋：

*   在评论区留言“AutoGen”，即可获得本文完整代码！
*   关注我，并在下一篇文章中查看如何用AutoGen创建更复杂的智能体！
*   分享本文到朋友圈，并截图发送给我，即可获得AutoGen学习资源大礼包！