在实际 AI 应用开发中，构建一个能够执行复杂、多步骤任务的智能体（Agent）系统，正从概念验证走向工程实践。无论是自动化办公、数据分析，还是像“价值投资研究”这类需要深度信息处理与决策的领域，单一 AI 模型的能力往往捉襟见肘。这时，多智能体（Multi-Agent）协作框架的价值就凸显出来——它允许我们将一个大问题分解，由多个具备不同“技能”的智能体分工协作，共同完成目标。Claude Code 作为 Anthropic 推出的 AI 编程助手，其强大的代码理解和生成能力，使其成为构建这类 AI Agent 的理想“大脑”或核心组件。

本文将以一个虚构但极具代表性的项目“AI Berkshire”（一个模拟价值投资研究的 AI 系统）为背景，带你从零开始，理解并实践如何利用 Claude Code 及相关技术栈，设计并实现一个多 Agent 协作系统。我们将重点关注如何定义 Agent 角色、设计它们之间的通信机制、集成外部工具（如网络搜索、数据获取），并最终让它们协同工作，完成一份初步的投资分析报告。整个过程将覆盖从环境搭建、核心概念理解、代码实现到运行验证和问题排查的完整链路，目标是让你不仅能跑通一个案例，更能掌握构建此类系统的通用方法论和工程细节。

1. 理解多 Agent 协作系统的核心概念与设计

在动手写代码之前，必须厘清几个关键概念，这决定了后续架构设计的合理性。多 Agent 系统不是简单地把几个 AI 调用堆在一起，而是一个有组织、有交互的微型社会。

1.1 什么是 AI Agent？

一个 AI Agent 可以理解为一个具备一定自主性的软件实体，它能够感知环境（输入），根据内部目标或指令进行推理和决策，并执行行动（输出）来影响环境。在我们的上下文中，一个 Agent 通常由以下几部分构成：

  • 核心 LLM（大语言模型） ：如 Claude Code，负责理解任务、进行推理和生成决策或内容。
  • 技能（Skills/Tools） ：Agent 可以调用的外部函数或工具，例如执行网络搜索、读取本地文件、调用计算 API、查询数据库等。这是 Agent 突破纯文本生成限制、与真实世界交互的关键。
  • 记忆（Memory） ：短期或长期的上下文存储，用于记住对话历史、任务状态或中间结果，保证协作的连贯性。
  • 规划器（Planner） ：在一些复杂架构中，会有一个专门的“规划”Agent 或模块，负责将高层目标分解为子任务，并分配给其他“执行”Agent。

1.2 多 Agent 协作的典型模式

根据任务复杂度和协作紧密程度，多 Agent 协作主要有以下几种模式：

  • 主从模式（Master-Worker） ：一个“管理者”Agent 接收用户请求，将其分解为子任务，然后协调并指挥多个“工作者”Agent 并行或串行执行，最后汇总结果。这是最常见且易于实现的模式。
  • 平等协作模式（Peer-to-Peer） ：多个 Agent 地位平等，通过约定的通信协议（如发布/订阅、消息队列）直接交换信息和任务。适合需要频繁 peer 间讨论、辩论或协商的场景。
  • 流水线模式（Pipeline） ：任务像生产线一样流经多个 Agent，每个 Agent 负责处理一个特定环节（如数据收集、清洗、分析、报告生成），并将处理后的结果传递给下一个 Agent。

对于“投资研究”这类任务，主从模式结合流水线模式通常比较合适：一个“研究主管”Agent 负责规划，指挥“信息收集”、“财务分析”、“风险评估”等专业 Agent 开展工作。

1.3 通信与协作机制

Agent 之间如何“对话”是实现协作的基础。在工程实现上，主要有以下几种方式：

  • 函数调用（Function Calling） ：LLM 生成结构化请求（如 JSON），由程序解析后调用对应的工具函数，函数执行结果再作为上下文返回给 LLM。这是当前最主流、最稳定的方式，Claude Code 对此有良好支持。
  • 共享内存/状态 ：通过一个全局的字典、数据库或消息队列，让 Agent 能够读取和写入中间状态、任务结果。这需要精心设计数据结构和并发控制。
  • 自然语言对话模拟 ：让 Agent 的输出直接作为另一个 Agent 的输入，模拟人类对话。这种方式简单但难以控制流程，容易偏离主题。

