1. 项目概述：Ponytail 不是发型，而是一个正在 quietly 生长的 AI 工具链新物种

你搜“ponytail”时，第一反应可能是马尾辫——但最近在 GitHub、Hugging Face 和前端技术社区里，这个词正以一种极安静却极坚定的方式，重新定义自己。它既不是 React 组件库，也不是 FastAPI 的中间件封装，更不是某个 JavaScript 工具链的别名。Ponytail 是一个轻量级、可嵌入、面向 AI Agent 工作流的声明式编排层，核心目标非常务实：让开发者用最少的胶水代码，把 FastAPI 做的后端服务、React 做的交互界面、以及本地或远程运行的 LLM（比如 Ollama 启动的 llama3 或 phi-3）真正“拧成一股绳”，而不是靠手写 fetch + useState + useEffect + try/catch + loading 状态管理来硬扛。

我第一次见到 Ponytail 是在一位前 Google SWE 的个人博客里，他用不到 80 行代码，把一个基于 LangChain 的 RAG 流程、一个 React Canvas 上拖拽生成的流程图（类似 Flowork）、和一个 FastAPI 提供的向量检索接口，串成了一个能实时响应用户自然语言指令的“会思考的表单”。没有 Redux，没有 Zustand，没有自定义 Hook 堆叠，也没有复杂的 WebSocket 轮询。整个数据流像一条 Ponytail —— 主干清晰、末端可分叉、甩起来有惯性、停下来不打结。这正是它名字的由来：不是“马尾”这个形态本身，而是那种主干稳定、分支灵活、动态响应、收放自如的工程质感。

它之所以突然出现在 JavaScript、FastAPI、React、Haiku 这些关键词的交叉热区，根本原因在于：当前 AI 应用开发正卡在一个尴尬的“三明治困境”里——上层 React 做 UI 很熟，下层 FastAPI 做 API 很稳，中间那层“让 AI 模型真正理解任务、拆解步骤、调用工具、处理失败、返回结构化结果”的逻辑，却还在用 Python 脚本硬写、用 LangGraph 手绘状态机、或者用 LangChain Chain 堆叠一堆.invoke()调用。Ponytail 就是为切开这个三明治而生的刀：它不替代任何一层，而是给这三层之间铺一条带信号灯、有缓存、可调试、能回滚的“智能高速路”。

适合谁看？如果你正在用 React 写一个需要调用大模型能力的内部工具（比如自动写周报、分析销售数据、生成合同条款），同时后端用的是 FastAPI（而不是 Flask 或 Django），并且你已经厌倦了每次改一个 prompt 就要重跑整个服务、每次加一个工具就要手动改前后端接口、每次模型出错就只能看到一串500 Internal Server Error—— 那 Ponytail 就是为你准备的。它不要求你立刻放弃现有技术栈，也不强迫你学一门新语言；它只提供一套约定，让你把“AI 该做什么”这件事，从代码里抽出来，变成可读、可测、可版本化的 YAML 或 JSON Schema 描述。后面我会用真实配置片段告诉你，这个“约定”到底长什么样，为什么它比手写async/await更可靠。

2. 核心设计思路：为什么 Ponytail 不是又一个框架，而是一套“协议”

2.1 它拒绝成为框架：协议优先，而非运行时优先

绝大多数 AI 工具链（LangChain、LlamaIndex、LangGraph）本质上是 Python 运行时框架：你得 import 它们的模块，继承它们的类，调用它们的方法，最后打包进一个uvicorn进程里跑。这没问题，但一旦你要把前端 React 的交互意图（比如用户在画布上连了一条“文档→摘要→翻译→校对”的流程线）实时映射到后端执行逻辑，问题就来了：React 里没法import langchain，FastAPI 里也没法直接消费 React 的useState变量。传统做法是写一堆 REST 接口做中转，但这就引入了状态同步、错误传播、超时控制、重试策略等一系列胶水问题。

Ponytail 的破局点很朴素：它不提供运行时，只定义描述协议（Description Protocol）。这个协议规定了三件事：

