第 3 课：一键自动化 query()

你给一个 prompt → AI 处理 → 返回结果

import anyio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)

async def ask(question: str):
    """问一个问题"""
    async for msg in query(prompt=question):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(lambda: ask("Python 的 list comprehension 怎么用？举个例子"))

import anyio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)

async def generate_daily_report(tasks_done: list[str], tasks_todo: list[str]):
    """根据完成和待做任务生成日报"""
    options = ClaudeAgentOptions(
        system_prompt="""你是效率助手。根据提供的任务信息生成日报。
格式要求：
1. 今日完成（简洁列出）
2. 明日计划（简洁列出）
3. 一句话总结今天的效率
用中文，总字数不超过 200 字。""",
        max_turns=1,
    )

    prompt = f"""今日完成的任务：
{chr(10).join(f'- {t}' for t in tasks_done)}

待做任务：
{chr(10).join(f'- {t}' for t in tasks_todo)}

请生成日报。"""

    result = ""
    async for msg in query(prompt=prompt, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    result += block.text
    return result


async def main():
    report = await generate_daily_report(
        tasks_done=["完成 API 接口开发", "修复登录 bug #42", "代码审查 3 个 PR"],
        tasks_todo=["写单元测试", "更新文档", "准备周五 Demo"],
    )
    print("📋 日报：")
    print(report)

anyio.run(main)

async def summarize(text: str, max_points: int = 5) -> str:
    """把一大段文字总结成要点"""
    options = ClaudeAgentOptions(
        system_prompt=f"你是摘要专家。把文本总结为不超过 {max_points} 个要点，用中文。每个要点一句话。",
        max_turns=1,
    )

    result = ""
    async for msg in query(prompt=f"请总结：\n\n{text}", options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    result += block.text
    return result

import json
import anyio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)

async def classify_item(item: str) -> dict:
    """智能分类：任务？笔记？提醒？"""
    options = ClaudeAgentOptions(
        system_prompt="""分析用户输入，判断类型和属性。
输出 JSON：
{
  "type": "task" | "note" | "reminder" | "question",
  "priority": "high" | "medium" | "low",
  "content": "整理后的内容",
  "tags": ["标签1", "标签2"],
  "deadline": "截止日期（如果有的话）"
}
只输出 JSON。""",
        max_turns=1,
    )

    result_text = ""
    async for msg in query(prompt=item, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    result_text += block.text

    # 清理 JSON
    clean = result_text.strip()
    if clean.startswith("```"):
        clean = clean.split("\n", 1)[1].rsplit("```", 1)[0]
    return json.loads(clean)


async def main():
    inputs = [
        "下周五之前把报告交了",
        "React 18 的 useTransition 可以做非阻塞渲染",
        "明天下午 3 点开会别忘了",
        "Python 的 GIL 到底是怎么回事？",
    ]

    for item in inputs:
        print(f"\n输入: {item}")
        result = await classify_item(item)
        print(f"  类型: {result['type']}")
        print(f"  优先: {result['priority']}")
        print(f"  标签: {result['tags']}")
        if result.get('deadline'):
            print(f"  截止: {result['deadline']}")

anyio.run(main)

输入: 下周五之前把报告交了
  类型: task
  优先: high
  标签: ['工作', '报告']
  截止: 下周五

输入: React 18 的 useTransition 可以做非阻塞渲染
  类型: note
  优先: low
  标签: ['react', '性能']

输入: 明天下午 3 点开会别忘了
  类型: reminder
  优先: medium
  标签: ['会议']
  截止: 明天15:00

async def batch_process(items: list[str]) -> list[dict]:
    """批量处理：分类 + 整理"""
    results = []
    for i, item in enumerate(items, 1):
        print(f"  处理 {i}/{len(items)}...", end=" ")
        try:
            result = await classify_item(item)
            result["original"] = item
            results.append(result)
            print(f"✅ {result['type']}")
        except Exception as e:
            print(f"❌ {e}")
            results.append({"type": "unknown", "original": item, "error": str(e)})
    return results

# 简单分类 → Haiku（便宜、快）
options = ClaudeAgentOptions(
    system_prompt="...",
    max_turns=1,
    model="claude-haiku-4-5",
)

# 文本摘要 → Sonnet（平衡）
options = ClaudeAgentOptions(
    system_prompt="...",
    max_turns=1,
    model="claude-sonnet-4-5",
)

# 深度分析 → Opus（最强）
options = ClaudeAgentOptions(
    system_prompt="...",
    max_turns=1,
    model="claude-opus-4-6",
)

async for msg in query(prompt="...", options=options):
    if isinstance(msg, ResultMessage):
        if msg.total_cost_usd:
            print(f"💰 本次花了 ${msg.total_cost_usd:.4f}")
        if msg.duration_ms:
            print(f"⏱️ 耗时 {msg.duration_ms}ms")

# ❌ 太模糊
"帮我整理一下"

# ✅ 具体明确
"把以下内容分类为 task/note/reminder，输出 JSON 格式"

try:
    result = json.loads(result_text)
except json.JSONDecodeError:
    result = {"type": "unknown", "content": result_text}