在我们的实现中，将主要采用“函数调用”作为 Agent 与工具交互的方式，并辅以一个简单的“任务状态管理器”（共享内存）来协调多个 Agent 的工作流。

2. 环境准备与核心工具链搭建

构建一个可运行的多 Agent 系统，需要选择合适的开发框架和工具。我们将基于 Python 生态，因为它拥有最丰富的 AI 和数据处理库。

2.1 基础开发环境配置

首先，确保你的本地环境满足以下要求：

组件 要求 说明
操作系统 Windows 10/11, macOS, Linux 推荐使用 macOS 或 Linux 子系统（WSL2）以获得最佳开发体验。
Python 3.9 或 3.10 避免使用 Python 3.11+ 的某些最新版本 ，部分 AI 库可能存在兼容性问题。建议使用 pyenv 或 conda 管理多版本。
包管理 pip (>=21.0) 确保 pip 已更新。
代码编辑器 VS Code 强烈推荐，因其对 Claude Code 插件支持最好。
版本控制 Git 用于代码管理和版本回溯。

在终端中验证 Python 环境：

python --version
# 应输出 Python 3.9.x 或 3.10.x
pip --version
# 确保 pip 版本较新

2.2 Claude Code 的安装与配置

Claude Code 是 Anthropic 为 VS Code 开发的 AI 编程助手插件。它并非一个独立的、可供 Python 代码调用的 SDK，而是你的“副驾驶”。在本文的语境中，我们将利用 Claude Code 来辅助我们编写、理解和调试构建多 Agent 系统的代码。它的代码生成、解释和问题排查能力，能极大提升开发效率。

  1. 安装 VS Code ：从官网下载并安装 Visual Studio Code。
  2. 安装 Claude Code 插件 ：
    • 打开 VS Code。
    • 进入扩展市场（Ctrl+Shift+X 或 Cmd+Shift+X）。
    • 搜索 “Claude Code”。
    • 点击安装。 请注意 ：由于服务可用性问题，部分地区可能无法直接安装或使用。如果遇到此情况，你需要自行解决网络连通性问题，本文无法提供相关指导。
  3. 配置与认证 ：
    • 安装后，VS Code 侧边栏会出现 Claude 图标。
    • 点击图标，通常会引导你进行登录或 API 密钥配置。你需要一个可用的 Anthropic 账户并获取 API 密钥。
    • 在 VS Code 的设置中，你也可以配置 Claude Code 的行为，如自动建议、代码补全风格等。

注意 ：Claude Code 是开发工具，不是运行时依赖。我们最终部署的系统运行时，依赖的是通过 Anthropic 官方 anthropic Python 库或其他兼容方式对 Claude API 的调用。

2.3 核心 Python 依赖库安装

创建一个新的项目目录，并初始化虚拟环境，然后安装核心库。

# 创建项目目录并进入
mkdir ai-berkshire && cd ai-berkshire

# 创建虚拟环境 (以 venv 为例)
python -m venv venv

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

# 安装核心依赖
pip install anthropic          # Claude API 官方客户端
pip install openai             # 如需使用其他模型如 GPT，或某些框架依赖此接口
pip install langchain          # 强大的 LLM 应用开发框架，内置多 Agent 原型
pip install langchain-community # LangChain 社区工具集成
pip install tavily-python      # 网络搜索 API (可选，用于信息收集Agent)
pip install python-dotenv      # 管理环境变量
pip install pandas             # 数据处理 (用于财务分析Agent)

langchain 库是我们构建多 Agent 系统的基石。它抽象了 LLM 调用、工具定义、记忆、链（Chain）和代理（Agent）等概念，让我们能更专注于业务逻辑而非底层通信细节。