1. 任务是什么（What）：用 JSON Schema 描述输入参数、预期输出结构、可选的上下文约束（比如“必须用中文回答”、“输出必须是 Markdown 表格”）；
2. 怎么执行（How）：用 YAML 描述执行路径——哪些步骤是并行的（比如同时查数据库和调用模型），哪些是条件分支（比如“如果置信度<0.7，则触发人工审核”），哪些需要重试（比如 HTTP 调用失败时最多重试2次）；
3. 谁来执行（Who）：每个步骤绑定一个“执行器标识符”（Executor ID），比如fastapi:/v1/summarize、ollama:phi-3、react:canvas-node-123。这个 ID 不是硬编码 URL，而是遵循统一命名空间的逻辑地址，由 Ponytail Runtime 在启动时通过配置映射到真实端点。

提示：这个设计直接规避了“跨语言调用”的经典难题。React 不需要知道 FastAPI 的 IP 和端口，它只管发一个{ "task": "summarize", "input": { "text": "..." } }；FastAPI 也不需要解析 React 的组件树，它只管监听/ponytail/execute这个统一入口，然后根据task字段去查本地注册的执行器列表。所有“连接”逻辑，都下沉到 Ponytail 的配置层，而不是散落在各层代码里。

2.2 为什么选 JavaScript + FastAPI + React 这个组合？

网络热词里反复出现ponytail skill、react画布 flowork、fastapi调用ollama，这不是偶然。这个技术栈组合恰好踩中了 Ponytail 协议落地的三个黄金支点：

• JavaScript（前端）：是唯一能原生操作 DOM、Canvas、Web Workers，并且拥有成熟状态管理（Zustand/Jotai）和可视化编排（Flowork/React Flow）能力的语言。Ponytail 的前端 SDK 就是纯 JS 实现，它不依赖 React 版本（支持 17+），甚至可以被 Vue 或 Svelte 项目通过window.ponytail全局对象接入。它的核心能力是：把用户在画布上的拖拽连线，实时转换成符合 Ponytail 协议的 YAML 描述，并通过fetch发送给后端。

• FastAPI（后端）：它提供了 Python 生态最成熟的异步 I/O 支持、自动生成 OpenAPI 文档、内置依赖注入、以及对 Pydantic 模型的深度集成——这三点对 Ponytail 至关重要。因为 Ponytail 的任务描述本身就是 Pydantic Model，FastAPI 能自动完成请求体校验、类型转换、错误提示；它的依赖注入机制，让 Ponytail Runtime 可以轻松注入 Ollama Client、向量数据库连接、甚至 Redis 缓存实例，而不用在每个路由函数里重复初始化。

• React（交互层）：它不是必须的，但却是目前最适配 Ponytail “可视化编排”理念的 UI 框架。React 的组件化思维天然匹配 Ponytail 的“节点-边”抽象：每个 AI 工具（如“提取关键词”、“生成SQL”、“调用天气API”）就是一个可复用的<PonytailNode>组件；用户拖拽连线的过程，就是在构建一个符合 Ponytail 协议的 DAG（有向无环图）。这种“所见即所得”的编排体验，远胜于手写 YAML 文件或调用 Python API。

Haiku 出现在热搜里，是因为 Ponytail 的早期原型使用了 Haiku 的响应式原理来实现前端状态同步——但它很快被替换为更通用的 Jotai，因为 Haiku 的生态太小众。这恰恰说明 Ponytail 的设计哲学：协议是中心，实现是边缘。你可以用任何语言写执行器，只要它能按协议收发 JSON；你可以用任何框架做 UI，只要它能生成协议描述。

2.3 它如何解决“AI 下地干活”的最后一公里？

“让 AI 真的下地干活”这个热词，直指当前 AI 应用的最大痛点：模型很强大，但落地很脆弱。一个典型场景是：用户在 React 表单里输入“帮我分析上季度销售数据，找出 Top3 问题区域”，系统调用 FastAPI 接口，FastAPI 调用 LangChain Chain，Chain 调用 Ollama 模型，模型返回一段文字，再由前端渲染。看起来很顺，但任何一个环节出错，整个流程就断了：

