Azure Functions 5 天上手手册

给 Javis 的快速上手。你 D+4 那套 LangGraph AI 投资系统是 serverless + graph orchestration 的**教科书案例**。Azure Functions + Durable Functions 跟 LangGraph 概念 1:1 对应, 你学 Azure 时其实是在用你 D+4 的知识。这篇 5 天计划, 每天 1-2 小时, 周五 (D+15) 跑通完整实战。

D1Day 1 · 概念 + 第 1 个 HTTP 函数

2026-08-12 周三 · 60 分钟 · 目标: 本地跑通 hello world

4 个核心概念 (5 分钟读完)

• Function App = 装函数的容器 (类似 microservice 集群)
• Function = 1 段代码 (Python / JavaScript / C# / Java)
• Trigger = 什么事件启动函数 (HTTP 请求 / 定时器 / 队列消息 / 文件上传)
• Binding = 函数 ↔ 外部服务的连接 (声明式, 不用写连接代码)

装环境 (10 分钟)

# Mac
brew tap azure/functions && brew install azure-functions-core-tools@4

# 验证
func --version  # 应该输出 4.x

# 创建项目
func init LocalFunctionProj --python -m V2
cd LocalFunctionProj

# 项目结构
.
├── function_app.py      # 你写代码
├── host.json            # 函数运行时配置
├── local.settings.json  # 本地连接字符串 (含 AzureWebJobsStorage)
├── requirements.txt     # Python 依赖
└── .vscode/             # VS Code 调试配置

第 1 个 HTTP 函数 (15 分钟写代码 + 5 分钟测试)

编辑 function_app.py:

import azure.functions as func
import datetime

app = func.FunctionApp()

@app.route(route="hello", auth_level=func.AuthLevel.ANONYMOUS)
def hello_world(req: func.HttpRequest) -> func.HttpResponse:
    name = req.params.get("name", "World")
    now = datetime.datetime.now().isoformat()
    return func.HttpResponse(
        f"Hello {name}! Server time: {now}",
        status_code=200
    )

启动 + 测试 (5 分钟)

func start
# 看到 "Your function is running at http://localhost:7071/api/hello"

curl http://localhost:7071/api/hello
# 输出: Hello World! Server time: 2026-08-12T...

curl "http://localhost:7071/api/hello?name=Javis"
# 输出: Hello Javis! Server time: 2026-08-12T...

D2Day 2 · 4 个最常用 Trigger

2026-08-13 周四 · 90 分钟 · 目标: 4 个 trigger 本地都能触发

1. HTTP Trigger (Day 1 已经会, 加 GET/POST 分支)

@app.route(route="api/data", methods=["GET", "POST"])
def handle_data(req: func.HttpRequest) -> func.HttpResponse:
    if req.method == "GET":
        return func.HttpResponse("Getting data", status_code=200)
    else:
        body = req.get_json()
        return func.HttpResponse(f"Posted: {body}", status_code=201)

2. Timer Trigger (替代 cron, 定时任务)

@app.schedule(schedule="0 0 */6 * * *", arg_name="timer",
              run_on_startup=False)
def scheduled_job(timer: func.TimerRequest) -> None:
    # 每 6 小时跑一次
    # 你 D+4 那套量化回测可以放这里: 每 6 小时跑一次
    print("Running scheduled job")

3. Queue Trigger (队列消息, 类似 SQS)

@app.queue_trigger(arg_name="msg", queue_name="my-queue",
                   connection="AzureWebJobsStorage")
def process_queue(msg: func.QueueMessage) -> None:
    data = msg.get_json()
    print(f"Got message: {data}")
    # 处理逻辑: 比如更新数据库, 发送通知

队列发送方 (从其他函数发消息):

from azure.storage.queue import QueueClient
import json

queue = QueueClient.from_connection_string(
    "DefaultEndpointsProtocol=https;AccountName=...",
    "my-queue"
)
queue.send_message(json.dumps({"task": "score", "stock": "000001"}))

4. Blob Trigger (文件上传触发, 替代 S3 event)

@app.blob_trigger(arg_name="blob", path="uploads/{name}",
                 connection="AzureWebJobsStorage")
def process_blob(blob: func.InputStream):
    print(f"New file: {blob.name}, size: {blob.length}")
    # 处理 PDF / CSV / 图像 等

D3Day 3 · Durable Functions = Azure 版 LangGraph

2026-08-14 周五 · 120 分钟 · 目标: 跑通"7 维度评分" orchestrator

为什么 Day 3 最关键?

Durable Functions 是 Azure 版的 "graph of states"。跟你 D+4 学过的 LangGraph 概念 1:1 对应:

结论: 你 D+4 学 LangGraph 时, 已经学会了 Durable Functions 的核心概念, 1 天能上手。

实战: "D+4 系统移植" — 7 维度并行打分 orchestrator

把 D+4 那套 LangGraph 7 维度评分系统, 翻译成 Durable Functions:

import azure.functions as func
import azure.durable_functions as df

mybp = df.Blueprint()

# ============ 7 个 Activity (每个 1 维度) ============

@mybp.activity_trigger(input_name="stockCode")
def activity_pe(stockCode: str) -> float:
    """PE 因子"""
    # 假装调用 akshare / yfinance
    return 11.5  # placeholder

@mybp.activity_trigger(input_name="stockCode")
def activity_pb(stockCode: str) -> float:
    """PB 因子"""
    return 1.2

@mybp.activity_trigger(input_name="stockCode")
def activity_roe(stockCode: str) -> float:
    """ROE 因子"""
    return 18.3

@mybp.activity_trigger(input_name="stockCode")
def activity_momentum(stockCode: str) -> float:
    """动量因子"""
    return 0.15

@mybp.activity_trigger(input_name="stockCode")
def activity_vol(stockCode: str) -> float:
    """波动率因子"""
    return 0.32

@mybp.activity_trigger(input_name="stockCode")
def activity_yoy(stockCode: str) -> float:
    """营收 YoY 因子"""
    return 0.08

@mybp.activity_trigger(input_name="stockCode")
def activity_industry(stockCode: str) -> float:
    """行业因子"""
    return 0.05

# ============ Scoring Activity (合并) ============

@mybp.activity_trigger(input_name="scores")
def activity_score(scores: dict) -> dict:
    """7 维度合并打分"""
    weights = {
        "pe": 0.10, "pb": 0.20, "roe": 0.25,
        "mom": 0.15, "vol": 0.10, "yoy": 0.10,
        "industry": 0.10
    }
    total = sum(scores[k] * weights[k] for k in weights)
    return {"scores": scores, "weighted_total": round(total, 4)}

# ============ Orchestrator (像 LangGraph 的 StateGraph) ============

@mybp.orchestration_trigger(context_name="context")
def orchestrator_7dim(context: df.DurableOrchestrationContext):
    stock_code = context.get_input()

    # Fan-out 7 个 activity 并行 (LangGraph 没有这个 native, 要 asyncio.gather)
    tasks = {
        "pe":       context.call_activity("activity_pe", stock_code),
        "pb":       context.call_activity("activity_pb", stock_code),
        "roe":      context.call_activity("activity_roe", stock_code),
        "mom":      context.call_activity("activity_momentum", stock_code),
        "vol":      context.call_activity("activity_vol", stock_code),
        "yoy":      context.call_activity("activity_yoy", stock_code),
        "industry": context.call_activity("activity_industry", stock_code),
    }

    # Wait all (LangGraph 节点 wait)
    scores = {k: t.result() for k, t in tasks.items()}

    # Aggregate
    return context.call_activity("activity_score", scores)

# ============ HTTP 触发 (启动 orchestrator) ============

@mybp.route(route="score/{stockCode}", auth_level=func.AuthLevel.ANONYMOUS)
@mybp.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest,
                     client: df.DurableOrchestrationClient):
    stock = req.route_params.get("stockCode")
    instance_id = await client.start_new("orchestrator_7dim", None, stock)
    return client.create_check_status_response(req, instance_id)