2.4 项目结构与配置文件

一个清晰的项目结构有助于管理复杂度。创建如下文件和目录：

ai-berkshire/
├── .env                    # 环境变量配置文件（切勿提交至Git）
├── .gitignore
├── requirements.txt        # 依赖列表
├── config/
│   └── settings.py         # 应用配置
├── agents/                 # 各个Agent的定义
│   ├── __init__.py
│   ├── research_director.py
│   ├── data_collector.py
│   └── financial_analyst.py
├── tools/                  # 自定义工具函数
│   ├── __init__.py
│   └── web_search.py
├── memory/                 # 记忆/状态管理
│   ├── __init__.py
│   └── task_state.py
├── main.py                 # 系统入口，协调Agent工作流
└── utils/
    ├── __init__.py
    └── prompts.py          # 存放给各Agent的提示词模板

首先，在 .env 文件中配置你的 API 密钥和其他敏感信息：

# .env
ANTHROPIC_API_KEY=your_anthropic_api_key_here
TAVILY_API_KEY=your_tavily_api_key_here  # 如果使用Tavily搜索
OPENAI_API_KEY=your_openai_api_key_here  # 如果备用或使用LangChain的OpenAI封装

然后在 config/settings.py 中读取这些配置：

# config/settings.py
import os
from dotenv import load_dotenv

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

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# 模型配置
CLAUDE_MODEL = "claude-3-5-sonnet-20241022"  # 根据可用性调整

3. 构建“AI Berkshire”多 Agent 系统

我们将实现一个简化但完整的主从式多 Agent 系统，包含三个核心 Agent：研究主管、数据收集员、财务分析师。

3.1 定义工具（Skills）

工具是 Agent 能力的延伸。我们先实现一个网络搜索工具，供数据收集员使用。这里使用 tavily-python 作为示例。

# tools/web_search.py
import asyncio
from tavily import AsyncTavilyClient
from config.settings import TAVILY_API_KEY

class WebSearchTool:
    def __init__(self):
        # 初始化异步客户端
        self.client = AsyncTavilyClient(api_key=TAVILY_API_KEY)

    async def search(self, query: str, max_results: int = 5) -> str:
        """执行网络搜索并返回格式化结果。"""
        if not TAVILY_API_KEY:
            return "错误：未配置 Tavily API 密钥。"
        try:
            # 执行搜索
            search_result = await self.client.search(query, max_results=max_results)
            # 格式化结果
            formatted_results = []
            for result in search_result.get('results', []):
                title = result.get('title', '无标题')
                url = result.get('url', '无链接')
                content = result.get('content', '无内容')
                formatted_results.append(f"标题: {title}\n链接: {url}\n摘要: {content[:200]}...\n")
            return "\n---\n".join(formatted_results) if formatted_results else "未找到相关结果。"
        except Exception as e:
            return f"搜索过程中发生错误: {str(e)}"

# 同步包装器，方便在 LangChain 的 Tool 接口中使用
def sync_search(query: str) -> str:
    """同步版本的搜索函数，供 LangChain Tool 调用。"""
    tool = WebSearchTool()
    return asyncio.run(tool.search(query))

3.2 实现数据收集员 Agent

这个 Agent 负责根据指令搜索公司基本信息、新闻、行业动态等。

# agents/data_collector.py
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain.tools import Tool
from config.settings import ANTHROPIC_API_KEY, CLAUDE_MODEL
from tools.web_search import sync_search

