框架与平台：把重复劳动交出去

打开文件

"""真实 create_agent + 两个本地工具，无密钥运行两个固定问题。"""
import ast
import operator
from scripted_model import ScriptedModel, call
from langchain.agents import create_agent
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.tools import tool


@tool
def weather(city: str) -> str:
    """返回北京或上海的固定教学天气，不是实时天气。"""
    data = {'北京': '22°C 晴', '上海': '25°C 多云'}
    return data.get(city, '错误：仅支持北京或上海')


@tool
def calculate(expression: str) -> str:
    """计算只含数字、括号及加减乘除的算式，不执行任意代码。"""
    ops = {ast.Add: operator.add, ast.Sub: operator.sub,
           ast.Mult: operator.mul, ast.Div: operator.truediv}
    def visit(node):
        if isinstance(node, ast.Constant) and type(node.value) in (int, float):
            if abs(node.value) > 1e9:
                raise ValueError('数字过大')
            return node.value
        if isinstance(node, ast.BinOp) and type(node.op) in ops:
            return ops[type(node.op)](visit(node.left), visit(node.right))
        raise ValueError('仅允许数字和加减乘除')
    try:
        if len(expression) > 100:
            raise ValueError('算式过长')
        return str(visit(ast.parse(expression, mode='eval').body))
    except (ValueError, SyntaxError, ZeroDivisionError, OverflowError) as exc:
        return f'错误：{exc}'


def run(city='北京', expression='(2+3)*6'):
    model = ScriptedModel(steps=[call('weather', {'city': city}, 1),
        call('calculate', {'expression': expression}, 2), AIMessage(content='工具执行完成。')])
    agent = create_agent(model=model, tools=[weather, calculate],
                         system_prompt='使用工具核对天气和算式。')
    return agent.invoke({'messages': [{'role': 'user',
        'content': f'{city}天气如何？计算 {expression}。'}]}, {'recursion_limit': 12})


if __name__ == '__main__':
    print('模式：固定模型响应；真实 LangChain 工具循环')
    for city, expr in [('北京', '(2+3)*6'), ('上海', '8/2')]:
        result = run(city, expr)
        print(f'问题：{city}天气；{expr}')
        for m in result['messages']:
            if isinstance(m, ToolMessage):
                print(f'  工具 {m.name}: {m.content}')
        print('  消息类型：' + ' → '.join(m.type for m in result['messages']))

打开文件

"""固定响应只替代模型；编排、工具执行和消息流仍使用真实框架。"""
import os

# 教学离线模式不启用云端 tracing。
os.environ['LANGSMITH_TRACING'] = 'false'
os.environ['LANGCHAIN_TRACING_V2'] = 'false'

from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult


class ScriptedModel(BaseChatModel):
    steps: list[AIMessage]

    @property
    def _llm_type(self):
        return 'teaching-script-no-network'

    def bind_tools(self, tools, **kwargs):
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        index = sum(isinstance(m, AIMessage) for m in messages)
        if index >= len(self.steps):
            raise RuntimeError('固定响应耗尽：检查工具错误或步骤配置')
        return ChatResult(generations=[ChatGeneration(message=self.steps[index])])


def call(name, args, number):
    return AIMessage(content='', tool_calls=[{
        'name': name, 'args': args, 'id': f'call_{number}', 'type': 'tool_call'}])

打开文件

langchain==1.4.7
langgraph==1.2.18
deepagents==0.7.32
langchain-openai==1.7.3

打开文件

"""真实 StateGraph：有界重试、拒答、checkpoint 暂停后恢复。"""
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver


class State(TypedDict):
    question: str
    attempts: int
    docs: list[str]
    sufficient: bool
    answer: str


def build_graph(pause=False):
    def retrieve(s):
        # 第一次故意未命中，用于确定性验证 retry 分支。
        n = s['attempts'] + 1
        docs = ['密钥放在 .env，不提交到版本库。'] if n == 2 and '密钥' in s['question'] else []
        return {'attempts': n, 'docs': docs}
    def grade(s):
        return {'sufficient': bool(s['docs'])}
    def retry(s):
        return {'answer': ''}
    def answer(s):
        return {'answer': s['docs'][0] if s['sufficient'] else '资料不足，停止检索。'}
    graph = StateGraph(State)
    for name, fn in [('retrieve', retrieve), ('grade', grade), ('retry', retry), ('answer', answer)]:
        graph.add_node(name, fn)
    graph.add_edge(START, 'retrieve')
    graph.add_edge('retrieve', 'grade')
    graph.add_conditional_edges('grade', lambda s:
        'answer' if s['sufficient'] or s['attempts'] >= 2 else 'retry')
    graph.add_edge('retry', 'retrieve')
    graph.add_edge('answer', END)
    return graph.compile(checkpointer=InMemorySaver(), interrupt_before=['answer'] if pause else [])


def initial(question):
    return {'question': question, 'attempts': 0, 'docs': [], 'sufficient': False, 'answer': ''}