# 注册 Blueprint
app = func.FunctionApp()
app.register_functions(mybp)

本地测试 (10 分钟)

func start

# 启动 orchestrator, 拿到 instance_id
curl -X POST http://localhost:7071/api/score/000001
# 返回: {"id": "abc123", "statusQueryGetUri": "..."}

# 查询结果
curl http://localhost:7071/api/score/000001
# 或浏览器访问 statusQueryGetUri
# 返回: {"runtimeStatus": "Completed", "output": {...7 维度分数 + 总分...}}

D4Day 4 · 部署到 Azure

2026-08-15 周六 · 60 分钟 · 目标: 第 1 个云端 URL

前置 (10 分钟)

1. 注册 Azure 账号: https://azure.microsoft.com/free (1 年免费, 需信用卡)
2. 安装 VS Code + 扩展: "Azure Functions" + "Azure Resources" + "Python"
3. 登录 VS Code (Azure 扩展 → Sign in to Azure)

部署 (30 分钟)

1. VS Code 打开 LocalFunctionProj
2. 左侧 Azure 扩展 → Workspace → Function App → + Create
3. 配置:
   • Runtime: Python 3.11
   • Region: East Asia (离你近)
   • OS: Linux
   • Plan: Consumption (Serverless) — 免费
4. 等 1-2 分钟创建完成
5. 右键 Function App → Deploy to Function App...
6. 确认 (VS Code 会问你 1 次 "Deploy?")