class DataCollectorAgent:
    def __init__(self):
        # 1. 初始化 LLM
        llm = ChatAnthropic(
            model=CLAUDE_MODEL,
            api_key=ANTHROPIC_API_KEY,
            temperature=0.1,  # 低随机性，确保信息准确
            max_tokens=4000
        )

        # 2. 定义工具列表
        tools = [
            Tool(
                name="web_search",
                func=sync_search,
                description="""用于搜索公司的最新新闻、行业报告、基本信息等。
                输入应为明确的搜索查询语句，例如：'苹果公司 2024年第一季度财报'。
                """
            ),
            # 未来可以添加更多工具，如数据库查询、文件读取等
        ]

        # 3. 定义提示词模板
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个专业的数据收集员。你的任务是准确、高效地根据用户问题，使用提供的工具搜集相关信息。
            只使用工具获取信息，不要编造。将搜集到的信息清晰、有条理地汇总。
            如果工具返回的结果不足或无关，请如实说明。"""),
            ("placeholder", "{chat_history}"),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ])

        # 4. 创建 Agent 和执行器
        agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=prompt)
        self.agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

    def run(self, task_description: str) -> str:
        """执行数据收集任务。"""
        try:
            result = self.agent_executor.invoke({"input": task_description})
            return result.get("output", "数据收集完成，但未返回明确结果。")
        except Exception as e:
            return f"数据收集员执行任务时出错: {str(e)}"

3.3 实现财务分析师 Agent

这个 Agent 负责处理数字、计算比率、进行初步的财务评估。它可能不需要调用外部 API，但需要强大的逻辑推理和计算提示。

# agents/financial_analyst.py
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from config.settings import ANTHROPIC_API_KEY, CLAUDE_MODEL

class FinancialAnalystAgent:
    def __init__(self):
        # 1. 初始化 LLM
        self.llm = ChatAnthropic(
            model=CLAUDE_MODEL,
            api_key=ANTHROPIC_API_KEY,
            temperature=0.2,
            max_tokens=3000
        )
        # 2. 定义分析链（一个简化的 LangChain Chain）
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一名严谨的价值投资财务分析师。你的核心工作是分析提供的财务数据、业务描述和行业信息。
            请遵循以下步骤：
            1. **识别关键数据**：找出营收、利润、增长率、负债率、现金流等核心指标。
            2. **计算与对比**：如果数据允许，计算常见的财务比率（如市盈率P/E、市净率P/B、净资产收益率ROE、负债权益比等），并与行业平均水平或历史数据进行对比。
            3. **优劣势分析**：列出公司的财务优势和潜在风险点。
            4. **初步判断**：基于以上分析，给出一个关于公司财务健康状况和估值水平的初步定性判断。
            请以结构清晰、论点有数据支撑的方式呈现你的分析。如果提供的信息不足以完成某项分析，请明确指出。"""),
            ("human", "请分析以下信息：\n\n{information}"),
        ])
        self.analysis_chain = prompt | self.llm | StrOutputParser()

    def run(self, information: str) -> str:
        """执行财务分析任务。"""
        try:
            analysis = self.analysis_chain.invoke({"information": information})
            return analysis
        except Exception as e:
            return f"财务分析师执行分析时出错: {str(e)}"

3.4 实现研究主管 Agent 与工作流协调

研究主管是大脑，它不直接调用搜索或计算工具，而是理解用户的高层意图（如“分析一下宁德时代的投资价值”），将其分解为子任务，协调其他 Agent 工作，并整合最终报告。

# agents/research_director.py
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from config.settings import ANTHROPIC_API_KEY, CLAUDE_MODEL
# 导入其他Agent，在实际项目中可能通过更解耦的方式调用（如消息队列）
from agents.data_collector import DataCollectorAgent
from agents.financial_analyst import FinancialAnalystAgent
from memory.task_state import TaskStateManager  # 假设有一个状态管理器