def demo():
    graph = build_graph(pause=True)
    for i, question in enumerate(['密钥放哪里？', '火星天气如何？']):
        config = {'configurable': {'thread_id': f'lesson-{i}'}, 'recursion_limit': 16}
        graph.invoke(initial(question), config)
        saved = graph.get_state(config)
        print(f'暂停：next={saved.next[0]}；检索={saved.values["attempts"]} 次')
        result = graph.invoke(None, config)
        print(f'恢复：{result["answer"]}')
    print('存储：进程内 InMemorySaver；退出进程后不保留')


if __name__ == '__main__':
    demo()

打开文件

"""三个实际 CrewAI 角色，固定模型输出；不衡量真实写作质量。"""
import os
import argparse
from pathlib import Path
os.environ['CREWAI_STORAGE_DIR'] = str(Path(__file__).resolve().parent / '.runtime' / 'crew')
os.environ['OTEL_SDK_DISABLED'] = 'true'
os.environ['CREWAI_TELEMETRY_DISABLED'] = 'true'
os.environ['CREWAI_TRACING_ENABLED'] = 'false'

from crewai import Agent, Task, Crew, Process, BaseLLM

FACTS = '事实：学习站提供分阶段课程、本地进度、搜索与可下载实验。进度仅保存在当前浏览器。'
DRAFT = '这款学习站帮助初学者练习 Agent，并保证每个人都能在一天内精通所有框架。'
FINAL = ('这是一款面向 Agent 初学者的分阶段学习站。你可以从模型调用开始，逐步练习工具调用、'
         '状态保存、任务编排与知识检索，再根据项目需要选择框架。每节课程包含目标、动手任务与验收标准。'
         '配套示例提供固定依赖、运行命令和预期输出，方便对照自己的执行结果。遇到错误时，可以先检查'
         '常见报错，再缩小问题范围。搜索入口帮助定位概念，移动端目录方便切换课程，继续学习入口可以'
         '回到尚未完成的内容。学习进度保存在当前浏览器，不会自动同步到其他设备。建议先运行无需密钥的'
         '实验，确认程序流程，再按课程说明接入真实模型。示例中的固定响应用于验证编排逻辑，不能证明模型'
         '质量，也不保证学习时长。完成练习后，请用自己的问题复测，并记录失败场景。')


class RoleModel(BaseLLM):
    def __init__(self, output):
        super().__init__(model='offline-script', temperature=0)
        self.output = output
        self.calls = 0
        self.seen = []
        self.task_seconds = 0.0

    def call(self, messages, tools=None, callbacks=None, available_functions=None, **kwargs):
        self.calls += 1
        self.seen.append(str(messages))
        return 'Final Answer: ' + self.output

    def supports_function_calling(self):
        return False


def run(with_editor=True):
    agents, tasks, models = [], [], []
    roles = [('researcher', '整理可核对事实', FACTS), ('writer', '写约 300 字介绍', DRAFT)]
    if with_editor:
        roles.append(('editor', '删除无依据承诺，输出约 300 字介绍', FINAL))
    for role, goal, output in roles:
        model = RoleModel(output)
        agent = Agent(role=role, goal=goal, backstory='只处理学习站产品资料。',
                      llm=model, allow_delegation=False, max_iter=2, verbose=False)
        task = Task(description=goal, expected_output=goal, agent=agent, context=list(tasks))
        agents.append(agent); tasks.append(task); models.append(model)
    crew = Crew(agents=agents, tasks=tasks, process=Process.sequential,
                memory=False, cache=False, verbose=False, tracing=False)
    result = crew.kickoff()
    for task, model in zip(tasks, models):
        model.task_seconds = task.execution_duration or 0.0
    return result, models


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--timings', action='store_true')
    args = parser.parse_args()
    result, models = run()
    print('角色：' + ' → '.join(t.agent for t in result.tasks_output))
    print('各角色模型调用数：' + ', '.join(str(m.calls) for m in models))
    print('编辑收到草稿：' + str(DRAFT in models[-1].seen[0]))
    if args.timings:
        for output, model in zip(result.tasks_output, models):
            print(f'{output.agent}：{model.task_seconds:.3f}s；输出：{output.raw}')
    print('最终介绍：' + result.raw)
    without, _ = run(False)
    print('去掉编辑后仍有无依据承诺：' + str('保证' in without.raw))
    print('注意：这是固定对照样本；实际耗时与质量需使用真实模型另行测量。')

打开文件

crewai==1.15.30

打开文件

"""真实 Deep Agents 内置文件工具与 todos；文件仅在图状态中。"""
from scripted_model import ScriptedModel, call
from langchain_core.messages import AIMessage, ToolMessage
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
from langchain.agents.middleware import TodoListMiddleware

DOCS = {'/docs/plan.md': '课程：先运行离线示例，再接入真实模型。',
        '/docs/security.md': '密钥：放在环境变量中，不写进公开代码。',
        '/docs/testing.md': '测试：同时检查成功、失败和停止条件。'}
SUMMARY = '# 学习摘要\n\n先离线验证，再接入模型；密钥不公开；测试应覆盖失败与停止条件。\n'


