大家好，我是专注于技术实战分享的博主。在探索AI与金融科技结合的领域时，我们常常面临一个挑战：如何将前沿的大语言模型（LLM）能力，系统性地应用于专业的投资研究场景？手动拼接提示词、切换不同工具不仅效率低下，也难以形成可复用、可协作的分析流程。今天，我将为大家深度解析一个名为 ai-berkshire 的开源项目，它正是为解决这一问题而生。本文将带你从零开始，理解其作为“AI驱动的投资研究框架”的核心设计，并手把手教你搭建环境、运行示例，最终掌握如何定制属于你自己的多智能体（Multi-Agent）投资分析系统。无论你是对AI应用感兴趣的开发者，还是希望用技术赋能金融研究的从业者，都能从中获得一套完整的工程化解决方案。

1. 项目背景与核心概念解析

在深入代码之前，我们首先要厘清几个关键概念，理解 ai-berkshire 究竟要解决什么问题，以及它是如何构思的。

1.1 什么是 AI Agent 与多智能体协作？

AI Agent（智能体） 在此语境下，并非指某个具体的软件，而是一个设计范式。一个基本的AI智能体通常具备以下能力：

  1. 感知（Perception） ：能理解用户的指令（自然语言）和所处的环境（如访问的网页、读取的文件）。
  2. 规划（Planning） ：能将复杂任务分解为一系列可执行的子步骤。
  3. 行动（Action） ：可以调用工具（Tools）来执行具体操作，如搜索网络、执行代码、读写文件。
  4. 反思（Reflection） ：能评估行动结果，并据此调整后续策略。

而 多智能体协作（Multi-Agent Collaboration） 则是指多个具备不同专长的AI智能体，通过通信与协作，共同完成一个更为复杂的任务。这模拟了人类团队的工作模式，例如在一个投资研究团队中，可能有“行业分析师”、“财务专家”、“数据挖掘员”和“报告撰写员”等角色。

ai-berkshire 项目正是基于多智能体协作的思想，构建了一个专门用于投资研究的框架。

1.2 ai-berkshire 项目定位与价值

根据项目名称和其设计思路， ai-berkshire 旨在构建一个 模块化、可扩展的AI驱动投资研究框架 。其核心价值在于：

  • 流程标准化 ：将松散、临时的投资研究动作（如搜集信息、分析财报、评估风险、生成报告）固化为一套可重复执行的自动化或半自动化流程。
  • 能力专业化 ：通过设计不同的智能体角色，让每个“AI研究员”专注于其最擅长的领域，提升单项任务的分析深度和准确性。
  • 协作自动化 ：智能体之间可以自动传递任务、交换信息和初步结论，减少人工干预，提高研究效率。
  • 知识沉淀 ：整个研究过程产生的中间数据、分析逻辑和最终结论可以被系统化地记录和管理，形成可查询、可复用的知识库。

它不是一个提供“投资建议”的黑盒系统，而是一个 增强研究能力的工具平台 ，研究者（用户）是流程的设计者和决策的最终把控者。

1.3 核心组件与技术栈推测

虽然未直接看到源码，但结合当前AI开源生态，我们可以合理推测 ai-berkshire 可能涉及的技术栈：

  • 智能体框架 ：很可能基于 LangChain、LlamaIndex 或 AutoGen 等主流AI应用开发框架构建，这些框架提供了智能体、工具链、记忆等核心抽象。
  • 大语言模型（LLM） ：作为智能体的“大脑”，可能集成 OpenAI GPT、Claude、或开源的 Llama、DeepSeek 等模型。项目可能需要配置API Key或本地模型路径。
  • 工具集（Tools） ：智能体可以调用的外部能力，例如：
    • 网络搜索（SerpAPI， DuckDuckGo）
    • 金融数据API（Yahoo Finance， Alpha Vantage， Tushare等）
    • 文档处理（读取PDF、Excel财报）
    • 代码执行（Python， 用于数据计算）
  • 编排与通信 ：负责管理多个智能体的工作流，定义它们之间的触发条件和信息交换格式。这可能通过预定义的流程脚本或动态的智能体对话来实现。
  • 前端/接口 ：可能是命令行界面（CLI）、Web界面或API服务，用于用户启动任务和查看结果。

