第 2 章 · 03 example.py 跑通首个任务

第 2 章 · 03 example.py 跑通首个任务

本节摘要:本节把环境装好、.env 配好后,用项目自带的 example.py 跑通第一个分析任务。example.py 仅 10 行,却是理解 AutoHedge 调用链的最佳入口:它 from autohedge import AutoHedge(触发 .env 加载)、实例化 AutoHedge(name=..., description=...)、把一句自然语言任务交给 run(task=...)。本节逐行拆这 10 行,再追到 main.py 的 __init__ 与 run 看它内部做了什么(初始化 Conversation、把任务交给 director_agent.run、按 output_type 返回 list/dict/str),最后讲清日志里该看到什么、常见报错怎么排查。读完本节,你能在本机跑出第一条 Agent 输出。

内容来源:原项目源码 example.py、autohedge/__init__.py、autohedge/main.py,精读并套用体系化模板。

学习目标

阅读完本节,你应当能够:

1. 逐行读懂 example.py 这 10 行。
2. 说清 from autohedge import AutoHedge 触发了哪些副作用(.env 加载)。
3. 解释 AutoHedge.__init__ 里 Conversation(time_enabled=True) 与 output_dir 的作用。
4. 描述 run(task=...) 的完整调用链(add → director.run → 按格式返回)。
5. 从日志/报错里定位常见问题(key 缺失、网络、模型限流)。

一、example.py 全文逐行精读

example.py 在项目根目录,全文仅 10 行:

from autohedge import AutoHedge  # loads .env from project root

# Initialize the trading system (tickers are derived from the task by the director)
trading_system = AutoHedge(
    name="swarms-fund",
    description="Private Hedge Fund for Swarms Corp",
)

task = "Analyze the sentiment of oil market and provide a thesis on the overall market position and expected trends."
print(trading_system.run(task=task))
​

逐行拆解:

第 1 行:import 触发的副作用

from autohedge import AutoHedge  # loads .env from project root
​

注释「loads .env from project root」点明了:这行 import 不是无副作用的。看 autohedge/__init__.py:

from autohedge.env_loader import load_env
load_env()                                   # ← 副作用:递归找 .env 并加载
from autohedge.main import AutoHedge
__all__ = ["AutoHedge"]
​

import 时会先调 load_env()(上一节讲过的递归 .env 查找),再暴露 AutoHedge 类。这意味着:只要你能成功 from autohedge import AutoHedge,.env 就已经被加载进环境变量了。后续 os.getenv("OPENAI_API_KEY") 才能取到值。

第 3-7 行:实例化

trading_system = AutoHedge(
    name="swarms-fund",
    description="Private Hedge Fund for Swarms Corp",
)
​

只传了 name 和 description 两个参数。其余参数用默认值(见下一节):output_dir="outputs"、output_type="list"、output_file_path=None。注释「tickers are derived from the task by the director」是关键信息——AutoHedge 不预设要分析哪只股票,股票代码由 Director 从任务文本里自己推断(对应第 4 章会讲的 DIRECTOR_TICKER_DISCOVERY_PROMPT)。

第 9-10 行:跑任务并打印

task = "Analyze the sentiment of oil market and provide a thesis on the overall market position and expected trends."
print(trading_system.run(task=task))
​

任务是一句自然语言英文——「分析原油市场情绪,给出整体市场定位与趋势假设」。为什么用英文?因为 prompts.py 里所有 system_prompt 都是英文,LLM 在英文上下文里表现更稳。run(task=task) 把它交给 Director,返回结果后 print 出来。

💡 核心心法:AutoHedge 的「任务接口」就是一句自然语言。这是 LLM Agent 系统的典型范式——不暴露结构化 API,而是用自然语言当契约。好处是灵活,代价是不可预测(同样的句子可能产出不同结果)。

二、AutoHedge.init 内部做了什么

main.py 第 15-31 行的构造函数:

def __init__(
    self,
    name: str = "autohedge",
    description: str = "fully autonomous hedgefund",
    output_dir: str = "outputs",
    output_file_path: str = None,
    output_type: str = "list",
):
    self.name = name
    self.description = description
    self.output_type = output_type
    self.output_file_path = output_file_path
    self.output_dir = Path(output_dir)
    self.output_dir.mkdir(exist_ok=True)

    logger.info("Initializing Automated Trading System")
    self.conversation = Conversation(time_enabled=True)
​

逐段解读:

注意 Conversation(time_enabled=True)——这是 swarms 框架的消息历史抽象,后面 run() 会往里 add 消息,最终按 output_type 取出。第 3 章第 4 节会专门讲它。

⚠️ 现实澄清:output_file_path 参数被存了但全程没用——构造函数里赋值给 self.output_file_path,却没有任何地方写文件。这是又一处「声明了没用」的残留。同理 name、description 也只是存着,不影响行为。

三、run(task) 的完整调用链