def run():
    steps = [call('write_todos', {'todos': [{'content': '读三份资料并写摘要', 'status': 'in_progress'}]}, 1),
             call('read_file', {'file_path': '/docs/missing.md'}, 2),
             call('ls', {'path': '/docs'}, 3)]
    steps += [call('read_file', {'file_path': p}, i) for i, p in enumerate(DOCS, 4)]
    steps += [call('write_file', {'file_path': '/summary.md', 'content': SUMMARY}, 7),
              call('write_todos', {'todos': [{'content': '读三份资料并写摘要', 'status': 'completed'}]}, 8),
              AIMessage(content='摘要已写入状态文件 /summary.md。')]
    agent = create_deep_agent(model=ScriptedModel(steps=steps), backend=StateBackend(),
                             middleware=[TodoListMiddleware()],
                             system_prompt='只读提供的资料，在状态文件中写摘要。')
    return agent.invoke({'messages': [{'role': 'user', 'content': '读取三份资料并写摘要。'}],
                         'files': {p: create_file_data(c) for p, c in DOCS.items()}},
                        {'recursion_limit': 30})


if __name__ == '__main__':
    result = run()
    names = [m.name for m in result['messages'] if isinstance(m, ToolMessage)]
    print('实际工具：' + ' → '.join(names))
    print('待办：' + result['todos'][0]['status'])
    print('状态文件 /summary.md：')
    print(result['files']['/summary.md']['content'])
    print('宿主磁盘未写入；模型响应为固定脚本，缺文件恢复路径为预设。')

打开文件

"""Dify Chatflow API v1 客户端。默认回放契约样本，--live 才发送网络请求。"""
import argparse
import json
import os
from pathlib import Path
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse


def ask(query, base_url, key, opener=urlopen):
    parsed = urlparse(base_url)
    if parsed.scheme not in ('https', 'http') or not parsed.hostname:
        raise ValueError('DIFY_BASE_URL 必须是完整 http(s) URL，含 /v1')
    if parsed.scheme == 'http' and parsed.hostname not in ('localhost', '127.0.0.1', '::1'):
        raise ValueError('远程服务请使用 HTTPS')
    if not key.strip():
        raise ValueError('缺少 DIFY_API_KEY：使用已发布 Chatflow 的应用密钥')
    payload = {'inputs': {}, 'query': query, 'response_mode': 'blocking', 'user': 'stage6-learner'}
    request = Request(base_url.rstrip('/') + '/chat-messages',
                      data=json.dumps(payload).encode(),
                      headers={'Authorization': 'Bearer ' + key, 'Content-Type': 'application/json'})
    try:
        with opener(request, timeout=30) as response:
            data = json.load(response)
    except HTTPError as exc:
        raise RuntimeError(f'Dify HTTP {exc.code}：检查应用发布、密钥、模型与额度') from None
    except (URLError, TimeoutError):
        raise RuntimeError('Dify 连接失败或超时：检查地址与网络') from None
    except (ValueError, UnicodeError):
        raise RuntimeError('Dify 返回的不是有效 JSON') from None
    return summarize(data)


def summarize(data):
    if not isinstance(data, dict) or not isinstance(data.get('answer'), str) or not data['answer'].strip():
        raise ValueError('响应缺少非空 answer；确认应用类型为 Chatflow')
    metadata = data.get('metadata') or {}
    if not isinstance(metadata, dict):
        raise ValueError('响应 metadata 必须是对象')
    resources = metadata.get('retriever_resources') or []
    if not isinstance(resources, list):
        raise ValueError('retriever_resources 必须是列表')
    names = sorted({r['document_name'] for r in resources if isinstance(r, dict) and r.get('document_name')})
    return {'answer': data['answer'], 'sources': names}


def demo():
    cases = json.loads((Path(__file__).parent/'data/dify-cases.json').read_text())
    print('模式：API 契约样本回放；尚未连接 Dify 知识库')
    for case in cases:
        result = summarize(case['response'])
        print('问题：' + case['query'])
        print('回答：' + result['answer'])
        print('来源：' + ('、'.join(result['sources']) or '无（资料外拒答）'))


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--live', action='store_true')
    parser.add_argument('--query', default='进度如何保存，换设备后还能继续吗？')
    args = parser.parse_args()
    try:
        if args.live:
            print(json.dumps(ask(args.query, os.getenv('DIFY_BASE_URL', 'https://api.dify.ai/v1'),
                                 os.getenv('DIFY_API_KEY', '')), ensure_ascii=False, indent=2))
        else:
            demo()
    except (ValueError, RuntimeError) as exc:
        parser.exit(1, str(exc) + '\n')

打开文件

# 学习站 FAQ（教学数据）

## 1. 面向谁？
学习站面向有基础 Python 知识、想练习 Agent 开发的初学者。

## 2. 进度怎么保存？
学习进度保存在当前浏览器的本地存储中。保存失败时页面会显示提示，可以重试。

## 3. 能跨设备继续吗？
本示例不提供账号和云端同步。换设备或清除浏览器数据后，原设备的本地进度不会自动迁移。

## 4. 必须填写 API 密钥吗？
离线练习不需要模型密钥。真实模型模式需要自行配置密钥，并可能产生 API 费用。

## 5. 如何查找课程？
使用搜索入口输入关键词，或在当前阶段目录中选择小节。