class ResearchDirectorAgent:
    def __init__(self):
        self.llm = ChatAnthropic(
            model=CLAUDE_MODEL,
            api_key=ANTHROPIC_API_KEY,
            temperature=0.3,  # 稍高创造性，用于任务规划和报告合成
            max_tokens=4000
        )
        self.data_collector = DataCollectorAgent()
        self.financial_analyst = FinancialAnalystAgent()
        self.state_manager = TaskStateManager()
        # 定义规划链
        planning_prompt = ChatPromptTemplate.from_messages([
            ("system", """你是投资研究团队的主管。收到一个研究请求后，你需要制定一个分步计划。
            通常，一个完整的研究计划包括：
            1. **信息收集阶段**：搜集目标公司的基本信息、最新动态、行业地位、竞争格局。
            2. **财务分析阶段**：基于收集到的信息，提取并分析财务数据，评估其财务健康度和估值。
            3. **综合评估阶段**：整合前两个阶段的发现，形成最终的投资建议摘要。
            请根据用户请求，生成一个具体的、可执行的任务列表。每个任务应明确指派给`数据收集员`或`财务分析师`。
            输出格式为：
            - [任务1描述] (指派给：数据收集员)
            - [任务2描述] (指派给：财务分析师)
            ..."""),
            ("human", "研究请求：{request}"),
        ])
        self.planning_chain = planning_prompt | self.llm | StrOutputParser()
        # 定义报告合成链
        synthesis_prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一名资深投资顾问。现在你手头有关于某公司的信息收集结果和财务分析报告。
            你的任务是撰写一份简洁、专业、面向投资者的最终摘要报告。
            报告应包含：
            - **公司概览**：核心业务与市场地位。
            - **近期动态**：关键新闻或事件。
            - **财务亮点**：核心指标与比率分析。
            - **潜在风险**：识别的主要风险点。
            - **初步结论**：基于现有信息的投资吸引力评估（例如：积极关注、保持中性、谨慎看待）。
            请确保结论有前文的分析作为支撑。"""),
            ("human", """请基于以下材料撰写投资摘要报告：
            信息收集结果：
            {collected_info}
            
            财务分析结果：
            {financial_analysis}
            """),
        ])
        self.synthesis_chain = synthesis_prompt | self.llm | StrOutputParser()

    def orchestrate_research(self, user_request: str) -> str:
        """协调整个研究流程。"""
        print(f"【研究主管】收到请求: {user_request}")
        
        # 步骤1：规划任务
        print("【研究主管】正在制定研究计划...")
        plan = self.planning_chain.invoke({"request": user_request})
        print(f"生成的研究计划：\n{plan}\n")
        
        # 步骤2：执行任务（这里简化了任务解析和分配，实际项目需要更复杂的解析逻辑）
        collected_info = ""
        financial_analysis = ""
        
        # 模拟根据计划分配任务。实际中需要解析plan文本。
        if "数据收集员" in plan:
            print("【研究主管】指派任务给数据收集员...")
            # 这里简化：直接让数据收集员搜索公司名
            company_name = user_request.split('分析')[-1].split('的')[0].strip() or "目标公司"
            search_task = f"搜索关于{company_name}公司的最新新闻、业务介绍和行业信息"
            collected_info = self.data_collector.run(search_task)
            print(f"数据收集员返回信息长度：{len(collected_info)} 字符\n")
        
        if "财务分析师" in plan:
            print("【研究主管】指派任务给财务分析师...")
            # 将收集到的信息交给财务分析师
            financial_analysis = self.financial_analyst.run(collected_info)
            print(f"财务分析师返回分析长度：{len(financial_analysis)} 字符\n")
        
        # 步骤3：合成最终报告
        print("【研究主管】正在整合信息，撰写最终报告...")
        final_report = self.synthesis_chain.invoke({
            "collected_info": collected_info,
            "financial_analysis": financial_analysis
        })
        
        return final_report

3.5 实现简单的任务状态管理

为了在 Agent 间传递信息，我们需要一个简单的共享状态。这里用一个全局字典模拟。

# memory/task_state.py
class TaskStateManager:
    _instance = None
    _state = {}  # 全局状态字典
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(TaskStateManager, cls).__new__(cls)
        return cls._instance
    
    def set(self, key: str, value: any):
        self._state[key] = value
    
    def get(self, key: str, default=None):
        return self._state.get(key, default)
    
    def append_to_list(self, key: str, value: any):
        if key not in self._state:
            self._state[key] = []
        self._state[key].append(value)

3.6 系统入口与主流程

最后，在 main.py 中创建系统入口。

# main.py
import asyncio
from agents.research_director import ResearchDirectorAgent

async def main():
    print("=== AI Berkshire 价值投资研究系统启动 ===")
    director = ResearchDirectorAgent()
    
    # 模拟用户输入
    user_request = "分析一下宁德时代的投资价值"
    # user_request = input("请输入您想研究的公司或主题：")
    
    print(f"开始处理请求：'{user_request}'")
    print("-" * 50)
    
    try:
        final_report = director.orchestrate_research(user_request)
        print("\n" + "="*60)
        print("最终投资研究报告摘要")
        print("="*60)
        print(final_report)
        print("="*60)
    except Exception as e:
        print(f"系统执行过程中发生错误: {e}")

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

4. 运行验证与结果分析

完成所有代码后，我们可以运行系统，观察多 Agent 是如何协作的。

4.1 启动系统

在项目根目录下，确保虚拟环境已激活，然后运行：

python main.py

4.2 观察控制台输出

你应该能看到类似以下的输出，清晰地展示了工作流的推进：

=== AI Berkshire 价值投资研究系统启动 ===
开始处理请求：'分析一下宁德时代的投资价值'
--------------------------------------------------
【研究主管】收到请求: 分析一下宁德时代的投资价值
【研究主管】正在制定研究计划...
生成的研究计划：
- 搜索宁德时代新能源科技股份有限公司的最新公司新闻、业务发展及市场动态 (指派给：数据收集员)
- 查找宁德时代近期的财务报告或相关财务数据摘要 (指派给：数据收集员)
- 基于收集到的财务信息，进行财务比率计算和健康状况评估 (指派给：财务分析师)
- 整合业务动态和财务分析，评估其投资潜力和主要风险 (指派给：研究主管)

【研究主管】指派任务给数据收集员...
> Entering new AgentExecutor chain...
调用 `web_search` 工具，搜索“宁德时代新能源科技股份有限公司的最新公司新闻、业务发展及市场动态”...
工具返回大量搜索结果摘要...
数据收集员 Agent 经过思考，可能继续调用工具搜索财务数据...
> Finished chain.
数据收集员返回信息长度：3250 字符

【研究主管】指派任务给财务分析师...
财务分析师返回分析长度：1200 字符

【研究主管】正在整合信息，撰写最终报告...

============================================================
最终投资研究报告摘要
============================================================
**公司概览**：宁德时代新能源科技股份有限公司是全球领先的锂离子电池研发制造公司，专注于新能源汽车动力电池系统、储能系统的研发、生产和销售...
**近期动态**：根据近期信息，宁德时代在固态电池技术研发、海外工厂建设（如德国、匈牙利）方面取得进展，并与多家主流车企达成长期供应协议...
**财务亮点**：基于可获得的财务信息摘要，公司营收保持高速增长，毛利率在行业中处于领先地位。然而，需要完整的财务报表来计算精确的市盈率(P/E)和净资产收益率(ROE)...
**潜在风险**：1. 行业竞争加剧，国内外竞争对手产能扩张。2. 原材料（如锂、钴）价格波动风险。3. 技术路线迭代风险（如固态电池对现有体系的潜在冲击）...
**初步结论**：宁德时代在其领域拥有显著的先发优势和规模效应，业务动态积极。但鉴于提供的财务数据有限，难以进行精确估值。建议**保持积极关注**，并获取其最新完整年报以进行更深入的定量分析。
============================================================

4.3 结果解读

从输出可以看出：

  1. 规划阶段 ：研究主管成功地将模糊的用户请求，分解为四个具体的、可指派的任务。
  2. 执行阶段 ：
    • 数据收集员 Agent 被调用，它自动使用了我们提供的 web_search 工具，执行了搜索并返回了格式化信息。
    • 财务分析师 Agent 接收了收集到的文本信息，并输出了一段结构化的财务分析，尽管它指出原始数据不足以进行精确计算。
  3. 合成阶段 ：研究主管 Agent 将前两个阶段的结果作为输入，生成了一份包含五个部分的最终投资摘要报告。

这个流程验证了多 Agent 协作的基本可行性：任务分解、工具调用、信息传递和结果整合。

5. 常见问题排查与调试

在开发此类系统时，你会遇到各种问题。以下是一些典型问题及其排查路径。

5.1 Agent 不调用工具或调用错误

问题现象 可能原因 检查方式 处理建议
Agent 完全忽略工具描述，自行生成答案。 1. 提示词（System Prompt）未强调使用工具。
2. 工具描述（ description ）不够清晰或与问题不匹配。
3. LLM 温度（ temperature ）设置过高，导致随机性太大。		 1. 检查 Agent 的 System Prompt，确保有“请使用工具”、“你必须使用提供的工具”等指令。
2. 检查 Tool 的 description 字段，确保它准确描述了工具的功能和输入格式。
3. 将 temperature 调低（如 0.1）。		 1. 优化提示词，明确角色和工具使用规则。
2. 重写工具描述，使其更具体、更具操作性。
3. 在开发调试阶段，使用低 temperature 。
Agent 尝试调用工具，但参数格式错误。 1. 工具函数期望的参数类型（如 str ）与 Agent 传递的不符。
2. LangChain 的 Tool 封装或 Agent 类型选择有误。		 1. 查看错误堆栈，确认是哪里解析失败。
2. 打印 Agent 生成的中间 agent_scratchpad 日志（设置 verbose=True ）。		 1. 确保工具函数接收一个字符串参数。
2. 使用 create_tool_calling_agent （推荐）或 create_react_agent 等现代 Agent 构造器，它们对工具调用的支持更好。

5.2 API 密钥或网络连接问题

问题现象 可能原因 检查方式 处理建议
anthropic.BadRequestError 或 AuthenticationError 。 1. API 密钥未设置或错误。
2. 密钥对应的模型权限不足或余额耗尽。
3. 请求区域受限。		 1. 检查 .env 文件中的 ANTHROPIC_API_KEY 变量名和值是否正确。
2. 在 Anthropic 控制台检查密钥状态和用量。
3. 尝试一个简单的直接 API 调用测试连通性。		 1. 重新生成并配置 API 密钥。
2. 确认所请求的模型（如 claude-3-5-sonnet ）在你的计划中可用。
3. 遵循服务商的使用条款。
Timeout 或网络错误。 1. 本地网络不稳定。
2. 服务器端响应慢。		 1. 使用 curl 或 ping 测试基础网络。
2. 在代码中增加请求超时（ timeout ）参数。		 1. 检查本地代理或防火墙设置。
2. 在初始化 ChatAnthropic 时设置 timeout 参数。

5.3 工作流逻辑错误

问题现象 可能原因 检查方式 处理建议
研究主管没有正确分解任务，或分配混乱。 1. 规划链的提示词不够具体。
2. 对规划链输出的文本解析逻辑太简单或脆弱。		 1. 打印 planning_chain 生成的原始计划文本。
2. 用多种不同的用户请求测试，观察计划的稳定性。		 1. 细化规划提示词，要求输出更结构化的格式（如 JSON），便于程序解析。
2. 实现一个简单的解析器来提取“任务描述”和“指派对象”，而不是依赖字符串包含判断。
各个 Agent 的输出质量低下，信息重复或矛盾。 1. 每个 Agent 的提示词角色定义不清，任务边界模糊。
2. 传递给下游 Agent 的上下文信息过多或过杂。		 1. 逐一测试每个 Agent 的独立表现，给定明确输入看输出。
2. 检查从数据收集员到财务分析师传递的信息是否包含大量无关噪音。		 1. 为每个 Agent 精心设计 System Prompt，明确其专长和职责范围。
2. 在 Agent 间传递信息前，可以增加一个“信息过滤”或“总结”步骤，只传递关键信息。

6. 生产环境最佳实践与扩展方向

上述示例是一个用于学习和验证概念的原型。要将其发展为可用的生产系统，需要考虑更多工程因素。

6.1 架构升级建议

  1. 解耦与消息队列 ：当前 Agent 间是直接函数调用，耦合度高。生产环境应使用消息队列（如 RabbitMQ、Redis Streams）或工作流引擎（如 Apache Airflow、Prefect）来管理任务分发和状态，提高系统的可扩展性和容错性。
  2. 强化记忆与状态管理 ：使用向量数据库（如 Pinecone、Weaviate、Chroma）存储长期记忆和对话历史，使 Agent 能进行更复杂的多轮协作和参考历史结论。
  3. 可观测性与日志 ：为每个 Agent 的执行过程添加结构化日志，记录输入、输出、工具调用详情、耗时和 Token 使用量。这便于监控、计费和调试。
  4. 异步与并发 ：利用 asyncio 等机制让可以并行的任务（如同时搜索多家竞争对手信息）真正并发执行，大幅缩短整体响应时间。

6.2 工具生态扩展

一个强大的 Agent 系统依赖于丰富的工具集。可以考虑集成：

  • 金融数据工具 ：接入 akshare 、 yfinance 等库获取结构化财务数据。
  • 文档处理工具 ：解析 PDF 年报、研报，提取表格和文本。
  • 代码执行工具 ：让 Agent 能够运行 Python 代码进行复杂计算或数据处理。
  • 内部系统工具 ：连接公司内部的 CRM、ERP 数据库，获取私有数据。

6.3 提示词工程优化

提示词是 Agent 的“灵魂”，需要持续迭代：

  • 少样本学习（Few-shot） ：在提示词中提供几个高质量的任务分解、工具调用或分析报告的示例，能显著提升 Agent 输出的稳定性和质量。
  • 动态提示词 ：根据任务类型、用户身份或历史交互，动态组装或调整提示词。
  • 输出结构化 ：严格要求 Agent 以 JSON、XML 或特定 Markdown 格式输出，便于后续程序化处理。

6.4 安全与成本控制

  • 工具权限管控 ：不是所有 Agent 都能调用所有工具。需要建立权限机制，防止越权操作（如删除文件、发送邮件）。
  • 输入输出过滤 ：对用户输入和 Agent 输出进行内容安全过滤，防止注入攻击或生成不当内容。
  • 成本监控 ：监控每个请求的 Token 消耗，特别是当使用高成本模型或频繁调用外部 API（如搜索）时。设置预算和用量告警。
  • 限流与降级 ：对 API 调用实施限流，并在主要服务不可用时，有备用的模型或降级方案。

通过从原型到生产级的持续迭代，一个基于 Claude Code 和多 Agent 协作的“AI Berkshire”系统，才能真正成为辅助投资研究的可靠工具。这个过程中，对 Agent 行为的设计、工具生态的构建以及系统可靠性的保障，远比单纯调优模型参数更为关键。

Logo
亚马逊云科技技术品牌专区

更多推荐

Kiro Editor 开发实战：使用 Cargo 构建、测试与性能优化指南

欢迎来到这篇终极指南，我们将深入探索如何使用Rust构建高性能的终端文本编辑器Kiro Editor。无论你是Rust新手还是经验丰富的开发者，这篇完整教程将带你了解如何利用Cargo工具链进行高效的开发、测试和性能优化，打造一款快速、轻量且功能强大的UTF-8文本编辑器。## 什么是Kiro Editor？Kiro Editor是一款使用Rust编写的极简终端文本编辑器，它最初是著名编辑

avatar 亚马逊云科技技术品牌专区

近期零基础学量化，先按基础差异拆学习顺序

读者应明白，拆解学习顺序时要先看自己的基础缺口，再决定哪一段需要放慢。不同流程阶段都有不同检查任务，学习路线要能反映这些差异。

avatar 亚马逊云科技技术品牌专区

2026年量化工具选择，要跟着能力基础走

读者应理解，选择工具前要先判断自己的能力基础和当前任务。工具应该帮助自己补上当前最关键的缺口，而不是替代学习顺序本身。

avatar 亚马逊云科技技术品牌专区