接下来，我们将基于这些理解，进入实战环节。

2. 环境准备与项目初始化

在开始搭建之前，请确保你的开发环境满足基本要求。我们将以最常见的Python环境为例进行说明。

2.1 基础环境要求

  • 操作系统 ：Windows 10/11， macOS 或 Linux（如Ubuntu）均可。本文示例将在 macOS/Linux 环境下演示，Windows 用户请注意命令的细微差别（如使用 dir 代替 ls ）。
  • Python ：版本 3.8 或更高。这是运行绝大多数AI框架的硬性要求。
  • 包管理工具 ： pip （Python自带）或 conda （如果你使用Anaconda）。
  • 代码编辑器/IDE ：VS Code（推荐，有丰富的Python和AI插件）、PyCharm 或任何你熟悉的编辑器。
  • 版本控制 ：Git（用于克隆项目）。

首先，检查你的Python环境：

python --version
# 或
python3 --version

如果版本低于3.8，请先升级Python。

2.2 获取项目代码

由于 ai-berkshire 是一个GitHub项目（从名称 xbtlin / ai-berkshire 推断），我们首先需要将其克隆到本地。

打开终端（命令行），切换到你希望存放项目的目录，然后执行：

git clone https://github.com/xbtlin/ai-berkshire.git
cd ai-berkshire

如果 git clone 失败，可能是项目地址有误或仓库已设为私有。请尝试在GitHub上搜索 xbtlin/ai-berkshire 确认其存在性和访问权限。作为替代，你也可以下载项目的ZIP包并解压。

进入项目目录后，首先查看项目结构，这能帮助我们理解其组织方式。

ls -la
# 或查看README.md
cat README.md

2.3 安装项目依赖

一个规范的Python项目通常会通过 requirements.txt 或 pyproject.toml 文件来管理依赖。我们优先寻找这些文件。

# 查看是否有 requirements.txt
ls requirements.txt
# 如果有，使用pip安装
pip install -r requirements.txt

# 或者查看是否有 pyproject.toml，并使用pip安装
pip install -e .

重要提示 ：在安装依赖前，强烈建议创建一个独立的Python虚拟环境，以避免与系统或其他项目的包发生冲突。

# 创建虚拟环境（以venv为例）
python -m venv venv

# 激活虚拟环境
# 在 macOS/Linux 上：
source venv/bin/activate
# 在 Windows 上：
venv\Scripts\activate

# 激活后，命令行提示符前通常会显示 (venv)
# 然后在虚拟环境中安装依赖
pip install -r requirements.txt

如果项目没有提供依赖文件，或者安装过程中出现错误，我们需要根据项目可能用到的技术栈手动安装核心包。一个典型的组合可能包括：

pip install langchain langchain-community langchain-openai
pip install duckduckgo-search  # 用于网络搜索
pip install pandas yfinance  # 用于数据处理和金融数据
pip install jupyter  # 可能用于示例笔记本
pip install python-dotenv  # 用于管理环境变量（如API密钥）

安装过程可能会因操作系统和Python版本略有不同，请根据终端提示解决可能出现的编译或依赖问题。

2.4 配置API密钥与环境变量

AI项目通常需要接入外部服务，如OpenAI、SerpAPI（谷歌搜索）或金融数据API。这些服务的密钥不应硬编码在代码中，最佳实践是使用环境变量管理。

在项目根目录下，寻找或创建一个名为 .env 的文件。

touch .env

用文本编辑器打开 .env 文件，填入你的API密钥。格式如下：

# .env 文件示例
OPENAI_API_KEY=sk-your-openai-api-key-here
SERPAPI_API_KEY=your-serpapi-api-key-here
ALPHA_VANTAGE_API_KEY=your-alpha-vantage-key-here
# 其他可能的配置，如模型名称、温度等
MODEL_NAME=gpt-4-turbo-preview
TEMPERATURE=0.1

安全警告 ：务必确保 .env 文件被添加到 .gitignore 中，防止将密钥意外提交到公开仓库。

