Claude Opus 5.5 登陆 Microsoft Foundry：如何设计可持续数小时的 AI 工作流

import argparse
import json
import os
from pathlib import Path
from urllib.request import Request, urlopen

STATE_FILE = Path('.agent-state.json')
STEPS = [
    '分析目标，列出范围、假设和风险',
    '制定可验证的执行计划，并标出需要人工批准的动作',
    '根据已有结果生成最终交付物和验证清单',
]


def load_state(goal):
    if STATE_FILE.exists():
        state = json.loads(STATE_FILE.read_text(encoding='utf-8'))
        if state.get('goal') != goal:
            raise SystemExit('检查点属于另一个目标，请删除 .agent-state.json 后重试')
        return state
    return {'goal': goal, 'next_step': 0, 'outputs': []}


def save_state(state):
    temp = STATE_FILE.with_suffix('.tmp')
    temp.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8')
    temp.replace(STATE_FILE)


def call_model(prompt):
    endpoint = os.getenv('FOUNDRY_ENDPOINT')
    api_key = os.getenv('FOUNDRY_API_KEY')
    deployment = os.getenv('FOUNDRY_MODEL', 'claude-opus-5-5')

    if not endpoint:
        return '[DRY RUN] 将发送给模型的提示词：\n' + prompt
    if not api_key:
        raise RuntimeError('设置 FOUNDRY_API_KEY 后才能进行实际请求')

    # 假设性请求格式：请按 Foundry 中实际部署的 API/SDK 契约修改。
    body = json.dumps({
        'model': deployment,
        'messages': [{'role': 'user', 'content': prompt}],
    }).encode('utf-8')
    request = Request(
        endpoint,
        data=body,
        method='POST',
        headers={
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {api_key}',
        },
    )
    with urlopen(request, timeout=300) as response:
        result = json.loads(response.read().decode('utf-8'))

    # 同样需要按实际响应结构修改。
    return result.get('output_text') or json.dumps(result, ensure_ascii=False)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--goal', required=True)
    args = parser.parse_args()

    state = load_state(args.goal)
    for index in range(state['next_step'], len(STEPS)):
        previous = '\n\n'.join(state['outputs'])[-12000:]
        prompt = f'''你正在执行一个可恢复的长任务。
总目标：{args.goal}
当前步骤：{STEPS[index]}
已有结果：
{previous or '暂无'}

仅完成当前步骤。明确区分事实、假设和待验证事项。不要声称执行了未提供的工具。'''

        output = call_model(prompt)
        state['outputs'].append(output)
        state['next_step'] = index + 1
        save_state(state)
        print(f'已完成步骤 {index + 1}/{len(STEPS)}，检查点已保存')

    Path('result.md').write_text(
        '\n\n---\n\n'.join(state['outputs']), encoding='utf-8'
    )
    print('任务完成，结果已写入 result.md')


if __name__ == '__main__':
    main()

python runner.py --goal '分析一个旧服务的审计日志改造方案'

export FOUNDRY_ENDPOINT='按实际部署填写请求地址'
export FOUNDRY_API_KEY='从安全的密钥存储注入，不要提交到 Git'
export FOUNDRY_MODEL='按实际部署名称填写'
python runner.py --goal '分析一个旧服务的审计日志改造方案'