main.py 第 33-63 行的 run 方法:

def run(self, task: str, *args, **kwargs):
    logger.info("Starting trading cycle")
    self.conversation.add(role="user", content=f"Task: {task}")

    try:
        output = director_agent.run(task=task)
        self.conversation.add(role="director", content=output)

        if self.output_type == "list":
            return self.conversation.return_messages_as_list()
        if self.output_type == "dict":
            return (
                self.conversation.return_messages_as_dictionary()
            )
        if self.output_type == "str":
            return self.conversation.return_history_as_string()
        return self.conversation.return_messages_as_list()
    except Exception as e:
        logger.error(f"Error in trading cycle: {str(e)}")
        raise
​

调用链拆成四步:

步骤 1:记录任务到会话历史

self.conversation.add(role="user", content=f"Task: {task}")
​

把用户任务以 role="user" 写进 Conversation。注意它给内容加了前缀 "Task: "——后续 Director 收到的并不是裸任务,而是带这个前缀的字符串(但其实 director_agent.run(task=task) 传的是不带前缀的原 task,前缀只出现在历史里)。

步骤 2:交给 Director 执行

output = director_agent.run(task=task)
​

这是整个系统的心脏——把任务交给 director_agent(第 3 章第 1 节那个 handoffs=ALL_AGENTS 的 Agent)。Director 自己决定要不要 handoff 给专家、handoff 给谁,最终产出一串文本。这一步会触发多次 LLM 调用(Director 自己一次,每个被 handoff 的专家各一次),可能耗时几十秒到几分钟。

步骤 3:记录 Director 输出到会话历史

self.conversation.add(role="director", content=output)
​

把 Director 的最终回复以 role="director" 存进历史。

步骤 4:按 output_type 返回

if self.output_type == "list":
    return self.conversation.return_messages_as_list()
...
​

默认 output_type="list",返回 Conversation 里所有消息的列表(每条含 role/content/时间戳)。也支持 "dict" 和 "str"。三种格式的差异在第 5 章第 3 节详讲。

异常处理:except Exception 捕获后用 loguru 记错误,再 raise 向上抛——不吞异常,便于调试。

四、跑起来:命令与预期输出

确保上一节配好了 .env(至少 OPENAI_API_KEY),在项目根目录执行:

python example.py
​

终端预期会看到(节选):

# loguru 打的日志(时间戳+级别)
2026-08-05 14:30:12.123 | INFO     | autohedge.main:__init__:30 - Initializing Automated Trading System
2026-08-05 14:30:12.456 | INFO     | autohedge.main:run:45 - Starting trading cycle

# swarms verbose 日志(Director 与专家的 LLM 调用,可能很长)
... (Director 思考、handoff、各专家输出的过程日志) ...

# 最终 print 出的 Conversation 消息列表(example.py 第 10 行)
[{'role': 'user', 'content': 'Task: Analyze the sentiment of oil market ...',
  'timestamp': '...'},
 {'role': 'director', 'content': '...(Director 的最终回答)...',
  'timestamp': '...'}]
​

因为 verbose=True(workers.py 里四个专家都开了),过程日志会非常详细——包括每次 LLM 调用的输入输出。这正是学习多 Agent 编排的宝贵材料:你可以从日志里看到 Director 把哪些子任务交给了哪个专家。

五、常见报错排查

💡 调试技巧:第一次跑建议把 task 改简单点,比如 "Give a brief thesis on NVDA."——任务越短、Agent 越少被 handoff,跑通越快。等确认链路通了,再跑 example.py 那个原油情绪任务。

六、本节在全景图的位置

example.py ─import─► AutoHedge(__init__) ─run(task)─► director_agent.run
                              │                          │ handoffs
                              ▼                          ▼
                     Conversation 历史            四大专家(下一章)
​

到这里,第 2 章把 AutoHedge 跑起来了。从下一章开始,我们钻进 director_agent.run 内部,看 Director 如何编排四个专家。

本节要点回顾

1. example.py 10 行:import(触发 .env 加载)→ 实例化(只传 name/description)→ run(task=一句自然语言) → print。
2. import 副作用:from autohedge import AutoHedge 会先调 load_env(),所以 import 成功就等于 .env 已加载。
3. init 关键两行:output_dir.mkdir(exist_ok=True) 建目录、Conversation(time_enabled=True) 初始化带时间戳的消息历史;output_file_path 声明了但没用。
4. run 四步:add(user) → director_agent.run(task) → add(director) → 按 output_type(默认 list)返回;异常 raise 不吞。
5. 任务即自然语言:不暴露结构化 API,Director 从任务文本自行推断股票代码与分工,灵活但不可预测。
6. 常见坑:401(key)、EXA 缺失、429(限流)、超时;首次建议用短任务跑通链路再上复杂任务。

下一章,我们钻进 director_agent.run 内部——先看它 handoff 的四个专家 Agent 各自的配置与产出约定。