在Python代码中，使用 python-dotenv 来加载这些变量：

# config.py 或主程序开头
from dotenv import load_dotenv
import os

load_dotenv()  # 加载 .env 文件中的变量

openai_api_key = os.getenv("OPENAI_API_KEY")
if not openai_api_key:
    raise ValueError("请在 .env 文件中设置 OPENAI_API_KEY")

3. 核心架构与代码拆解

现在，让我们深入项目内部，理解其智能体系统的设计。由于无法看到确切源码，我将基于多智能体投资研究的通用模式，构建一个简化的、可运行的示例框架。你可以将此视为对 ai-berkshire 核心思想的实现，并据此理解你实际项目中的代码。

3.1 智能体角色定义

一个典型的投资研究多智能体系统可能包含以下角色：

  1. Coordinator（协调员） ：接收用户查询，分解任务，分配子任务给其他智能体，并汇总最终结果。
  2. WebResearcher（网络研究员） ：擅长使用搜索工具，获取公司新闻、行业动态、市场情绪等最新信息。
  3. FinancialAnalyst（财务分析师） ：擅长读取和处理财务报表（如PDF、Excel），计算财务比率，进行基本面分析。
  4. DataAnalyst（数据分析师） ：擅长获取和处理金融市场时间序列数据，进行简单的技术分析或数据可视化。
  5. Reporter（报告撰写员） ：负责整合所有分析结果，生成结构清晰、语言流畅的中文或英文投资研究报告。

3.2 基于 LangChain 的智能体实现示例

我们将使用 LangChain 框架来构建这些智能体。首先确保已安装 langchain 和 langchain-openai 。

# agent_definitions.py
import os
from dotenv import load_dotenv
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.utilities import FinancialDatasets # 示例，实际可能需要其他库
import yfinance as yf
import pandas as pd

load_dotenv()

# 初始化LLM
llm = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4-turbo-preview"),
    temperature=float(os.getenv("TEMPERATURE", 0.1)),
    api_key=os.getenv("OPENAI_API_KEY")
)

# 定义工具
# 工具1：网络搜索
search_tool = DuckDuckGoSearchRun(name="web_search", description="当需要搜索最新的公司新闻、行业报告或宏观信息时使用此工具。")

# 工具2：获取股票基础信息（示例，使用yfinance）
def get_stock_info(symbol: str) -> str:
    """获取指定股票代码的基础信息，如公司名称、行业、当前价格等。"""
    try:
        ticker = yf.Ticker(symbol)
        info = ticker.info
        # 提取关键信息
        result = {
            "name": info.get('longName', 'N/A'),
            "sector": info.get('sector', 'N/A'),
            "industry": info.get('industry', 'N/A'),
            "currentPrice": info.get('currentPrice', 'N/A'),
            "marketCap": info.get('marketCap', 'N/A'),
            "summary": info.get('longBusinessSummary', 'N/A')[:500] + "..." # 截取部分
        }
        return str(result)
    except Exception as e:
        return f"获取 {symbol} 信息时出错：{e}"

stock_info_tool = Tool(
    name="get_stock_basic_info",
    func=get_stock_info,
    description="输入股票代码（如AAPL, 000001.SZ），获取该公司的基础信息、行业分类和当前市值等。"
)

# 工具3：获取历史股价（示例）
def get_historical_price(symbol: str, period: str = "1mo") -> str:
    """获取股票历史价格数据。period可以是‘1d’， ‘5d’， ‘1mo’， ‘3mo’， ‘6mo’， ‘1y’， ‘5y’， ‘max’"""
    try:
        ticker = yf.Ticker(symbol)
        hist = ticker.history(period=period)
        if hist.empty:
            return f"未找到 {symbol} 在 {period} 内的历史数据。"
        # 返回一个简化的统计摘要
        summary = {
            "period": period,
            "data_points": len(hist),
            "start_date": str(hist.index[0].date()),
            "end_date": str(hist.index[-1].date()),
            "price_range": f"{hist['Low'].min():.2f} - {hist['High'].max():.2f}",
            "last_close": hist['Close'].iloc[-1]
        }
        return str(summary)
    except Exception as e:
        return f"获取 {symbol} 历史价格时出错：{e}"