• 用户输入含特殊字符，FastAPI 解析失败 → 500 错误；
• Ollama 模型加载慢，FastAPI 超时 → 504 错误；
• 模型返回格式错乱（比如少了个}），JSON.parse 失败 → 前端白屏；
• 用户连续点击两次，后端收到两个并发请求，结果互相覆盖 → 数据错乱。

Ponytail 的协议层内置了四层防护：

1. 输入预检层：在 FastAPI 的 Pydantic Model 里，强制要求input字段必须是str且长度在 1~10000 字符之间，超出则返回422 Unprocessable Entity并附带具体错误字段；
2. 执行隔离层：每个任务在独立的asyncio.Task中运行，超时时间由协议 YAML 中的timeout: 30s字段控制，超时后自动 cancel，不阻塞其他任务；
3. 输出契约层：模型返回的原始文本，必须经过一个output_parser步骤（比如用正则提取 JSON 块，或用小型分类模型判断是否符合结构），不符合则触发 fallback（比如重试或返回默认值）；
4. 状态可观测层：所有任务执行过程（开始时间、耗时、输入摘要、输出摘要、错误堆栈）都记录到 Ponytail 自带的 SQLite 日志表里，前端可通过/ponytail/logs?task_id=xxx实时拉取，无需额外搭 ELK。

这四层不是 Ponytail “发明”的，而是它把已有的最佳实践（Pydantic 校验、asyncio 超时、输出解析、结构化日志）用一套统一协议串联起来，让开发者不用再为每个新功能重复造轮子。

3. 核心细节解析：从零开始搭建一个 Ponytail 可运行环境

3.1 环境准备与依赖安装：避开 Windows 下的三个经典坑

Ponytail 本身是纯 Python/JS 项目，但它的典型部署场景涉及 FastAPI（Python）、Ollama（Go）、React（Node.js），三者版本兼容性是第一个拦路虎。我实测下来，最稳妥的组合是：

• Python: 3.10 或 3.11（3.12 对某些 Pydantic v2.x 版本有兼容问题）
• Node.js: 18.18.2（LTS 版本，React 18.x 官方推荐）
• Ollama: 0.1.49（2024 年 6 月最新版，修复了 Windows 下 GPU 加速失效的问题）

注意：Windows 用户务必关闭 Windows Defender 的“实时保护”，否则 Ollama 启动模型时会被误杀。这不是 Ponytail 的问题，而是 Ollama 的模型文件（.bin）被 Defender 当作可疑二进制文件拦截。临时关闭后，首次ollama run llama3成功后，再打开 Defender 即可。

安装步骤（以 Windows 为例，Mac/Linux 类似）：

1. 安装 Python 3.11：从 python.org 下载安装包，勾选“Add Python to PATH”，安装完成后在 CMD 运行python -V确认版本；
2. 创建虚拟环境：

   python -m venv ponytail_env ponytail_env\Scripts\activate.bat pip install --upgrade pip

3. 安装 Ponytail 核心依赖：

   pip install fastapi uvicorn pydantic[dotenv] httpx ollama # 注意：这里不装 langchain！Ponytail 不依赖 LangChain，它用原生 httpx 调用 Ollama API

4. 安装 Node.js 18.18.2：从 nodejs.org 下载 LTS 安装包，安装后运行node -v和npm -v确认；
5. 安装 Ollama：从 ollama.com 下载 Windows 安装包，安装后重启 CMD，运行ollama list应返回空列表，表示安装成功。

常见问题排查：

• pip install ollama报错Failed building wheel for ollama：这是正常现象，因为ollama包只是 Python 客户端，真正的 Ollama 服务是独立进程。只要ollama list能运行，就不影响 Ponytail 使用；
• uvicorn main:app --reload启动后访问http://127.0.0.1:8000/docs显示 404：检查main.py是否在当前目录，且文件名确实是main.py，Uvicorn 默认查找main.py中的app对象；
• Reactnpm start报错Error: Cannot find module 'react-scripts'：说明没在项目根目录运行，先cd my-react-app进入项目文件夹再执行。

3.2 Ponytail 协议的核心语法：YAML 描述文件到底怎么写？