打开文件

[
  {"query":"如何查找课程？","response":{"answer":"使用搜索入口输入关键词，或从阶段目录选择小节。","metadata":{"retriever_resources":[{"document_name":"faq.md","content":"使用搜索入口输入关键词，或在当前阶段目录中选择小节。"}]}}},
  {"query":"进度如何保存，换设备后还能继续吗？","response":{"answer":"进度保存在当前浏览器；换设备不会自动同步。","metadata":{"retriever_resources":[{"document_name":"faq.md","segment_id":"faq-2"},{"document_name":"faq.md","segment_id":"faq-3"}]}}},
  {"query":"明天北京会下雨吗？","response":{"answer":"资料不足，无法回答天气问题。","metadata":{"retriever_resources":[]}}}
]

打开文件

# 6.5 Dify：从知识库到已发布 Chatflow

客户端仅依赖 Python 3.11 标准库，契约为 Dify API v1（核对日期 2026-10-27）。Dify Cloud 持续更新，无可锁定的云端版本；本例不声称已验证某个自托管版本。离线回放证明响应处理可运行，不证明平台检索正确。

1. 创建知识库，上传 `data/faq.md`。按标题分段，确保 FAQ 2 和 FAQ 3 分成两个 chunk。选择你有权限的 embedding 模型并等待索引完成。
2. 新建 **Chatflow**，连接 Start → Knowledge Retrieval → LLM → Answer。检索 query 绑定 Start 的用户问题，选择上述知识库，Top K 设为 3（阈值按实际结果调节）。
3. LLM 的 Context 绑定检索节点 result，用户输入绑定用户问题。系统提示词：`只根据提供的上下文回答。上下文没有答案时说“资料不足”。不要使用外部知识补全。` 在 Answer 节点绑定 LLM 的 text。
4. 启用引用与归属显示。在调试器运行下面 3 个问题，查看命中的 chunk，再发布应用，从该应用的 API Access 获取 API key。
5. 终端设置 DIFY_API_KEY（不要写进 HTML/工作流）；Cloud 使用 `https://api.dify.ai/v1`，自托管使用自己的 API 基地址。脚本不自动读取 .env。

macOS / Linux（隐藏输入；变量仅在当前 shell 有效）：
```sh
read -s DIFY_API_KEY
export DIFY_API_KEY
export DIFY_BASE_URL=https://api.dify.ai/v1
python s65_dify.py --live --query "如何查找课程？"
python s65_dify.py --live --query "进度如何保存，换设备后还能继续吗？"
python s65_dify.py --live --query "明天北京会下雨吗？"
unset DIFY_API_KEY
```

PowerShell 可通过 `$env:DIFY_BASE_URL='https://api.dify.ai/v1'` 设置地址，使用终端安全输入将密钥存入 `$env:DIFY_API_KEY`；结束后 `Remove-Item Env:DIFY_API_KEY`。不要把密钥粘贴进代码或截图。

验收：第一个问题回答查找方法；第二个同时覆盖 FAQ 2、3（跨段综合，而非多个文档）；第三个明确拒答。前两个 metadata.retriever_resources 应有 faq.md；引用的存在不等于语义正确，需核对 chunk 和答案。云端实际措辞与分数会变化。若要做跨文档检索，将 FAQ 2、3 分别另存文件上传，再重复第二问。

常见报错：401 检查应用密钥；404 检查 /v1/chat-messages 路径和已发布应用；400 检查 inputs 必填变量及 Chatflow 类型；429 检查额度；返回空来源时检查索引、检索绑定与引用设置；无关问题仍回答时检查上下文限制和调试轨迹。

官方契约：https://docs.dify.ai/en/api-reference/chat-messages/send-chat-message

打开文件

"""在断网的临时 n8n 容器中实际执行工作流，并检查成功/失败路由。"""
import json
from pathlib import Path
import subprocess
import uuid

IMAGE = 'docker.n8n.io/n8nio/n8n:2.44.7'
ROOT = Path(__file__).resolve().parent


def parse_execution(output):
    # n8n --rawOutput 前仍可能打印迁移与 runner 日志。
    decoder = json.JSONDecoder()
    for i, char in enumerate(output):
        if char != '{':
            continue
        try:
            value, _ = decoder.raw_decode(output[i:])
        except ValueError:
            continue
        if isinstance(value, dict) and 'data' in value and 'finished' in value:
            return value
    raise RuntimeError('没有找到 n8n 执行结果；检查容器日志')


def verify(result):
    if result.get('status') != 'success':
        raise RuntimeError('工作流失败：' + str(result.get('status')))
    runs = result['data']['resultData']['runData']
    items = [item['json'] for name in ('Success', 'Fallback')
             for run in runs.get(name, []) for branch in run['data']['main'] for item in branch]
    routes = {i['case']: i['route'] for i in items}
    if routes != {'valid': 'success', 'malformed': 'fallback', 'missing-field': 'fallback'}:
        raise RuntimeError('分支结果与验收不符：' + str(routes))
    return items