historical_price_tool = Tool(
    name="get_historical_price",
    func=get_historical_price,
    description="输入股票代码和时间周期（如‘1mo’代表一个月），获取该时间段内的历史价格摘要。"
)

# 为不同角色的智能体组合不同的工具集
web_researcher_tools = [search_tool, stock_info_tool]
financial_analyst_tools = [stock_info_tool, historical_price_tool] # 实际应有财报解析工具
data_analyst_tools = [historical_price_tool] # 实际应有更复杂的数据分析工具

# 定义智能体提示词模板
def create_agent_prompt(role: str, instructions: str):
    """为不同角色创建提示词"""
    return ChatPromptTemplate.from_messages([
        ("system", f"""你是一个专业的{role}。
        你的职责是：{instructions}
        请严格根据你掌握的工具和知识来回答问题。如果你无法通过工具获得信息，请如实说明。
        你的思考过程应该清晰。"""),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])

# 创建智能体执行器
def create_agent_executor(tools, prompt):
    memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True, handle_parsing_errors=True)
    return executor

# 实例化不同角色的智能体
web_researcher_prompt = create_agent_prompt(
    "投资研究网络信息搜集员",
    "利用搜索引擎和金融数据库，查找关于公司、行业、竞争对手的最新动态、新闻和公开信息。"
)
web_researcher = create_agent_executor(web_researcher_tools, web_researcher_prompt)

data_analyst_prompt = create_agent_prompt(
    "金融市场数据分析师",
    "处理历史股价、交易量等市场数据，提供数据摘要、基本统计和趋势观察。"
)
data_analyst = create_agent_executor(data_analyst_tools, data_analyst_prompt)

# 协调员智能体（工具是调用其他智能体，这里简化处理）
# 在实际的ai-berkshire项目中，这里可能会有更复杂的编排逻辑（如使用LangGraph或AutoGen的GroupChat）。

3.3 智能体协作流程设计

多智能体协作的核心是“编排”。一个简单的编排模式可以是顺序链式调用：

# orchestration.py
from agent_definitions import web_researcher, data_analyst, llm
from langchain_core.messages import HumanMessage, SystemMessage

class SimpleResearchOrchestrator:
    """一个简单的顺序研究编排器"""
    
    def research_company(self, company_name: str, stock_symbol: str):
        """对一个公司进行初步研究"""
        print(f"开始对 {company_name} ({stock_symbol}) 进行研究...\n")
        
        # 阶段1：网络研究员搜集信息
        print("=== 阶段1：网络信息搜集 ===")
        web_task = f"请搜索关于{company_name}的最新新闻、近期重大事件以及其主要竞争对手。"
        web_result = web_researcher.invoke({"input": web_task})
        web_findings = web_result['output']
        print(f"网络研究员发现：{web_findings[:300]}...\n") # 打印部分结果
        
        # 阶段2：数据分析师处理市场数据
        print("=== 阶段2：市场数据分析 ===")
        data_task = f"获取{stock_symbol}过去一年的历史价格数据摘要。"
        data_result = data_analyst.invoke({"input": data_task})
        data_findings = data_result['output']
        print(f"数据分析师发现：{data_findings}\n")
        
        # 阶段3：协调员（这里用LLM直接模拟）整合报告
        print("=== 阶段3：生成初步研究摘要 ===")
        report_prompt = [
            SystemMessage(content="你是一名高级投资研究协调员。请根据以下两位分析员的研究发现，生成一份简短、结构化的初步研究摘要。摘要应包括：公司近期动态、市场表现概览、以及初步的关注点或风险提示。"),
            HumanMessage(content=f"""
            公司：{company_name} ({stock_symbol})
            
            网络研究员发现：
            {web_findings}
            
            数据分析师发现：
            {data_findings}
            
            请基于以上信息生成一份投资研究摘要。
            """)
        ]
        final_report = llm.invoke(report_prompt)
        print("=== 初步研究摘要 ===")
        print(final_report.content)
        
        return {
            "web_research": web_findings,
            "data_analysis": data_findings,
            "summary": final_report.content
        }

