34 KiB
开发低代码智能体平台 - 完整开发提示词
📋 项目概述
你是一位资深的全栈开发专家,负责开发一个工作流和Agent低代码智能体平台。该平台需要支持可视化工作流设计、智能Agent配置、低代码开发等功能。
🏗️ 技术架构
核心技术栈
前端技术栈:
-
框架: Vue 3 + TypeScript + Vite
-
状态管理: Pinia
-
UI组件库: Element Plus / Ant Design Vue
-
工作流可视化: Vue Flow
-
HTTP客户端: Axios
-
WebSocket: Socket.io Client
-
路由: Vue Router
-
表单验证: VeeValidate
-
代码编辑器: Monaco Editor
后端技术栈:
-
API框架: Python FastAPI
-
任务队列: Celery + Redis
-
缓存/队列: Redis
-
数据库: PostgreSQL
-
ORM: SQLAlchemy
-
数据验证: Pydantic
-
Agent框架: LangChain
-
RAG框架: LlamaIndex
-
向量数据库: ChromaDB
-
WebSocket: FastAPI WebSocket
-
API文档: Swagger/OpenAPI
基础设施:
-
容器化: Docker + Docker Compose
-
反向代理: Nginx
-
进程管理: Supervisor / systemd
-
监控: Prometheus + Grafana
-
日志: ELK Stack (可选)
-
CI/CD: GitHub Actions / GitLab CI
🎯 核心功能需求
1. 低代码工作流设计器
功能要求:
-
可视化拖拽式工作流编辑器
-
支持多种节点类型(开始、模板、条件、数据、工具、结束)
-
节点间连线配置
-
节点参数配置面板
-
工作流预览和验证
-
工作流版本管理
-
工作流模板库
技术实现:
-
使用 Vue Flow 实现可视化编辑器
-
支持节点拖拽、连线、删除、复制等操作
-
节点配置表单动态生成
-
工作流JSON格式存储
-
实时验证工作流逻辑
2. 智能Agent平台
功能要求:
-
Agent角色定义和配置
-
Agent工具链管理
-
Agent工作流编排
-
多Agent协作支持
-
Agent执行监控
-
Agent性能分析
技术实现:
-
集成 LangChain Agent框架
-
支持多种Agent类型(ReAct、Plan-and-Execute等)
-
工具注册和管理系统
-
Agent执行状态实时推送(WebSocket)
-
执行日志和结果存储
3. 工作流执行引擎
功能要求:
-
异步任务执行(Celery)
-
实时状态推送(WebSocket)
-
错误处理和重试机制
-
执行历史记录
-
性能监控和分析
技术实现:
-
FastAPI提供RESTful API
-
Celery处理长时间运行的任务
-
Redis作为消息代理
-
WebSocket推送执行状态
-
PostgreSQL存储执行记录
4. 低代码开发功能
功能要求:
-
代码生成器(Python、前端代码)
-
自定义节点开发
-
模板库管理
-
组件市场
技术实现:
-
基于工作流定义生成代码
-
支持自定义节点插件系统
-
模板存储和管理
-
组件导入导出功能
📐 项目结构设计
前端项目结构
frontend/
├── src/
│ ├── api/ # API接口封装
│ ├── assets/ # 静态资源
│ ├── components/ # 公共组件
│ │ ├── WorkflowEditor/ # 工作流编辑器
│ │ ├── NodePanel/ # 节点配置面板
│ │ ├── AgentConfig/ # Agent配置组件
│ │ └── CodeGenerator/ # 代码生成器
│ ├── composables/ # 组合式函数
│ ├── layouts/ # 布局组件
│ ├── router/ # 路由配置
│ ├── stores/ # Pinia状态管理
│ │ ├── workflow.ts # 工作流状态
│ │ ├── agent.ts # Agent状态
│ │ └── user.ts # 用户状态
│ ├── types/ # TypeScript类型定义
│ ├── utils/ # 工具函数
│ └── views/ # 页面组件
│ ├── WorkflowDesigner.vue
│ ├── AgentStudio.vue
│ └── CodeGenerator.vue
├── public/
└── package.json
后端项目结构
backend/
├── app/
│ ├── api/ # API路由
│ │ ├── workflows.py # 工作流API
│ │ ├── agents.py # Agent API
│ │ └── executions.py # 执行API
│ ├── core/ # 核心功能
│ │ ├── config.py # 配置
│ │ ├── security.py # 安全
│ │ └── database.py # 数据库
│ ├── models/ # 数据模型
│ │ ├── workflow.py
│ │ ├── agent.py
│ │ └── execution.py
│ ├── services/ # 业务逻辑
│ │ ├── workflow_engine.py # 工作流引擎
│ │ ├── agent_service.py # Agent服务
│ │ └── code_generator.py # 代码生成
│ ├── tasks/ # Celery任务
│ │ ├── workflow_tasks.py
│ │ └── agent_tasks.py
│ ├── websocket/ # WebSocket处理
│ └── utils/ # 工具函数
├── tests/ # 测试
├── alembic/ # 数据库迁移
├── requirements.txt
└── main.py
🔧 开发任务清单
阶段一: 项目初始化(1-2周)
1.1 前端项目搭建
-
使用 Vite 创建 Vue 3 + TypeScript 项目
-
配置 ESLint、Prettier
-
安装核心依赖(Vue Flow、Pinia、Element Plus等)
-
配置路由和布局
-
设置API请求拦截器
-
配置WebSocket连接
1.2 后端项目搭建
-
创建 FastAPI 项目结构
-
配置数据库连接(PostgreSQL)
-
配置 Redis 连接
-
设置 Celery 任务队列
-
配置 CORS 和中间件
-
设置日志系统
-
配置环境变量管理
1.3 数据库设计
-
设计工作流表结构
-
设计Agent表结构
-
设计执行记录表结构
-
设计用户和权限表
-
创建数据库迁移脚本
阶段二: 核心功能开发(4-6周)
2.1 工作流编辑器开发
-
实现 Vue Flow 基础编辑器
-
实现节点拖拽功能
-
实现节点连线功能
-
实现节点配置面板
-
实现工作流保存和加载
-
实现工作流验证逻辑
-
实现节点类型扩展机制
2.2 工作流引擎开发
-
实现工作流执行引擎核心逻辑
-
实现节点执行器(模板、条件、数据等)
-
实现数据流转机制
-
实现错误处理和重试
-
实现执行状态管理
-
集成 Celery 异步任务
2.3 Agent平台开发
-
实现Agent配置界面
-
集成 LangChain Agent框架
-
实现工具注册和管理
-
实现Agent执行逻辑
-
实现多Agent协作机制
-
实现Agent监控和分析
2.4 实时通信开发
-
实现 FastAPI WebSocket
-
实现前端 WebSocket 客户端
-
实现执行状态实时推送
-
实现消息队列集成
阶段三: 低代码功能开发(2-3周)
3.1 代码生成器开发
-
实现Python工作流引擎代码生成
-
实现前端组件代码生成
-
实现API接口代码生成
-
实现代码模板管理
-
实现代码预览和下载
3.2 模板和组件库
-
实现工作流模板库
-
实现自定义节点开发
-
实现组件导入导出
-
实现模板市场功能
阶段四: 系统集成和优化(2-3周)
4.1 系统集成
-
集成向量数据库(ChromaDB)
-
集成RAG框架(LlamaIndex)
-
实现第三方API集成
-
实现数据库集成
-
实现消息队列集成
4.2 性能优化
-
优化工作流执行性能
-
优化前端渲染性能
-
实现缓存机制
-
优化数据库查询
-
实现负载均衡
4.3 监控和日志
-
集成 Prometheus 监控
-
配置 Grafana 仪表板
-
实现日志收集和分析
-
实现告警机制
阶段五: 测试和部署(1-2周)
5.1 测试
-
编写单元测试
-
编写集成测试
-
编写端到端测试
-
性能测试
-
安全测试
5.2 部署
-
编写 Dockerfile
-
编写 docker-compose.yml
-
配置 Nginx 反向代理
-
配置 CI/CD 流水线
-
编写部署文档
💻 关键代码实现示例
1. 工作流执行引擎(Python)
# app/services/workflow_engine.py
from typing import Dict, Any, List
from celery import Task
import asyncio
from langchain.llms import OpenAI
from langchain.agents import initialize_agent
class WorkflowEngine:
def __init__(self, workflow_id: str):
self.workflow_id = workflow_id
self.llm = OpenAI(temperature=0.7)
self.nodes = []
self.edges = []
async def execute_node(self, node: Dict[str, Any], input_data: Dict[str, Any]) -> Dict[str, Any]:
"""执行单个节点"""
node_type = node.get('type')
if node_type == 'template':
# 模板节点:调用AI模型
prompt = self.build_prompt(node, input_data)
result = await self.llm.agenerate([prompt])
return {'output': result.generations[0][0].text}
elif node_type == 'condition':
# 条件节点:判断分支
condition = node.get('condition')
result = self.evaluate_condition(condition, input_data)
return {'output': result, 'branch': 'true' if result else 'false'}
elif node_type == 'data':
# 数据节点:数据转换
mapping = node.get('mapping')
result = self.transform_data(input_data, mapping)
return {'output': result}
elif node_type == 'agent':
# Agent节点:执行Agent
agent = initialize_agent(
tools=node.get('tools', []),
llm=self.llm,
agent="zero-shot-react-description"
)
result = await agent.arun(node.get('task'))
return {'output': result}
else:
raise ValueError(f"Unknown node type: {node_type}")
async def execute(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
"""执行完整工作流"""
workflow = self.load_workflow()
self.nodes = workflow['nodes']
self.edges = workflow['edges']
# 构建执行图
execution_graph = self.build_execution_graph()
# 按拓扑顺序执行
results = {}
node_outputs = {'start': input_data}
for node_id in execution_graph:
node = self.get_node_by_id(node_id)
node_input = self.get_node_input(node_id, node_outputs)
result = await self.execute_node(node, node_input)
results[node_id] = result
node_outputs[node_id] = result['output']
return results
2. 工作流编辑器(Vue 3)
<!-- src/components/WorkflowEditor/WorkflowEditor.vue -->
<template>
<div class="workflow-editor">
<VueFlow
v-model="nodes"
v-model:edges="edges"
:node-types="nodeTypes"
@node-click="onNodeClick"
@edge-click="onEdgeClick"
@pane-ready="onPaneReady"
>
<Background />
<Controls />
<MiniMap />
<NodeToolbar />
</VueFlow>
<NodeConfigPanel
v-if="selectedNode"
:node="selectedNode"
@save="handleNodeSave"
@close="selectedNode = null"
/>
</div>
</template>
<script setup lang="ts">
import { ref, computed } from 'vue'
import { VueFlow, useVueFlow, Background, Controls, MiniMap } from '@vue-flow/core'
import NodeConfigPanel from './NodeConfigPanel.vue'
import { useWorkflowStore } from '@/stores/workflow'
const workflowStore = useWorkflowStore()
const { nodes, edges, addNodes, addEdges } = useVueFlow()
const selectedNode = ref(null)
const nodeTypes = {
start: StartNode,
template: TemplateNode,
condition: ConditionNode,
data: DataNode,
agent: AgentNode,
end: EndNode
}
const onNodeClick = (event: NodeClickEvent) => {
selectedNode.value = event.node
}
const handleNodeSave = (nodeData: any) => {
workflowStore.updateNode(nodeData)
selectedNode.value = null
}
const saveWorkflow = async () => {
const workflow = {
nodes: nodes.value,
edges: edges.value
}
await workflowStore.saveWorkflow(workflow)
}
</script>
3. FastAPI工作流API
# app/api/workflows.py
from fastapi import APIRouter, Depends, WebSocket
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.services.workflow_engine import WorkflowEngine
from app.tasks.workflow_tasks import execute_workflow_task
from app.models.workflow import Workflow
router = APIRouter(prefix="/api/workflows", tags=["workflows"])
@router.post("/")
async def create_workflow(workflow_data: dict, db: Session = Depends(get_db)):
"""创建工作流"""
workflow = Workflow(**workflow_data)
db.add(workflow)
db.commit()
return {"id": workflow.id, "status": "created"}
@router.post("/{workflow_id}/execute")
async def execute_workflow(workflow_id: str, input_data: dict, db: Session = Depends(get_db)):
"""执行工作流"""
task = execute_workflow_task.delay(workflow_id, input_data)
return {"task_id": task.id, "status": "pending"}
@router.get("/executions/{task_id}")
async def get_execution_status(task_id: str):
"""获取执行状态"""
from app.core.celery_app import celery_app
task = celery_app.AsyncResult(task_id)
return {
"task_id": task_id,
"status": task.status,
"result": task.result if task.ready() else None
}
@router.websocket("/executions/{task_id}/ws")
async def websocket_execution_status(websocket: WebSocket, task_id: str):
"""WebSocket实时推送执行状态"""
await websocket.accept()
# 实现实时状态推送逻辑
while True:
status = get_task_status(task_id)
await websocket.send_json(status)
await asyncio.sleep(1)
4. Celery任务定义
# app/tasks/workflow_tasks.py
from celery import Task
from app.core.celery_app import celery_app
from app.services.workflow_engine import WorkflowEngine
from app.websocket.manager import WebSocketManager
@celery_app.task(bind=True)
def execute_workflow_task(self, workflow_id: str, input_data: dict):
"""执行工作流任务"""
try:
engine = WorkflowEngine(workflow_id)
# 更新任务状态
self.update_state(state='PROGRESS', meta={'progress': 0})
# 执行工作流
result = asyncio.run(engine.execute(input_data))
# 通过WebSocket推送结果
WebSocketManager.broadcast(workflow_id, {
'status': 'completed',
'result': result
})
return result
except Exception as e:
# 错误处理
WebSocketManager.broadcast(workflow_id, {
'status': 'failed',
'error': str(e)
})
raise
🎨 UI/UX设计要点
1. 工作流编辑器界面
-
左侧:节点工具箱(可拖拽的节点类型)
-
中间:画布区域(工作流可视化编辑)
-
右侧:节点配置面板(选中节点时显示)
-
顶部:工具栏(保存、运行、预览等)
-
底部:执行日志和状态栏
2. Agent配置界面
-
Agent基本信息配置
-
工具链选择和管理
-
提示词编辑器
-
参数调优面板
-
测试和预览功能
3. 代码生成器界面
-
代码模板选择
-
参数配置表单
-
代码预览(语法高亮)
-
下载和导出功能
🔒 安全和权限
1. 认证和授权
-
JWT Token认证
-
基于角色的访问控制(RBAC)
-
API接口权限控制
-
工作流和Agent的访问权限
2. 数据安全
-
敏感数据加密存储
-
API请求加密传输(HTTPS)
-
工作流执行日志脱敏
-
数据备份和恢复机制
3. 安全审计
-
操作日志记录
-
异常行为检测
-
安全漏洞扫描
-
合规性检查
📊 性能优化策略
1. 前端优化
-
代码分割和懒加载
-
虚拟滚动(处理大量节点)
-
防抖和节流(搜索、保存等操作)
-
组件缓存(keep-alive)
-
图片懒加载和压缩
-
使用 Web Workers 处理复杂计算
-
优化 Vue Flow 渲染性能
-
减少不必要的响应式数据
2. 后端优化
-
数据库查询优化(索引、查询优化)
-
使用连接池管理数据库连接
-
Redis 缓存热点数据
-
异步任务队列(Celery)处理耗时操作
-
API 响应压缩(gzip)
-
分页和限制查询结果
-
使用批量操作减少数据库交互
-
实现请求限流和熔断机制
3. 工作流执行优化
-
并行执行独立节点
-
缓存节点执行结果
-
优化 AI 模型调用(批量请求、模型选择)
-
实现执行计划优化
-
支持断点续传和状态恢复
-
资源池管理(连接池、线程池)
4. Agent 执行优化
-
Agent 响应缓存
-
工具调用结果缓存
-
优化提示词长度
-
选择合适的模型(速度 vs 质量)
-
实现 Agent 执行超时控制
-
批量处理相似任务
🧪 测试策略
1. 单元测试
前端测试:
-
使用 Vitest 进行组件测试
-
测试工具函数和组合式函数
-
测试状态管理逻辑
-
测试表单验证
后端测试:
-
使用 pytest 进行单元测试
-
测试服务层逻辑
-
测试数据模型
-
测试工具函数
2. 集成测试
API 集成测试:
-
测试完整的 API 流程
-
测试数据库交互
-
测试 Celery 任务执行
-
测试 WebSocket 通信
前端集成测试:
-
测试组件间交互
-
测试 API 调用
-
测试路由导航
-
测试状态管理
3. 端到端测试
-
使用 Playwright 或 Cypress
-
测试完整用户流程
-
测试工作流创建和执行
-
测试 Agent 配置和使用
-
测试错误处理场景
4. 性能测试
-
工作流执行性能测试
-
API 响应时间测试
-
并发执行测试
-
负载测试
-
压力测试
5. 安全测试
-
API 安全测试
-
认证授权测试
-
输入验证测试
-
SQL 注入防护测试
-
XSS 防护测试
🚀 部署指南
1. 开发环境部署
使用 Docker Compose:
# docker-compose.dev.yml
version: '3.8'
services:
frontend:
build: ./frontend
ports:
- "3000:3000"
volumes:
- ./frontend:/app
environment:
- VITE_API_URL=http://localhost:8000
backend:
build: ./backend
ports:
- "8000:8000"
volumes:
- ./backend:/app
environment:
- DATABASE_URL=postgresql://user:pass@db:5432/workflow
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
celery:
build: ./backend
command: celery -A app.core.celery_app worker --loglevel=info
volumes:
- ./backend:/app
depends_on:
- db
- redis
db:
image: postgres:15
environment:
- POSTGRES_DB=workflow
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
volumes:
- postgres_data:/var/lib/postgresql/data
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
2. 生产环境部署
使用 Kubernetes:
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: workflow-backend
spec:
replicas: 3
selector:
matchLabels:
app: workflow-backend
template:
metadata:
labels:
app: workflow-backend
spec:
containers:
- name: backend
image: workflow-backend:latest
ports:
- containerPort: 8000
env:
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: workflow-secrets
key: database-url
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2000m"
3. CI/CD 配置
GitHub Actions 示例:
# .github/workflows/deploy.yml
name: Deploy
on:
push:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Run tests
run: |
cd backend && pytest
cd frontend && npm test
build:
needs: test
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Build Docker images
run: |
docker build -t workflow-backend ./backend
docker build -t workflow-frontend ./frontend
deploy:
needs: build
runs-on: ubuntu-latest
steps:
- name: Deploy to production
run: |
# 部署脚本
💡 开发最佳实践
1. 代码规范
前端:
-
使用 ESLint + Prettier 统一代码风格
-
遵循 Vue 3 组合式 API 最佳实践
-
使用 TypeScript 严格模式
-
组件命名使用 PascalCase
-
文件命名使用 kebab-case
后端:
-
遵循 PEP 8 Python 代码规范
-
使用类型提示(Type Hints)
-
使用 Pydantic 进行数据验证
-
API 路由使用 RESTful 规范
-
函数和类添加文档字符串
2. Git 工作流
-
使用 Git Flow 或 GitHub Flow
-
提交信息遵循 Conventional Commits
-
代码审查(Code Review)必须
-
主分支保护,禁止直接推送
-
使用语义化版本号(Semantic Versioning)
3. 错误处理
前端错误处理:
// 统一错误处理
try {
const result = await api.workflow.execute(id, data)
} catch (error) {
if (error.response?.status === 401) {
// 处理未授权
router.push('/login')
} else {
// 显示错误提示
ElMessage.error(error.message)
}
}
后端错误处理:
# 统一异常处理
from fastapi import HTTPException
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
return JSONResponse(
status_code=400,
content={"message": str(exc)}
)
4. 日志记录
前端日志:
-
使用 console 进行开发调试
-
生产环境使用日志服务(Sentry)
-
记录关键操作和错误
后端日志:
import logging
logger = logging.getLogger(__name__)
logger.info("Workflow execution started", extra={
"workflow_id": workflow_id,
"user_id": user_id
})
logger.error("Workflow execution failed", exc_info=True)
5. 文档编写
-
API 文档使用 OpenAPI/Swagger
-
代码注释清晰完整
-
README 包含项目介绍和快速开始
-
编写用户使用手册
-
维护更新日志(CHANGELOG)
🔍 常见问题解决
1. 工作流执行超时
问题: 工作流执行时间过长导致超时
解决方案:
-
使用 Celery 异步任务处理
-
增加任务超时时间配置
-
优化节点执行逻辑
-
实现任务分片执行
2. WebSocket 连接断开
问题: WebSocket 连接不稳定,经常断开
解决方案:
-
实现自动重连机制
-
使用心跳检测保持连接
-
增加连接池管理
-
处理网络异常情况
3. 前端性能问题
问题: 工作流编辑器在处理大量节点时卡顿
解决方案:
-
使用虚拟滚动
-
实现节点懒加载
-
优化 Vue Flow 渲染
-
使用 Web Workers 处理计算
4. 数据库性能问题
问题: 数据库查询慢,影响系统性能
解决方案:
-
添加合适的数据库索引
-
优化 SQL 查询语句
-
使用 Redis 缓存热点数据
-
实现数据库读写分离
5. Agent 响应慢
问题: Agent 执行响应时间过长
解决方案:
-
选择合适的 AI 模型(速度优先)
-
实现响应缓存机制
-
优化提示词长度
-
使用流式响应
📚 参考资源和文档
技术文档
前端:
后端:
基础设施:
架构设计参考
开发工具
前端开发工具:
后端开发工具:
🎯 开发里程碑
MVP 版本(3-4个月)
核心功能:
-
✅ 基础工作流编辑器(拖拽、连线、配置)
-
✅ 工作流执行引擎(支持基础节点类型)
-
✅ Agent 基础配置和执行
-
✅ 用户认证和权限管理
-
✅ 工作流执行历史记录
技术目标:
-
前后端基础架构搭建完成
-
核心功能可用
-
基础测试覆盖
V1.0 版本(6-8个月)
增强功能:
-
✅ 完整的工作流节点类型支持
-
✅ 多Agent协作机制
-
✅ 代码生成器(Python、前端)
-
✅ 工作流模板库
-
✅ 性能监控和分析
-
✅ 完整的错误处理和重试机制
技术目标:
-
系统稳定性和性能优化
-
完整的测试覆盖
-
生产环境部署就绪
V2.0 版本(10-12个月)
高级功能:
-
✅ RAG 和向量数据库集成
-
✅ 自定义节点开发平台
-
✅ 组件市场和插件系统
-
✅ 高级分析和报表
-
✅ 多租户支持
-
✅ 企业级安全和合规
技术目标:
-
企业级功能完善
-
高可用和可扩展架构
-
完整的文档和培训材料
🔄 迭代开发流程
1. 需求分析阶段
-
收集用户需求和反馈
-
分析业务场景
-
确定功能优先级
-
制定开发计划
2. 设计阶段
-
系统架构设计
-
数据库设计
-
API 接口设计
-
UI/UX 设计
-
技术方案评审
3. 开发阶段
-
前端开发
-
后端开发
-
单元测试编写
-
代码审查
-
持续集成
4. 测试阶段
-
功能测试
-
集成测试
-
性能测试
-
安全测试
-
用户验收测试
5. 部署阶段
-
预发布环境测试
-
生产环境部署
-
监控和告警配置
-
用户培训
6. 维护阶段
-
问题修复
-
性能优化
-
功能迭代
-
用户支持
📋 开发检查清单
代码提交前检查
-
代码通过 ESLint/Black 检查
-
所有测试用例通过
-
代码审查完成
-
更新相关文档
-
提交信息符合规范
功能开发检查
-
功能需求明确
-
技术方案确定
-
数据库变更(如有)已迁移
-
API 接口文档已更新
-
前端组件已测试
-
错误处理已实现
-
日志记录已添加
发布前检查
-
所有测试通过
-
性能测试通过
-
安全扫描通过
-
文档已更新
-
版本号已更新
-
更新日志已编写
-
回滚方案已准备
🛠️ 开发环境配置
前端开发环境
# 安装 Node.js (推荐 v18+)
# 安装 pnpm
npm install -g pnpm
# 安装依赖
cd frontend
pnpm install
# 启动开发服务器
pnpm dev
# 运行测试
pnpm test
# 构建生产版本
pnpm build
后端开发环境
# 安装 Python (推荐 3.11+)
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 安装依赖
cd backend
pip install -r requirements.txt
# 配置环境变量
cp .env.example .env
# 编辑 .env 文件
# 运行数据库迁移
alembic upgrade head
# 启动开发服务器
uvicorn app.main:app --reload
# 启动 Celery Worker
celery -A app.core.celery_app worker --loglevel=info
# 运行测试
pytest
Docker 开发环境
# 启动所有服务
docker-compose -f docker-compose.dev.yml up -d
# 查看日志
docker-compose logs -f
# 停止服务
docker-compose down
📝 API 设计规范
RESTful API 设计
资源命名:
-
使用名词,复数形式
-
使用小写字母和连字符
-
示例:
/api/workflows,/api/agents,/api/executions
HTTP 方法:
-
GET: 获取资源 -
POST: 创建资源 -
PUT: 完整更新资源 -
PATCH: 部分更新资源 -
DELETE: 删除资源
状态码:
-
200 OK: 成功 -
201 Created: 创建成功 -
400 Bad Request: 请求错误 -
401 Unauthorized: 未授权 -
403 Forbidden: 禁止访问 -
404 Not Found: 资源不存在 -
500 Internal Server Error: 服务器错误
响应格式:
{
"code": 200,
"message": "success",
"data": {},
"timestamp": "2024-01-01T00:00:00Z"
}
WebSocket 消息格式
{
"type": "workflow:progress",
"task_id": "xxx",
"data": {
"node_id": "node_1",
"status": "running",
"progress": 50
},
"timestamp": "2024-01-01T00:00:00Z"
}
🎓 学习路径建议
新手开发者
- 基础学习 (1-2周)
- Vue 3 基础语法和组合式 API
- TypeScript 基础
- Python 基础
- FastAPI 快速入门
- 工具学习 (1周)
- Git 版本控制
- Docker 基础
- 开发工具使用
- 项目实践 (2-3周)
- 熟悉项目结构
- 完成简单功能开发
- 参与代码审查
中级开发者
- 进阶学习 (2-3周)
- Vue Flow 工作流可视化
- Celery 异步任务
- LangChain Agent 开发
- 数据库优化
- 架构理解 (1-2周)
- 系统架构设计
- 微服务模式
- 性能优化技巧
- 功能开发 (持续)
- 独立开发复杂功能
- 技术方案设计
- 代码审查和指导
高级开发者
- 架构设计 (持续)
- 系统架构优化
- 技术选型决策
- 性能瓶颈分析
- 团队协作 (持续)
- 技术方案评审
- 代码审查指导
- 技术分享和培训
- 技术创新 (持续)
- 新技术调研
- 最佳实践总结
- 开源贡献
🐛 调试技巧
前端调试
Vue DevTools:
-
安装 Vue DevTools 浏览器扩展
-
查看组件状态和 props
-
监控 Pinia store 变化
-
性能分析
浏览器调试:
-
使用 Chrome DevTools
-
网络请求监控
-
性能分析工具
-
内存泄漏检测
后端调试
日志调试:
import logging
logger = logging.getLogger(__name__)
# 设置日志级别
logging.basicConfig(level=logging.DEBUG)
# 使用结构化日志
logger.debug("Debug message", extra={"key": "value"})
调试工具:
-
使用
pdb进行断点调试 -
使用
ipdb增强调试体验 -
使用 FastAPI 的自动文档进行 API 测试
Celery 调试:
# 查看任务状态
from app.core.celery_app import celery_app
task = celery_app.AsyncResult('task-id')
print(task.state)
print(task.result)
📞 支持和帮助
内部资源
-
项目 Wiki 文档
-
技术分享会议记录
-
代码审查记录
-
问题追踪系统
外部社区
-
Stack Overflow
问题反馈
-
GitHub Issues(技术问题)
-
内部工单系统(业务问题)
-
团队会议(架构讨论)
📅 版本历史
v1.0.0 (2024-01-01)
-
初始版本发布
-
基础工作流编辑器
-
工作流执行引擎
-
Agent 基础功能
文档维护: 开发团队
最后更新: 2024年
版本: 1.0