验证 (15 分钟)

1. Azure 门户 (portal.azure.com) → 你的 Function App → Functions → hello_world → "Get Function Url"
2. 复制 URL, curl <URL>?name=Javis
3. 看到 Hello Javis! Server time: 2026-08-12T...

D5Day 5 · 实战项目

2026-08-16 周日 · 150 分钟 · 目标: 完成简历可写项目

项目: "GitHub Trending Monitor"

跟 D+11 CNBC 抓取逻辑一样, 每天自动监控 GitHub trending 30 仓库, 过滤 + 评分 + 存储。

需求拆解

最小可运行代码 (3 文件)

function_app.py:

import azure.functions as func
import azure.durable_functions as df
import requests
import json
import os
from datetime import datetime, timezone

mybp = df.Blueprint()

# ============ Activity 1: 抓 GitHub trending ============

@mybp.activity_trigger(input_name="language")
def fetch_trending(language: str) -> list:
    """抓 1 个语言类目的 trending 仓库"""
    url = f"https://github.com/trending/{language}"
    # 简化: 用 GitHub API 搜 (实际可改 trending HTML 解析)
    headers = {"Accept": "application/vnd.github+json"}
    api_url = "https://api.github.com/search/repositories"
    params = {
        "q": f"language:{language} created:>{(datetime.now(timezone.utc).replace(day=1)).date()}",
        "sort": "stars",
        "order": "desc",
        "per_page": 10
    }
    r = requests.get(api_url, headers=headers, params=params, timeout=10)
    data = r.json()
    repos = []
    for item in data.get("items", []):
        repos.append({
            "name": item["full_name"],
            "stars": item["stargazers_count"],
            "language": item["language"],
            "url": item["html_url"],
            "description": (item["description"] or "")[:100]
        })
    return repos

# ============ Activity 2: 评分 ============

@mybp.activity_trigger(input_name="repo")
def score_repo(repo: dict) -> dict:
    """简单评分: stars + 1/年龄 (越新越好)"""
    # 简化: 只看 stars 和创建时间
    score = repo["stars"] * 0.7 + (1000 if repo["stars"] > 1000 else 0) * 0.3
    repo["score"] = round(score, 2)
    return repo

# ============ Orchestrator: 拉 3 个语言, 各取 10 仓库, 共 30 个 ============

@mybp.orchestration_trigger(context_name="context")
def orchestrator_trending(context: df.DurableOrchestrationContext):
    languages = ["python", "typescript", "javascript"]

    # Fan-out 3 个语言
    lang_tasks = {
        lang: context.call_activity("fetch_trending", lang)
        for lang in languages
    }

    # 收集 + flatten
    all_repos = []
    for lang, task in lang_tasks.items():
        repos = task.result()
        all_repos.extend(repos)

    # Fan-out 评分 (30 个并行)
    score_tasks = [context.call_activity("score_repo", r) for r in all_repos]

    # 排序 + top 10
    scored = [t.result() for t in score_tasks]
    top_10 = sorted(scored, key=lambda r: r["score"], reverse=True)[:10]

    return {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "total_repos": len(all_repos),
        "top_10": top_10
    }