# 使用示例
if __name__ == "__main__":
    orchestrator = SimpleResearchOrchestrator()
    result = orchestrator.research_company("苹果公司", "AAPL")
    # 可以将result保存为JSON文件
    import json
    with open("research_output_AAPL.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    print("\n研究结果已保存至 research_output_AAPL.json")

这个简化的例子展示了从任务分解、智能体执行到结果整合的基本流程。真实的 ai-berkshire 项目会有更健壮的架构，例如使用 LangGraph 来定义有状态的、带循环的工作流，或者使用 AutoGen 的 GroupChat 来实现智能体间的自由对话与协作。

4. 完整实战：运行你的第一个多智能体研究任务

让我们将上面的代码片段整合，创建一个最小可运行的研究项目。

4.1 项目结构创建

在你的工作区（可以是 ai-berkshire 项目目录内，或新建一个目录）创建如下结构：

my_ai_researcher/
├── .env                    # 你的API密钥（不要提交）
├── .gitignore             # 忽略.env等文件
├── requirements.txt       # 依赖列表
├── config.py              # 配置加载
├── tools.py               # 自定义工具定义
├── agents.py              # 智能体定义
├── orchestration.py       # 编排逻辑
└── main.py                # 主程序入口

4.2 编写核心文件

1. requirements.txt 

langchain==0.1.0
langchain-community==0.0.10
langchain-openai==0.0.5
openai==1.6.1
python-dotenv==1.0.0
duckduckgo-search==4.1.1
yfinance==0.2.33
pandas==2.1.4

2. config.py 

import os
from dotenv import load_dotenv

load_dotenv()

def get_config():
    config = {
        "openai_api_key": os.getenv("OPENAI_API_KEY"),
        "model_name": os.getenv("MODEL_NAME", "gpt-4-turbo-preview"),
        "temperature": float(os.getenv("TEMPERATURE", 0.1)),
    }
    # 检查必要配置
    if not config["openai_api_key"]:
        raise ValueError("致命错误：未在 .env 文件中找到 OPENAI_API_KEY。请先配置。")
    return config

3. tools.py 

# 这里集成上一节定义的三个工具：search_tool, stock_info_tool, historical_price_tool
from langchain_community.tools import DuckDuckGoSearchRun
from langchain.tools import Tool
import yfinance as yf

def create_tools():
    """创建并返回工具列表"""
    search_tool = DuckDuckGoSearchRun(name="web_search", 
                                      description="搜索互联网获取最新信息。")
    
    def get_stock_info(symbol: str) -> str:
        try:
            ticker = yf.Ticker(symbol)
            info = ticker.info
            result = {
                "name": info.get('longName', 'N/A'),
                "sector": info.get('sector', 'N/A'),
                "currentPrice": info.get('currentPrice', 'N/A'),
                "marketCap": info.get('marketCap', 'N/A'),
                "summary": info.get('longBusinessSummary', 'N/A')[:300]
            }
            return str(result)
        except Exception as e:
            return f"Error: {e}"
    
    stock_info_tool = Tool(
        name="get_stock_info",
        func=get_stock_info,
        description="输入股票代码（如AAPL），获取公司基本信息。"
    )
    
    def get_historical_price(symbol: str, period: str = "1y") -> str:
        try:
            ticker = yf.Ticker(symbol)
            hist = ticker.history(period=period)
            if hist.empty:
                return "No data."
            summary = {
                "period": period,
                "data_points": len(hist),
                "start": str(hist.index[0].date()),
                "end": str(hist.index[-1].date()),
                "close_price_latest": round(hist['Close'].iloc[-1], 2)
            }
            return str(summary)
        except Exception as e:
            return f"Error: {e}"
    
    historical_price_tool = Tool(
        name="get_historical_price",
        func=get_historical_price,
        description="输入股票代码和周期（如‘1y’），获取历史价格摘要。"
    )
    
    return [search_tool, stock_info_tool, historical_price_tool]

4. agents.py 

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from config import get_config
from tools import create_tools

config = get_config()
llm = ChatOpenAI(
    model=config["model_name"],
    temperature=config["temperature"],
    api_key=config["openai_api_key"]
)

def create_agent(role_description, tools):
    """通用智能体创建函数"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""你是一个AI助手，扮演以下角色：{role_description}。请使用提供的工具完成任务。如果工具无法获得信息，请说明。"""),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])
    memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True, handle_parsing_errors=True)
    return executor