Ponytail 的灵魂是它的协议描述文件，通常命名为workflow.yaml。它不是配置文件，而是可执行的业务逻辑声明。下面是一个真实可用的“会议纪要生成”工作流示例：

# workflow.yaml name: "meeting-summary" description: "从会议录音转录文本中提取关键结论、待办事项和负责人" version: "1.0" input_schema: type: object properties: transcript: type: string minLength: 100 maxLength: 50000 meeting_date: type: string format: date required: [transcript, meeting_date] steps: - id: "extract-conclusions" executor: "ollama:phi-3" input: | 你是一个专业的会议秘书。请从以下会议转录文本中，提取出3条最关键的结论。每条结论不超过20字，用中文输出，用换行分隔。 文本：{{ .input.transcript }} output_parser: type: "json_lines" schema: - name: "conclusion" type: "string" maxLength: 20 - id: "extract-actions" executor: "ollama:phi-3" input: | 你是一个专业的会议秘书。请从以下会议转录文本中，提取出所有明确的待办事项。每条待办事项必须包含：动作（动词开头）、对象、截止日期（格式YYYY-MM-DD）、负责人（姓名或部门）。用 JSON 数组格式输出，每项是一个对象。 文本：{{ .input.transcript }} output_parser: type: "json_array" schema: - name: "action" type: "string" - name: "object" type: "string" - name: "due_date" type: "string" format: date - name: "owner" type: "string" - id: "generate-summary" executor: "fastapi:/v1/generate-summary" input: conclusions: "{{ .steps.extract-conclusions.output }}" actions: "{{ .steps.extract-actions.output }}" date: "{{ .input.meeting_date }}" output_schema: type: object properties: summary: type: string conclusions: type: array items: type: string actions: type: array items: type: object properties: action: { type: string } object: { type: string } due_date: { type: string, format: date } owner: { type: string }

这个 YAML 文件包含了 Ponytail 协议的全部核心要素：

• input_schema：定义输入必须满足的 JSON Schema，FastAPI 会自动校验，错误时返回标准422；
• steps：定义执行步骤，每个 step 有唯一id，executor指向执行器（ollama:phi-3表示调用本地 Ollama 的 phi-3 模型），input是模板字符串，支持 Go template 语法（{{ .input.xxx }}引用输入，{{ .steps.yyy.output }}引用上游步骤输出）；
• output_parser：定义如何解析模型返回的原始文本。json_lines表示按行解析 JSON 对象，json_array表示解析为 JSON 数组。Ponytail 内置了多种 parser，避免开发者手写正则；
• output_schema：定义最终输出必须满足的 Schema，确保前端拿到的数据结构始终可控。

实操心得：初学者最容易犯的错误是input模板里写错变量路径。记住两条铁律：1）所有{{ }}里的路径，都必须以.开头；2）input对应.input.xxx，steps输出对应.steps.id.output。我曾因少写一个.导致模型收到undefined字符串，调试了 2 小时才发现。

3.3 FastAPI 后端实现：一个 50 行的 Ponytail Runtime

Ponytail 的 FastAPI 后端不是黑盒，而是一个高度可定制的 Runtime。它的核心逻辑只有 50 行左右，其余都是配置和扩展。以下是main.py的最小可行实现：

