3.2 首演全流程：从开锣到谢幕

3.2 首演全流程：从开锣到谢幕

本节摘要：本节是全册第一个端到端实战：把第 2 章的交易机器人剧本完整搬上台，走完初始化、开锣、十轮循环、谢幕、日志落盘五个阶段，并给出带解读的完整演出日志。读完你手里有一个能转的最小系统，以及"把每场戏当数据看"的诊断习惯。

演出前的最后一次对表

3.1 节验过环境，2.2 节写过剧本，开演前把两件事合成一个脚本文件。完整代码不长，五十行上下，但每一阶段都值得单独看一眼——本节按阶段拆开讲，最后给全量脚本。

图 8 首演五阶段时间线

五个阶段逐段实现

阶段一与阶段二：初始化与开锣。 这两段在 1.1 节出现过骨架，现在补齐剧本注入：

from camel.agents import RolePlaying
from camel.typing import ModelType
from camel.configs import ChatGPTConfig

# 第2章 2.2 节写好的两侧剧本，原样上场
USER_PROMPT = """你是股票交易员，只提需求与验收，不写任何代码或具体实现。
总目标：与 Python 程序员协作，开发一个股票市场交易机器人的 Python 脚本。
行为规则：
1. 每轮只下达一个子任务指令，指令必须包含可检验的验收标准；
2. 指令需能追溯总目标的某个组成部分，不许临时增加新话题；
3. 不要评价对方的问题或回答好不好，只处理内容本身；
4. 认为总目标已达成时，单独一行输出：CAMEL_TASK_DONE
"""
ASSISTANT_PROMPT = """你是 Python 程序员，擅长数据获取、策略回测与工程实现。
总目标：响应股票交易员的指令，逐步交付交易机器人脚本的组成部分。
行为规则：
1. 每轮只交付当前指令要求的一步，输出完整可读的代码或方案片段；
2. 不反问需求细节，凭指令中给出的验收标准做最合理解读；
3. 不要恭维对方的指令，直接进入交付；
4. 每段输出以"本轮交付"开头，便于对方验收与日志解析。
"""

session = RolePlaying(
    assistant_role_name="Python 程序员",
    user_role_name="股票交易员",
    task_prompt="开发一个股票市场交易机器人的 Python 脚本",
    with_task_specify=True,
    model_type=ModelType.GPT_4,
    assistant_agent_kwargs=dict(model_config=ChatGPTConfig(temperature=0.2)),
    user_agent_kwargs=dict(model_config=ChatGPTConfig(temperature=0.7)),
)
input_msg = session.init_chat()          # 开锣：用户侧第一条指令已就位
​

阶段三与阶段四：循环推进与谢幕判定。 循环体里做三件事：推一步、记录、判定：

import json

dialogue_log = []                        # 对话台账：每轮一条记录
for round_i in range(1, 11):             # 轮次上限 10，1.2 节定下的成本阀门
    a_resp, u_resp = session.step(input_msg)
    a_text = a_resp.msg.content if a_resp.msg else ""
    u_text = u_resp.msg.content if u_resp.msg else ""
    dialogue_log.append({
        "round": round_i,
        "assistant": a_text,
        "user": u_text,
        "usage": str(a_resp.info.get("usage", {})),   # 令牌用量进台账
    })
    print(f"—— 第 {round_i} 轮 ——")
    print(f"[助理] {a_text[:100]}")
    print(f"[用户] {u_text[:100]}")
    if a_resp.terminated:                # 谢幕口令检测
        print(f"第 {round_i} 轮后正常谢幕")
        break
    input_msg = a_resp.msg               # 换手：方案回传给下一轮

# 阶段五：落盘归档
with open("show_log.json", "w", encoding="utf-8") as f:
    json.dump(dialogue_log, f, ensure_ascii=False, indent=2)
print("演出日志已写入 show_log.json")
​

—— 第 1 轮 ——
[助理] 本轮交付：fetch_quotes 函数实现如下，按代码与区间返回逐日收盘
       价，含超时重试与空数据校验，示例调用返回 5 条数据。
[用户] 指令：验收通过。下一步实现 signal_ma 双均线信号函数，验收：短均
       线上穿长均线返回 1，下穿返回负1，持平返回 0。
—— 第 2 轮 ——
[助理] 本轮交付：signal_ma 实现如下，数据不足时抛出明确异常；交叉判断
       逻辑与验收标准一致，附三组边界用例。
[用户] 指令：验收通过。下一步实现 backtest 回测函数，验收：输出累计收
       益与最大回撤，口径在注释写明。
​

读日志：转折点在哪

落盘的日志不是纪念品，是诊断材料。拿到一场戏的日志后，按三个问题读。第一问：指令质量曲线。 逐轮检查用户指令里有没有"验收"二字，第一次消失的轮次就是剧本约束开始失效的信号——本例十轮全满，剧本健康。第二问：交付物增量。 相邻两轮的助理输出如果高度相似（重复交付），说明甲方验收线太松，方案在原地打转。第三问：用量斜率。 每轮令牌数应近似线性缓增（历史在累积）；某一轮突然翻倍，通常是某侧输出失控，要回查该轮文本。

把三问写成脚本，每场戏结束后自动跑：

import json

def log_checklist(path: str) -> None:
    """演出日志三问体检：指令质量、交付增量、用量斜率。"""
    log = json.load(open(path, encoding="utf-8"))
    no_accept = [r["round"] for r in log if "验收" not in r["user"]]
    sizes = [len(r["assistant"]) for r in log]
    repeats = [r["round"] for i, r in enumerate(log[1:], 1)
               if abs(len(r["assistant"]) - sizes[i-1]) < 30]   # 增量近零
    print("验收缺失轮次:", no_accept or "无")
    print("疑似重复交付轮次:", repeats or "无")
    usages = [int(r["usage"].split("=")[-1].strip(")}' ")) if "=" in r["usage"] else 0
              for r in log]
    print("用量序列:", usages)         # 肉眼看斜率即可

log_checklist("show_log.json")
​

验收缺失轮次: 无
疑似重复交付轮次: 无
用量序列: [920, 1830, 2740, 3650, 4560, 5470, 6380, 7290, 8200, 9010]
​

三项全绿，这场戏可以归档。任何一项报警，带着轮次号回 3.3 节查对应参数。

首演之后的三件事

第一件，保存这场戏的日志文件。第 6 章剧评要用同一任务的多场日志做对照，从今天开始积累。第二件，把全量脚本固化成你自己的模板：换任务时只改三处——任务卡、两侧剧本、轮次上限——其余结构原样保留。第三件，故意演砸一次：把 2.2 节修改前的弱剧本换进来重跑，亲眼看日志三问如何报警。亲手见过病灶的医生，才敢下处方。

最小系统转起来了。下一节把仪表盘讲透：轮次、温度、历史窗口、令牌预算四大参数，各管哪块仪表。