def get_web_researcher():
    tools = create_tools()[:2]  # 使用搜索和股票信息工具
    return create_agent("网络信息研究员，负责搜索公司新闻和基础信息。", tools)

def get_data_analyst():
    tools = [create_tools()[2]]  # 仅使用历史价格工具
    return create_agent("金融市场数据分析师，负责处理股价数据。", tools)

5. orchestration.py 

from agents import get_web_researcher, get_data_analyst, llm
from langchain_core.messages import HumanMessage, SystemMessage
import json

class ResearchOrchestrator:
    def __init__(self):
        self.web_agent = get_web_researcher()
        self.data_agent = get_data_analyst()
    
    def run_research(self, company: str, symbol: str):
        print(f"\n{'='*50}")
        print(f"启动对 {company} ({symbol}) 的研究流程")
        print(f"{'='*50}\n")
        
        findings = {}
        
        # 任务1：网络研究
        print("[任务1] 网络信息搜集...")
        web_task = f"搜索关于{company}的最新新闻和重大事件。"
        try:
            web_result = self.web_agent.invoke({"input": web_task})
            web_output = web_result['output']
            findings['web_research'] = web_output
            print(f"网络研究完成。摘要：{web_output[:200]}...\n")
        except Exception as e:
            print(f"网络研究任务出错：{e}")
            findings['web_research'] = f"Error: {e}"
        
        # 任务2：数据分析
        print("[任务2] 市场数据分析...")
        data_task = f"获取{symbol}过去半年的股价数据摘要。"
        try:
            data_result = self.data_agent.invoke({"input": data_task})
            data_output = data_result['output']
            findings['data_analysis'] = data_output
            print(f"数据分析完成。结果：{data_output}\n")
        except Exception as e:
            print(f"数据分析任务出错：{e}")
            findings['data_analysis'] = f"Error: {e}"
        
        # 任务3：生成报告
        print("[任务3] 生成综合摘要报告...")
        report_prompt = [
            SystemMessage(content="你是一名投资研究助理。请基于以下信息，生成一段简洁、客观的综合性摘要，涵盖公司近况和市场表现。"),
            HumanMessage(content=f"""
            公司：{company} ({symbol})
            
            网络信息：
            {findings.get('web_research', '无')}
            
            市场数据：
            {findings.get('data_analysis', '无')}
            
            请生成一份摘要。
            """)
        ]
        try:
            final_report = llm.invoke(report_prompt)
            findings['summary'] = final_report.content
            print(f"报告生成完成。\n")
            print(f"{'='*50}")
            print("最终摘要：")
            print(final_report.content)
            print(f"{'='*50}")
        except Exception as e:
            print(f"报告生成出错：{e}")
            findings['summary'] = f"Error: {e}"
        
        return findings

6. main.py 

from orchestration import ResearchOrchestrator
import sys

def main():
    if len(sys.argv) < 3:
        print("用法：python main.py <公司名称> <股票代码>")
        print("示例：python main.py \"苹果公司\" AAPL")
        print("示例：python main.py \"腾讯控股\" 0700.HK")
        sys.exit(1)
    
    company_name = sys.argv[1]
    stock_symbol = sys.argv[2]
    
    orchestrator = ResearchOrchestrator()
    results = orchestrator.run_research(company_name, stock_symbol)
    
    # 保存结果到文件
    filename = f"research_{stock_symbol}.json"
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    print(f"\n完整研究数据已保存至：{filename}")

if __name__ == "__main__":
    main()

4.3 运行与验证

  1. 安装依赖 ：在项目根目录 my_ai_researcher 下，确保虚拟环境已激活，运行：

    pip install -r requirements.txt

  2. 配置API密钥 ：在 .env 文件中正确设置你的 OPENAI_API_KEY 。

  3. 运行研究任务 ：在命令行中执行：

    python main.py "微软" MSFT

    你将看到类似以下的输出，智能体会依次执行任务：

    ==================================================