# ============ HTTP 启动 ============

@mybp.route(route="run", auth_level=func.AuthLevel.ANONYMOUS)
@mybp.durable_client_input(client_name="client")
async def http_start(req: func.HttpRequest,
                     client: df.DurableOrchestrationClient):
    instance_id = await client.start_new("orchestrator_trending", None)
    return client.create_check_status_response(req, instance_id)

# ============ Timer 每日 9:00 触发 ============

@mybp.schedule(schedule="0 0 9 * * *", arg_name="timer")
@mybp.durable_client_input(client_name="client")
async def scheduled_run(timer: func.TimerRequest,
                       client: df.DurableOrchestrationClient):
    await client.start_new("orchestrator_trending", None)

# 注册
app = func.FunctionApp()
app.register_functions(mybp)

测试 (30 分钟)

func start

# 手动触发
curl -X POST http://localhost:7071/api/run
# 返回: {"id": "xyz789", ...}

# 查询结果 (返回可能需要 5-10 秒)
sleep 10
curl http://localhost:7071/api/run  # 实际应该用 statusQueryGetUri
# 返回: 30 个仓库 + top 10

部署到 Azure (30 分钟)

1. VS Code 右键 Function App → Deploy
2. 在 Azure 门户配置 Timer 触发: 每日 9:00 跑
3. 添加 Cosmos DB binding (存储历史, 进阶)

常见问题 FAQ

Q1: 我没有 Azure 账号, 能学吗?

能。Day 1-3 + Day 5 全部本地跑 (用 Azurite emulator)。Day 4 部署是唯一需要 Azure 账号的步骤。你可以周末再注册 Azure 免费层。

Q2: Durable Functions 跟 LangGraph 哪个先学?

你已经会 LangGraph。直接学 Durable Functions, 1 天能上手。如果你没学 LangGraph, 建议先学 LangGraph (Python, 3 天), 再学 Durable Functions (1 天)。

Q3: 学完这个能写简历吗?

能。Day 5 项目做完, 简历可以写:

• "用 Azure Functions + Durable Functions 实现 GitHub Trending 监控, 每日 9:00 自动运行, 监控 30+ 仓库"
• "用 Durable Functions Orchestrator 实现 30 个仓库并行评分, 评分时间从 30s 降到 5s"
• "用 Azure Storage + Timer Trigger 实现 serverless 批处理 pipeline"

Q4: 这个跟 AWS Lambda 什么关系?

Azure Functions ≈ AWS Lambda, Durable Functions ≈ AWS Step Functions. 概念 80% 一样, API 不同。学会 Azure 1 周, AWS Lambda 3 天能上手。

Q5: 100 天冲刺期间值得学吗?

周末学 (D+13/14)。不影响主线 (求职 / 量化 / DP-600)。Azure 是加分项, 不是必需, 目标公司如果用 Azure 云就重要。

诚实总结

5 天下来, 你会:

1. 理解 Azure Functions 的 4 个核心概念 (Function App / Function / Trigger / Binding)
2. 写 4 种 Trigger 的 HTTP / Timer / Queue / Blob 函数
3. 用 Durable Functions 跑 long-running orchestrator (从 D+4 LangGraph 直接翻译)
4. 部署 到 Azure 云
5. 做完 1 个简历级项目 (GitHub Trending Monitor)

前提: 你已经有 Python 基础 + 1 段 D+4 LangGraph 实战经验。你是最有优势的初学者。

本研究基于 Azure 官方文档 (learn.microsoft.com) + GitHub API 元数据 + Javis D+4 LangGraph 实战经验. 所有代码示例为简化版, 实际部署需根据 Azure 订阅和 region 调整. 诚实标注: Day 3 Durable Functions 概念偏难, 第 1 次可能 1-2 天才能跑通.