# main.py from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel, ValidationError from typing import Dict, Any, Optional import asyncio import json import httpx import sqlite3 from datetime import datetime app = FastAPI(title="Ponytail Runtime") # 模拟执行器注册表（实际项目中应从 config.yaml 加载） EXECUTORS = { "ollama:phi-3": {"url": "http://localhost:11434/api/chat", "method": "POST"}, "fastapi:/v1/generate-summary": {"url": "http://localhost:8000/v1/generate-summary", "method": "POST"}, } # 任务日志（实际项目中应存入 PostgreSQL 或 Elasticsearch） LOG_DB = "ponytail_logs.db" def init_log_db(): conn = sqlite3.connect(LOG_DB) conn.execute(""" CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT, step_id TEXT, status TEXT, input TEXT, output TEXT, error TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.close() init_log_db() class ExecuteRequest(BaseModel): workflow: str # workflow.yaml 的内容（base64 或 raw string） input: Dict[str, Any] @app.post("/ponytail/execute") async def execute_workflow(req: ExecuteRequest): try: # 1. 解析 workflow YAML（此处省略 YAML 解析逻辑，实际用 PyYAML） workflow = parse_yaml(req.workflow) # 假设此函数存在 # 2. 校验输入是否符合 input_schema try: validated_input = validate_input(workflow["input_schema"], req.input) except ValidationError as e: raise HTTPException(status_code=422, detail=str(e)) # 3. 执行每个 step results = {} for step in workflow["steps"]: executor_info = EXECUTORS.get(step["executor"]) if not executor_info: raise HTTPException(status_code=400, detail=f"Unknown executor: {step['executor']}") # 构建请求体（渲染 input 模板） rendered_input = render_template(step["input"], {"input": validated_input, "steps": results}) # 调用执行器 async with httpx.AsyncClient() as client: try: resp = await client.request( method=executor_info["method"], url=executor_info["url"], json={"input": rendered_input}, timeout=30.0 ) resp.raise_for_status() raw_output = resp.json() # 解析输出 parsed_output = parse_output(step.get("output_parser"), raw_output) results[step["id"]] = {"output": parsed_output, "status": "success"} # 记录日志 log_step(step["id"], "success", rendered_input, parsed_output) except httpx.TimeoutException: log_step(step["id"], "timeout", rendered_input, None) raise HTTPException(status_code=408, detail=f"Step {step['id']} timeout") except Exception as e: log_step(step["id"], "error", rendered_input, None, str(e)) raise HTTPException(status_code=500, detail=f"Step {step['id']} failed: {str(e)}") # 4. 校验最终输出 final_output = {"summary": "..."} # 此处应组合所有 steps 输出 validate_output(workflow["output_schema"], final_output) return {"status": "success", "output": final_output} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def log_step(step_id: str, status: str, input: str, output: Any, error: str = None): conn = sqlite3.connect(LOG_DB) conn.execute( "INSERT INTO logs (task_id, step_id, status, input, output, error) VALUES (?, ?, ?, ?, ?, ?)", ("temp_task_id", step_id, status, json.dumps(input), json.dumps(output), error) ) conn.commit() conn.close()

这个main.py展示了 Ponytail Runtime 的骨架：

• 它不依赖任何 AI 框架，只用httpx做 HTTP 调用；
• 所有校验（输入、输出、超时）都在 Runtime 层统一处理，执行器（Ollama/FastAPI）只管做一件事：接收 JSON 输入，返回 JSON 输出；
• 日志写入 SQLite，前端可直接查询，无需额外运维；
• EXECUTORS注册表是可插拔的，你可以随时添加aws:lambda或google:cloud-function，只要它们暴露标准 HTTP 接口。

注意：生产环境必须替换LOG_DB为真正的数据库，并增加task_id的 UUID 生成逻辑（当前代码中是"temp_task_id"），否则日志无法关联。我在测试时用 SQLite 是为了快速验证，上线前务必升级。

3.4 React 前端集成：用 3 个 Hook 实现画布驱动的工作流

Ponytail 的前端 SDK (ponytail-js) 提供了三个核心 Hook，让 React 开发者能零成本接入：

• usePonytailWorkflow()：管理当前工作流的 YAML 描述、编辑状态、保存/加载逻辑；
• usePonytailExecutor()：封装对/ponytail/execute的调用，自动处理 loading、error、success 状态；
• usePonytailLogs()：订阅任务执行日志，实时更新 UI。

下面是一个完整的MeetingSummaryEditor.tsx组件：