启动对 微软 (MSFT) 的研究流程
==================================================

[任务1] 网络信息搜集...
> Entering new AgentExecutor chain...
... (详细的思考链和工具调用过程)
> Finished chain.
网络研究完成。摘要：根据搜索，微软近期公布了...财报，云业务Azure增长...宣布与OpenAI深化合作...

[任务2] 市场数据分析...
> Entering new AgentExecutor chain...
... (调用历史价格工具)
> Finished chain.
数据分析完成。结果：{'period': '6mo', 'data_points': 126, ... 'close_price_latest': 420.55}

[任务3] 生成综合摘要报告...
报告生成完成。

==================================================
最终摘要：
根据现有信息，微软（MSFT）近期...其股价在过去半年内...综合来看...
==================================================
完整研究数据已保存至：research_MSFT.json

  4. 检查输出 ：打开生成的 research_MSFT.json 文件，可以看到结构化的JSON数据，包含了每个步骤的原始输出和最终摘要。

至此，你已经成功运行了一个简易版的多智能体投资研究流程。这模拟了 ai-berkshire 项目的核心工作模式。

5. 常见问题与排查思路

在实际搭建和运行过程中，你可能会遇到以下问题：

问题现象 可能原因 排查与解决思路
ModuleNotFoundError: No module named ‘langchain’ 依赖未正确安装或虚拟环境未激活。 1. 确认已激活虚拟环境（命令行前有 (venv) ）。
2. 在项目根目录重新运行 pip install -r requirements.txt 。
3. 尝试直接安装 pip install langchain langchain-community 。
AuthenticationError: Incorrect API key provided OpenAI API密钥错误或未设置。 1. 检查 .env 文件是否存在，且 OPENAI_API_KEY 的值正确无误（无多余空格）。
2. 在Python中打印 os.getenv(“OPENAI_API_KEY”) 前几位，确认已加载。
3. 前往OpenAI平台检查API密钥是否有效、是否有余额。
智能体长时间无响应或报超时错误 网络问题，或LLM API调用缓慢。 1. 检查网络连接。
2. 尝试设置超时参数： ChatOpenAI(..., request_timeout=60) 。
3. 对于复杂任务，考虑使用能力更强但更慢的模型（如GPT-4），或简化提示词。
yfinance 无法获取A股或某些市场数据 yfinance 库对某些交易所或股票代码支持不完整。 1. 确认股票代码格式正确（如A股需加后缀 .SS 或 .SZ ）。
2. 考虑使用其他数据源，如 akshare （国内）、 tushare 或直接调用财经网站API。
3. 在工具函数中添加更完善的错误处理和备选数据源。
搜索工具返回内容不相关或为空 DuckDuckGo搜索受地域或查询词影响。 1. 优化搜索查询词，使其更具体（例如“Apple Inc 2024 Q1 earnings report” 比 “Apple news” 更好）。
2. 考虑使用更强大的搜索API，如SerpAPI（需付费但更稳定）。
3. 在智能体提示词中要求其进行多轮、更精确的搜索。
智能体陷入循环或执行错误工具 提示词不够清晰，或工具描述不准确。 1. 优化系统提示词，明确角色职责和边界。
2. 检查工具（ Tool ）的 description 字段，确保其清晰、无歧义，能准确引导LLM选择。
3. 启用 verbose=True 观察智能体的思考链，定位问题步骤。
生成的研究报告质量不高 整合信息的提示词过于简单，或上游智能体提供的信息质量差。 1. 为“报告员”角色设计更详细的提示词模板，要求结构化输出（如分点：公司近况、财务亮点、市场表现、潜在风险）。
2. 提升上游智能体（研究员、分析师）的工具能力，例如接入更专业的金融数据库或财报解析工具。
3. 引入“反思”或“验证”步骤，让一个智能体检查另一个智能体的输出。

6. 最佳实践与进阶优化建议