def main():
    name = 'stage6-test-' + uuid.uuid4().hex[:12]
    cmd = ['docker', 'run', '--rm', '--name', name, '--network', 'none',
           '-e', 'N8N_DIAGNOSTICS_ENABLED=false', '-e', 'N8N_VERSION_NOTIFICATIONS_ENABLED=false',
           '-e', 'N8N_TEMPLATES_ENABLED=false', '-e', 'N8N_ENFORCE_SETTINGS_FILE_PERMISSIONS=true',
           '-v', f'{ROOT / "n8n"}:/work:ro', '--entrypoint', '/bin/sh', IMAGE, '-c',
           'n8n import:workflow --input=/work/offline.json >/tmp/import.log && '
           'n8n execute --id=Stage6Smoke --rawOutput']
    try:
        result = subprocess.run(cmd, text=True, capture_output=True, timeout=120)
    finally:
        # 超时杀掉 Docker CLI 后也清理本次容器；不会操作其他容器。
        subprocess.run(['docker', 'rm', '-f', name], capture_output=True, timeout=15)
    if result.returncode:
        raise RuntimeError('n8n 执行失败：\n' + (result.stderr + result.stdout)[-2000:])
    items = verify(parse_execution(result.stdout))
    print('n8n 2.44.7：真实工作流；容器断网；模型输出为固定样本')
    for item in items:
        print(f'{item["case"]} → {item["route"]}：{item["summary"]}')
    print('3/3 分支验收通过；容器退出即删除数据库')


if __name__ == '__main__':
    try:
        main()
    except (RuntimeError, FileNotFoundError, subprocess.TimeoutExpired) as exc:
        raise SystemExit(str(exc))

打开文件

return $input.all().map(item => {
  let raw = item.json.modelText;
  if (raw === undefined) raw = (item.json.output || []).flatMap(x => x.content || []).filter(x => x.type === 'output_text').map(x => x.text).join('');
  try {
    const value = JSON.parse(raw);
    if (!value || typeof value.summary !== 'string' || !value.summary.trim() || value.summary.length > 1000) throw new Error('summary must be a nonempty string <= 1000 chars');
    return {json: {case: item.json.case || 'webhook', ok: true, summary: value.summary}};
  } catch (_) {
    return {json: {case: item.json.case || 'webhook', ok: false, summary: '生成结果格式错误，请稍后重试。'}};
  }
});

打开文件