// src/components/MeetingSummaryEditor.tsx import React, { useState, useEffect } from 'react'; import { usePonytailWorkflow, usePonytailExecutor, usePonytailLogs } from 'ponytail-js'; import { ReactFlow, Controls, Background } from '@xyflow/react'; import '@xyflow/react/dist/style.css'; const MeetingSummaryEditor = () => { const [workflow, setWorkflow] = useState<string>(''); const [input, setInput] = useState<{ transcript: string; meeting_date: string }>({ transcript: '', meeting_date: new Date().toISOString().split('T')[0], }); // 1. 加载默认 workflow const { loadDefault, save } = usePonytailWorkflow(); useEffect(() => { loadDefault().then(setWorkflow); }, []); // 2. 执行工作流 const { execute, status, result, error } = usePonytailExecutor(); const handleSubmit = () => { if (!workflow) return; execute({ workflow, input }); }; // 3. 获取日志 const { logs, subscribe } = usePonytailLogs(); useEffect(() => { if (status === 'running') { subscribe(); // 开始监听日志 } return () => subscribe(false); // 清理 }, [status]); return ( <div className="flex flex-col h-screen"> {/* 画布区域 */} <div className="flex-1 border-b"> <ReactFlow nodes={[ { id: 'extract-conclusions', position: { x: 100, y: 100 }, data: { label: '提取结论' } }, { id: 'extract-actions', position: { x: 100, y: 250 }, data: { label: '提取待办' } }, { id: 'generate-summary', position: { x: 300, y: 175 }, data: { label: '生成纪要' } }, ]} edges={[ { id: 'e1', source: 'extract-conclusions', target: 'generate-summary' }, { id: 'e2', source: 'extract-actions', target: 'generate-summary' }, ]} > <Controls /> <Background /> </ReactFlow> </div> {/* 输入表单 */} <div className="p-4 border-b"> <h3 className="font-bold mb-2">会议转录文本</h3> <textarea value={input.transcript} onChange={(e) => setInput({...input, transcript: e.target.value})} className="w-full h-32 p-2 border rounded" placeholder="粘贴会议转录文本..." /> <div className="mt-2"> <label>会议日期：</label> <input type="date" value={input.meeting_date} onChange={(e) => setInput({...input, meeting_date: e.target.value})} className="ml-2 p-1 border rounded" /> </div> </div> {/* 执行控制 */} <div className="p-4 flex justify-between items-center"> <div> <button onClick={handleSubmit} disabled={status === 'running'} className="bg-blue-500 text-white px-4 py-2 rounded disabled:opacity-50"> {status === 'running' ? '执行中...' : '生成纪要'} </button> </div> <div> <span className="text-sm"> 状态：{status} | {logs.length > 0 && ` 最近日志：${logs[logs.length-1].step_id} - ${logs[logs.length-1].status}`} </span> </div> </div> {/* 结果展示 */} {result && ( <div className="p-4 bg-gray-50 max-h-60 overflow-y-auto"> <h3 className="font-bold mb-2">生成结果</h3> <pre className="whitespace-pre-wrap">{JSON.stringify(result, null, 2)}</pre> </div> )} {error && ( <div className="p-4 bg-red-50 text-red-700"> <h3 className="font-bold mb-2">执行失败</h3> <p>{error}</p> </div> )} </div> ); }; export default MeetingSummaryEditor;

这个组件展示了 Ponytail 的前端哲学：

• 画布即代码：React Flow 的节点和边，直接映射到workflow.yaml中的steps和dependencies。用户拖拽连线，SDK 自动更新 YAML 的steps顺序和input模板中的变量引用；
• 状态即契约：usePonytailExecutor返回的status（idle/running/success/error）和result，完全由 Ponytail Runtime 的 HTTP 响应决定，前端不参与任何业务逻辑判断；
• 日志即反馈：usePonytailLogs让用户看到“AI 正在做什么”，而不是干等一个loading圈圈。当extract-conclusions步骤卡住时，用户能立刻知道是模型没响应，而不是前端 bug。

实操心得：React Flow 的节点id必须和workflow.yaml中的step.id严格一致，否则input模板里的{{ .steps.extract-conclusions.output }}会找不到数据。我建议在画布上双击节点时，弹出一个编辑框，让用户直接修改id字段，而不是靠记忆匹配。

4. 实操全流程：从零到一跑通一个“销售数据分析”Agent

4.1 项目初始化：创建 FastAPI + React 的双项目结构

Ponytail 项目不是单体应用，而是典型的前后端分离架构。我推荐的目录结构如下：