当你掌握了基础搭建后，可以参考以下建议来提升你的 ai-berkshire 风格项目的鲁棒性和实用性。

6.1 工程化与可维护性

  • 配置中心化 ：将所有配置（模型参数、API端点、工具开关）集中管理，例如使用 config.yaml 或更高级的配置库（如 hydra ），便于不同环境（开发、测试、生产）切换。
  • 日志与监控 ：为每个智能体的调用、工具的执行结果添加详细日志。这不仅是调试的需要，也能用于后续分析智能体的行为模式和效果。可以考虑使用 structlog 或 loguru 。
  • 错误处理与重试 ：网络请求和API调用必然失败。在所有外部调用（LLM、搜索、数据API）周围添加重试逻辑和优雅降级方案。例如，当主要数据源失败时，自动切换到备用源。
  • 版本控制与实验管理 ：使用 DVC （Data Version Control）或 MLflow 来跟踪不同的提示词版本、工具组合和模型参数对最终输出质量的影响。

6.2 智能体设计与协作优化

  • 角色专业化与工具精细化 ：不要给一个智能体太多不相关的工具。更精细的角色划分能带来更好的效果。例如，将“财务分析师”细分为“财报解析员”（专读PDF）和“比率计算员”（专做计算）。
  • 引入工作流引擎 ：对于复杂的多步骤研究，使用 LangGraph 来定义有向无环图（DAG）或带条件分支的工作流。这比简单的顺序调用更强大、更灵活。
  • 实现智能体间通信 ：探索让智能体之间直接对话，例如使用 AutoGen 的 GroupChat 。协调员可以主持讨论，让研究员和分析师互相提问、补充信息，更接近人类团队的头脑风暴。
  • 增加记忆与知识库 ：为智能体配备向量数据库（如 Chroma , Weaviate ），使其能记住历史对话，并能从内部知识库（如过往的研究报告、公司档案）中检索相关信息，实现持续学习和上下文感知。

6.3 数据质量与领域深化

  • 接入专业数据源 ： yfinance 和通用搜索只能提供基础信息。要做出有深度的研究，需要接入：
    • 金融数据API ：如 Alpha Vantage , Quandl , Tushare （国内），获取更丰富的历史数据和基本面指标。
    • 财报解析 ：使用 camelot 或 pdfplumber 解析上市公司PDF财报，并配合LLM提取关键表格和数字。
    • 另类数据 ：新闻情绪分析（如 TextBlob , VADER ）、社交媒体舆情、供应链数据等。
  • 构建评估体系 ：如何判断AI生成的研究报告质量？需要建立一套评估标准，可以是人工评分，也可以设计自动化指标（如信息完整性、数据引用准确性、逻辑连贯性）。这是迭代优化系统的关键。

6.4 安全与合规考量

  • 数据隐私 ：如果处理非公开或敏感信息，确保数据在传输和存储过程中加密，并遵守相关法律法规（如GDPR）。
  • 内容审核 ：对AI生成的研究摘要，尤其是涉及市场预测或建议时，应添加明确的免责声明，并考虑引入人工审核环节，避免产生误导性内容。
  • API成本控制 ：LLM和某些数据API调用成本不菲。实施用量监控、设置预算警报、对缓存结果进行复用，以避免意外的高额账单。

通过以上步骤，你不仅能够复现和运行类似 ai-berkshire 的项目，更能理解其背后的设计哲学，并具备能力根据自身的研究需求，定制和扩展一个更强大、更专业的AI辅助投资研究系统。从简单的信息聚合到深度的基本面分析，其演进路径清晰，充满了工程与AI结合的乐趣与挑战。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

agent面试必备40-AI Agent 核心架构：会话摘要与压缩（Memory Compression）

avatar AI Agent技术社区

SERP API vs Web Search API for AI Agents:选哪个?

背景做 AI agent 时,需要"实时搜索"能力。我做了 4 周实测,对比 5 维度,给 AI agent 选型参考。

avatar AI Agent技术社区
cover

鸿蒙HarmonyOS 多模态输入实战 —— 图片与文档输入智谱/DeepSeek

avatar AI Agent技术社区