{
  "id": "Stage6Smoke",
  "name": "Stage6 Offline JSON Guard",
  "active": false,
  "nodes": [
    {
      "id": "Start",
      "name": "Start",
      "type": "n8n-nodes-base.manualTrigger",
      "typeVersion": 1,
      "position": [
        0,
        0
      ],
      "parameters": {}
    },
    {
      "id": "Fixture",
      "name": "Fixture",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        240,
        0
      ],
      "parameters": {
        "jsCode": "return [{json:{case:'valid',modelText:'{\"summary\":\"学习站提供分阶段课程。\"}'}},{json:{case:'malformed',modelText:'{not json'}},{json:{case:'missing-field',modelText:'{\"text\":\"oops\"}'}}];"
      }
    },
    {
      "id": "Validate",
      "name": "Validate",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        480,
        0
      ],
      "parameters": {
        "jsCode": "return $input.all().map(item => {\n  let raw = item.json.modelText;\n  if (raw === undefined) raw = (item.json.output || []).flatMap(x => x.content || []).filter(x => x.type === 'output_text').map(x => x.text).join('');\n  try {\n    const value = JSON.parse(raw);\n    if (!value || typeof value.summary !== 'string' || !value.summary.trim() || value.summary.length > 1000) throw new Error('summary must be a nonempty string <= 1000 chars');\n    return {json: {case: item.json.case || 'webhook', ok: true, summary: value.summary}};\n  } catch (_) {\n    return {json: {case: item.json.case || 'webhook', ok: false, summary: '生成结果格式错误，请稍后重试。'}};\n  }\n});"
      }
    },
    {
      "id": "Valid?",
      "name": "Valid?",
      "type": "n8n-nodes-base.if",
      "typeVersion": 2.2,
      "position": [
        720,
        0
      ],
      "parameters": {
        "conditions": {
          "options": {
            "caseSensitive": true,
            "leftValue": "",
            "typeValidation": "strict",
            "version": 2
          },
          "conditions": [
            {
              "id": "valid",
              "leftValue": "={{ $json.ok }}",
              "rightValue": true,
              "operator": {
                "type": "boolean",
                "operation": "true",
                "singleValue": true
              }
            }
          ],
          "combinator": "and"
        },
        "options": {}
      }
    },
    {
      "id": "Success",
      "name": "Success",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        960,
        -100
      ],
      "parameters": {
        "jsCode": "return $input.all().map(x => ({json: {...x.json, route: 'success'}}));"
      }
    },
    {
      "id": "Fallback",
      "name": "Fallback",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        960,
        100
      ],
      "parameters": {
        "jsCode": "return $input.all().map(x => ({json: {...x.json, route: 'fallback'}}));"
      }
    }
  ],
  "connections": {
    "Validate": {
      "main": [
        [
          {
            "node": "Valid?",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Valid?": {
      "main": [
        [
          {
            "node": "Success",
            "type": "main",
            "index": 0
          }
        ],
        [
          {
            "node": "Fallback",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Start": {
      "main": [
        [
          {
            "node": "Fixture",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Fixture": {
      "main": [
        [
          {
            "node": "Validate",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  },
  "settings": {
    "executionOrder": "v1"
  }
}

打开文件

{
  "name": "Stage6 Webhook Summary",
  "active": false,
  "nodes": [
    {
      "id": "Webhook",
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "typeVersion": 2,
      "position": [
        0,
        0
      ],
      "parameters": {
        "httpMethod": "POST",
        "path": "stage6-summary",
        "responseMode": "responseNode",
        "options": {}
      },
      "webhookId": "stage6-summary"
    },
    {
      "id": "Model",
      "name": "Model",
      "type": "n8n-nodes-base.httpRequest",
      "typeVersion": 4.2,
      "position": [
        240,
        0
      ],
      "parameters": {
        "method": "POST",
        "url": "https://api.openai.com/v1/responses",
        "authentication": "genericCredentialType",
        "genericAuthType": "httpHeaderAuth",
        "sendBody": true,
        "specifyBody": "json",
        "jsonBody": "={{ JSON.stringify({model: \"gpt-5.4-mini-2026-03-17\", input: \"Return JSON only with one string field summary. Summarize: \" + String($json.body.text || \"\"), max_output_tokens: 256, store: false}) }}",
        "options": {
          "timeout": 30000
        }
      },
      "onError": "continueRegularOutput"
    },
    {
      "id": "Validate",
      "name": "Validate",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        480,
        0
      ],
      "parameters": {
        "jsCode": "return $input.all().map(item => {\n  let raw = item.json.modelText;\n  if (raw === undefined) raw = (item.json.output || []).flatMap(x => x.content || []).filter(x => x.type === 'output_text').map(x => x.text).join('');\n  try {\n    const value = JSON.parse(raw);\n    if (!value || typeof value.summary !== 'string' || !value.summary.trim() || value.summary.length > 1000) throw new Error('summary must be a nonempty string <= 1000 chars');\n    return {json: {case: item.json.case || 'webhook', ok: true, summary: value.summary}};\n  } catch (_) {\n    return {json: {case: item.json.case || 'webhook', ok: false, summary: '生成结果格式错误，请稍后重试。'}};\n  }\n});"
      }
    },
    {
      "id": "Valid?",
      "name": "Valid?",
      "type": "n8n-nodes-base.if",
      "typeVersion": 2.2,
      "position": [
        720,
        0
      ],
      "parameters": {
        "conditions": {
          "options": {
            "caseSensitive": true,
            "leftValue": "",
            "typeValidation": "strict",
            "version": 2
          },
          "conditions": [
            {
              "id": "valid",
              "leftValue": "={{ $json.ok }}",
              "rightValue": true,
              "operator": {
                "type": "boolean",
                "operation": "true",
                "singleValue": true
              }
            }
          ],
          "combinator": "and"
        },
        "options": {}
      }
    },
    {
      "id": "Success",
      "name": "Success",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        960,
        -100
      ],
      "parameters": {
        "jsCode": "return $input.all().map(x => ({json: {...x.json, route: 'success'}}));"
      }
    },
    {
      "id": "Fallback",
      "name": "Fallback",
      "type": "n8n-nodes-base.code",
      "typeVersion": 2,
      "position": [
        960,
        100
      ],
      "parameters": {
        "jsCode": "return $input.all().map(x => ({json: {...x.json, route: 'fallback'}}));"
      }
    },
    {
      "id": "Reply",
      "name": "Reply",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1.4,
      "position": [
        1200,
        0
      ],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ JSON.stringify($json) }}",
        "options": {}
      }
    }
  ],
  "connections": {
    "Validate": {
      "main": [
        [
          {
            "node": "Valid?",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Valid?": {
      "main": [
        [
          {
            "node": "Success",
            "type": "main",
            "index": 0
          }
        ],
        [
          {
            "node": "Fallback",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Webhook": {
      "main": [
        [
          {
            "node": "Model",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Model": {
      "main": [
        [
          {
            "node": "Validate",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Success": {
      "main": [
        [
          {
            "node": "Reply",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Fallback": {
      "main": [
        [
          {
            "node": "Reply",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  },
  "settings": {
    "executionOrder": "v1"
  }
}

打开文件

# 6.6 n8n 2.44.7

Python 启动器只用标准库；安装并启动 Docker Desktop，然后运行：
```sh
docker pull docker.n8n.io/n8nio/n8n:2.44.7
python s66_n8n.py
```
实测镜像 digest：`sha256:942bc274709f27f99f7c6024a785d8e8f75ac06838cabcba448abb8d9127ec7b`。可使用 `docker.n8n.io/n8nio/n8n@sha256:…` 固定同一镜像。版本标签可跨平台拉取；本次在 macOS Apple Silicon / Docker Server 29.5.3 运行。

`n8n/offline.json` 在真实 n8n 引擎中执行 Manual Trigger → Fixture → Validate → IF → Success/Fallback。三条样本为合法 JSON、损坏 JSON、缺少 summary。Validate 不会因 JSON.parse 报错终止整个任务，IF 选择友好兜底。容器没有网络，资料只读挂载，数据库不持久化，不发送通知。

## Webhook 与真实模型

`n8n/webhook.json` 是可导入的在线版本：Webhook → Model（HTTP Request）→ Validate → IF → Reply。Validate/IF 与离线版本完全相同。在线版已做结构检查，尚未在已发布 webhook 上端到端验证。

1. 在 n8n 2.44.7 的编辑器 Import from File，导入 webhook.json。
2. Model 节点选择 Header Auth 凭据；Header 名为 `Authorization`，值为 `Bearer 你的模型密钥`。凭据保存在 n8n 凭据管理器，不写进 JSON。模型默认 `gpt-5.4-mini-2026-03-17`，可换为账号有权限且支持 Responses 的模型。
3. 点击 Webhook 的 Listen for test event。复制节点显示的 Test URL（仅测试监听期间有效），以 JSON 发送 `{"text":"学习站提供分阶段课程和本地进度。"}`。
```sh
curl -X POST 'http://localhost:5678/webhook-test/stage6-summary' \
  -H 'Content-Type: application/json' \
  -d '{"text":"学习站提供分阶段课程和本地进度。"}'
```
4. 正常应返回 `ok:true`、非空 summary、`route:"success"`；Model 节点请求失败、非 JSON 或缺字段时返回 `ok:false`、友好提示及 `route:"fallback"`。响应措辞不固定。检查执行记录确认失败发生的位置。
5. 需要长期使用才发布，并使用 Production URL；对外暴露前给 Webhook 配置认证。示例只回应请求，不接通知服务。

默认最多一次模型 HTTP 请求，30 秒超时，256 输出 tokens；这不是费用上限。在线模式会把输入发给模型服务。当前 prompt 要求 JSON 但不保证模型遵守，所以仍保留验证与兜底。

## 常见报错

- Docker daemon 无法连接：启动 Docker Desktop；确认当前用户可访问 Docker。
- Mounts denied：允许 Docker 访问解压目录；完整解压，不要只复制启动器。
- 镜像不存在：先 pull 固定版本；离线容器不负责下载模型。
- 120 秒超时：检查 Docker 资源和启动日志；不要把超时改成无限。
- 401：Model 节点没有绑定 Header Auth 或密钥无效。工作流仍应进入 fallback。
- Webhook 404：测试 URL 需要监听；生产 URL 需要发布。
- 编辑器节点类型不支持：确认 n8n 2.44.7，旧版本不在本次验证范围。

官方 CLI：https://docs.n8n.io/hosting/cli-commands/

打开文件

"""实际 GraphRAG CLI 初始化/配置检查。默认不建索引、不调用模型。"""
import argparse
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parent


def prepare(path):
    if path.exists():
        raise ValueError('目标已存在：请指定新目录，避免覆盖已有配置和密钥')
    path.mkdir(parents=True)
    cli = Path(sys.executable).with_name('graphrag.exe' if os.name == 'nt' else 'graphrag')
    env = dict(os.environ, LITELLM_LOCAL_MODEL_COST_MAP='True', GRAPHRAG_API_KEY='offline-placeholder')
    commands = [[str(cli), 'init', '--root', str(path), '--model', 'gpt-5.4-mini-2026-03-17',
                 '--embedding', 'text-embedding-3-small'],
                [str(cli), 'index', '--root', str(path), '--dry-run', '--skip-validation']]
    for i, cmd in enumerate(commands):
        proc = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=90)
        if proc.returncode:
            raise RuntimeError('GraphRAG CLI 失败：\n' + (proc.stderr + proc.stdout)[-2000:])
        if i == 0:
            shutil.copytree(ROOT/'data/articles', path/'input', dirs_exist_ok=True)
    if not (path/'settings.yaml').is_file() or len(list((path/'input').glob('*.txt'))) != 5:
        raise RuntimeError('项目材料不完整')
    print('GraphRAG 3.3.0：init 完成；5 篇教学资料已复制')
    print('index --dry-run --skip-validation：配置检查通过')
    print('模型请求=0；未生成实体、社区报告或向量索引')


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--prepare', type=Path, help='保留项目供真实 API 实验，目标必须不存在')
    args = parser.parse_args()
    try:
        if args.prepare:
            prepare(args.prepare.resolve())
            print('下一步：按 GRAPHRAG.md 配置密钥，再真实 index/query。')
        else:
            with tempfile.TemporaryDirectory(prefix='stage6-graphrag-') as tmp:
                prepare(Path(tmp)/'project')
            print('临时项目已清理；真实检索质量与费用尚未测量')
    except (ValueError, RuntimeError, FileNotFoundError, subprocess.TimeoutExpired) as exc:
        parser.exit(1, str(exc) + '\n')

打开文件

graphrag==3.3.0

打开文件

# 6.7 GraphRAG 3.3.0

使用 Python 3.11 独立环境，依赖见 requirements/graph-lock.txt。5 篇中文教学短文在 data/articles/；短文足以练习流程，但不是验证 GraphRAG 优势的基准数据。

## 无密钥检查（已实跑）
```sh
python s67_graphrag.py
```
真正调用官方 CLI 的 init 和 index --dry-run --skip-validation。临时项目随后清理。skip-validation 跳过模型连接检查，dry-run 不建索引；成功仅代表配置通过，不代表 API 可用。脚本给子进程注入离线占位值，不读取或覆盖你的模型密钥。LiteLLM 使用随包价格表，避免启动时联网获取价格。

## 建立真实索引（需密钥，未在线实测）
```sh
python s67_graphrag.py --prepare .runtime/graphrag
```
已有目录时脚本拒绝覆盖。用编辑器打开新项目的 `.runtime/graphrag/.env`，把生成的 `GRAPHRAG_API_KEY` 填为自己的密钥；不要上传该文件。检查 settings.yaml 的 completion_models 与 embedding_models：默认模型分别为 gpt-5.4-mini-2026-03-17 和 text-embedding-3-small，账号必须有权限。再执行：
```sh
graphrag index --root .runtime/graphrag
graphrag query --root .runtime/graphrag --method global --query "这些材料共同关注哪些目标和风险？"
graphrag query --root .runtime/graphrag --method global --query "团队如何分工保障发布质量？"
graphrag query --root .runtime/graphrag --method local --query "李明负责什么，并与谁协作？"
graphrag query --root .runtime/graphrag --method local --query "导航学习站的进度能跨设备同步吗？"
```
命令需在 `.venv-graph` 环境内执行，或将 graphrag 换为该环境的完整路径。索引会对资料调用模型和 embedding，耗时与费用取决于模型、重试、分块和提取设置。先用这 5 篇小材料观察日志；本例没有预设美元费用，也没有宣称真实建索引已通过。

## 同数据、同问题比较普通 RAG

完成同一索引后，对上述四问分别改用 `--method basic`，这是官方提供的基础向量检索查询，可保持相同语料与 embedding，避免拿不同资料比较。
```sh
graphrag query --root .runtime/graphrag --method basic --query "团队如何分工保障发布质量？"
```
注意：basic 在已建好的 GraphRAG 项目上查询；本次总索引费用包含图提取，不能当作普通 RAG 独立索引成本。要比较完整系统成本，需另测只做分块与 embedding 的基线。

预期事实（不是承诺的模型原文）：

| 问题 | 应覆盖的证据 | 验收 |
| --- | --- | --- |
| 共同目标与风险 | 01 降低入门成本；02 可复现性；04 失败覆盖；05 费用与生产监测 | 全局问题应整合多篇，不只复述第一篇 |
| 分工保障质量 | 01、04、05：李明课程、王悦平台、陈晨评审、交叉核对 | 关系不混淆，可定位来源 |
| 李明的职责 | 02、04、05：课程、修复代码、与陈晨核对 | 局部实体相关事实完整 |
| 跨设备进度 | 03：仅当前浏览器、不自动同步 | 不因图检索补出不存在的云端功能 |

逐问记录方法、答案、引用、漏项、耗时、请求数和实际 token 用量；区分索引一次性成本与每次查询成本，费用按服务账单核对。GraphRAG 在这个小数据集上不一定更好，不预填任何质量或费用结论。

常见报错：

- `No module named graphrag` / CLI 找不到：使用 .venv-graph 及对应锁定依赖。
- 配置缺少 completion_models：不要沿用 2.x 的 models 配置；用本版本 init 生成。
- 找不到 input：运行 prepare 后确认有 5 个 .txt；单独 init 不会自动提供资料。
- 401/404/429：分别核对密钥、模型权限/名称、额度与速率；dry-run 不会发现这些问题。
- 缺少 entities / communities / community_reports 等表：先确认真实 index 成功，dry-run 不产生这些文件。
- 没有提取出实体：检查输入、extract_graph 的 entity_types 与语言提示，查看 logs，不能以空图结果下质量结论。

官方说明：https://microsoft.github.io/graphrag/get_started/ 及 https://microsoft.github.io/graphrag/cli/

打开文件

星河学习项目概览
星河团队开发了导航学习站，用于 Agent 入门课程。李明负责课程，王悦负责平台。项目共同目标是降低初学者第一次运行示例的成本。

打开文件

课程与可复现性
李明维护导航学习站的 Python 课程。课程提供离线示例、固定依赖和预期输出。离线示例与固定依赖共同降低环境差异造成的失败；固定模型响应不能衡量真实模型能力。

打开文件

平台与进度
王悦维护导航学习站的平台。平台有搜索、移动端目录和本地学习进度。本地进度保存在当前浏览器；平台没有账号，也不提供跨设备自动同步。

打开文件

评审与风险
陈晨评审导航学习站的课程和平台。李明负责修复课程代码，王悦负责处理界面问题。评审要求同时覆盖成功、失败和停止条件；引用与答案必须能追溯到资料。

打开文件

发布计划与限制
星河团队计划先发布离线练习，再开放真实模型练习。李明与陈晨核对依赖和输出，王悦检查移动端目录。真实模型调用可能产生费用；测试数据上的成功不能替代生产监测。