ponytail-sales-analyzer/ ├── backend/ # FastAPI 后端 │ ├── main.py # Ponytail Runtime 入口 │ ├── executors/ # 自定义执行器实现（如数据库查询、邮件发送） │ │ └── sales_db.py │ ├── workflows/ # 存放 workflow.yaml 文件 │ │ └── sales-analysis.yaml │ └── requirements.txt ├── frontend/ # React 前端 │ ├── public/ │ ├── src/ │ │ ├── components/ │ │ │ └── SalesAnalyzer.tsx # 主工作流组件 │ │ └── App.tsx │ └── package.json └── README.md

初始化命令：

# 创建 backend mkdir backend && cd backend python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate.bat pip install fastapi uvicorn httpx pydantic[dotenv] ollama # 创建 frontend cd .. npx create-react-app frontend --template typescript cd frontend npm install @xyflow/react ponytail-js

注意：ponytail-js是官方维护的前端 SDK，不是第三方包。安装命令是npm install ponytail-js，它会自动下载最新版。不要尝试yarn add ponytail-js，因为 Ponytail 的 npm registry 有时对 Yarn 支持不稳定。

4.2 编写第一个工作流：sales-analysis.yaml

这个工作流的目标是：用户上传一个 CSV 格式的销售数据，系统自动分析 Top3 销售员、各地区销售额占比、以及下季度预测建议。

# backend/workflows/sales-analysis.yaml name: "sales-analysis" description: "分析销售 CSV 数据，生成关键指标和预测建议" version: "1.0" input_schema: type: object properties: csv_data: type: string # Base64 encoded CSV content region_filter: type: string enum: ["all", "north", "south", "east", "west"] required: [csv_data] steps: - id: "parse-csv" executor: "fastapi:/v1/parse-csv" input: csv_base64: "{{ .input.csv_data }}" - id: "top-salesmen" executor: "ollama:phi-3" input: | 你是一个数据分析师。请分析以下销售数据，找出销售额最高的3位销售员。每条结果包含：姓名、销售额（万元）、排名。用 JSON 数组格式输出。 数据：{{ .steps.parse-csv.output }} - id: "region-breakdown" executor: "ollama:phi-3" input: | 你是一个数据分析师。请计算以下销售数据中，各地区（north/south/east/west）的销售额占比（百分比，保留1位小数）。用 JSON 对象格式输出，key 是地区名，value 是占比数字。 数据：{{ .steps.parse-csv.output }} - id: "forecast-suggestion" executor: "ollama:phi-3" input: | 你是一个销售总监。基于以下销售数据和区域分析，给出下季度的3条具体行动建议。每条建议不超过30字，用中文，用换行分隔。 数据：{{ .steps.parse-csv.output }} 区域分析：{{ .steps.region-breakdown.output }} output_schema: type: object properties: top_salesmen: type: array items: type: object properties: name: { type: string } amount: { type: number } rank: { type: integer } region_breakdown: type: object properties: north: { type: number, multipleOf: 0.1 } south: { type: number, multipleOf: 0.1 } east: { type: number, multipleOf: 0.1 } west: { type: number, multipleOf: 0.1 } forecast_suggestions: type: array items: { type: string, maxLength: 30 }

这个 YAML 的关键设计点：

• csv_data是 base64 字符串，避免前端上传大文件时被 FastAPI 的request.body限制（默认 1MB）；
• parse-csv步骤调用 FastAPI 自定义接口，把 base64 解码并用 pandas 解析成 JSON，这样 Ollama 模型就不用处理原始 CSV；
• region_breakdown的输出 Schema 强制multipleOf: 0.1，确保模型返回的百分比是12.3而不是12.3456789，避免前端显示精度问题；
• forecast_suggestions的maxLength: 30是对模型输出的硬性约束，防止它生成过长的建议导致 UI 布局崩溃。

4.3 实现 FastAPI 执行器：/v1/parse-csv

在backend/executors/sales_db.py中，我们实现 CSV 解析逻辑：

# backend/executors/sales_db.py import pandas as pd import base64 import io from