夜雨聆风本文档是《本地部署全能 AI Agent 完整方案》的进阶篇，聚焦于知识库自动化运营、Skill 动态发现与注册、生产级稳定性保障、用户自定义 Agent 体系，以及完整的工程化落地方案。适合已完成基础部署、希望将 Agent 平台打造为可持续运营的生产力工具的读者。
第一部分：知识库内容自动获取与持续更新
1.1 核心问题
手动上传文档到知识库效率低、容易过时。需要一套自动化采集 → 清洗 → 入库 → 更新的流水线。
1.2 自动采集架构
┌─────────────────────────────────────────────────────┐│ 知识库自动化流水线 ││ ││ ┌─────────┐ ┌─────────┐ ┌─────────┐ ││ │ 数据源 │──▶│ 采集器 │──▶│ 清洗器 │ ││ │ Sources │ │ Scrapers│ │ Cleaners│ ││ └─────────┘ └─────────┘ └─────────┘ ││ │ │ ││ │ ▼ ││ ┌─────────┐ ┌─────────┐ ┌─────────┐ ││ │ 变更检测 │──▶│ 增量更新 │──▶│ Dify │ ││ │ Monitor │ │ Updater │ │ 知识库API│ ││ └─────────┘ └─────────┘ └─────────┘ │└─────────────────────────────────────────────────────┘1.3 各领域知识源自动采集方案
1.3.1 编程开发知识
# knowledge_scrapers/coding_scraper.pyimport requestsimport osimport jsonfrom pathlib import Pathclass CodingKnowledgeScraper: """自动采集编程领域知识""" def __init__(self, output_dir: str = "./knowledge/coding"): self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) def scrape_github_awesome_lists(self): """从 GitHub Awesome 列表采集最佳实践""" awesome_repos = [ "sindresorhus/awesome", "enaqx/awesome-react", "vinta/awesome-python", "avelino/awesome-go", "typescript-cheatsheets/react", ] for repo in awesome_repos: url = f"https://api.github.com/repos/{repo}/readme" resp = requests.get(url, headers={"Accept": "application/vnd.github.v3.raw"}) if resp.status_code == 200: filename = repo.replace("/", "_") + ".md" (self.output_dir / filename).write_text(resp.text, encoding="utf-8") def scrape_official_docs(self): """采集官方文档（通过 Firecrawl / Crawl4AI）""" # Crawl4AI: 开源爬虫，专为 AI 知识采集设计 # pip install crawl4ai from crawl4ai import WebCrawler crawler = WebCrawler() doc_sites = [ {"url": "https://react.dev/learn", "name": "react_docs"}, {"url": "https://www.typescriptlang.org/docs/", "name": "typescript_docs"}, {"url": "https://docs.python.org/3/tutorial/", "name": "python_docs"}, ] for site in doc_sites: result = crawler.run(url=site["url"]) output_file = self.output_dir / f"{site['name']}.md" output_file.write_text(result.markdown, encoding="utf-8") def scrape_internal_api_docs(self, swagger_urls: list): """从 Swagger/OpenAPI 自动生成 API 知识文档""" for url in swagger_urls: resp = requests.get(url) spec = resp.json() markdown = self._openapi_to_markdown(spec) name = spec.get("info", {}).get("title", "api").replace(" ", "_") (self.output_dir / f"api_{name}.md").write_text(markdown, encoding="utf-8") def _openapi_to_markdown(self, spec: dict) -> str: """将 OpenAPI spec 转为可读的 Markdown 文档""" lines = [f"# {spec.get('info', {}).get('title', 'API Documentation')}\n"] for path, methods in spec.get("paths", {}).items(): for method, detail in methods.items(): if isinstance(detail, dict): summary = detail.get("summary", "") lines.append(f"## {method.upper()} {path}") lines.append(f"{summary}\n") # 参数 params = detail.get("parameters", []) if params: lines.append("**Parameters:**") for p in params: lines.append(f"- `{p.get('name')}` ({p.get('in')}): {p.get('description', '')}") lines.append("") return "\n".join(lines) def scrape_stackoverflow_tags(self, tags: list, count: int = 50): """采集 StackOverflow 高票问答作为 FAQ 知识""" for tag in tags: url = f"https://api.stackexchange.com/2.3/questions" params = { "order": "desc", "sort": "votes", "tagged": tag, "site": "stackoverflow", "pagesize": count, "filter": "withbody" } resp = requests.get(url, params=params) if resp.status_code == 200: items = resp.json().get("items", []) content = f"# {tag} FAQ (Top {count})\n\n" for item in items: content += f"## Q: {item['title']}\n" content += f"Score: {item['score']} | Answers: {item['answer_count']}\n" content += f"Link: {item['link']}\n\n" (self.output_dir / f"faq_{tag}.md").write_text(content, encoding="utf-8")1.3.2 文案写作知识
# knowledge_scrapers/writing_scraper.pyclass WritingKnowledgeScraper: """自动采集文案写作知识""" def scrape_copywriting_templates(self): """采集营销文案模板库""" # 从开源文案库采集 sources = [ # 中文文案排版指南 "https://raw.githubusercontent.com/sparanoid/chinese-copywriting-guidelines/master/README.zh-Hans.md", ] for url in sources: resp = requests.get(url) if resp.status_code == 200: filename = url.split("/")[-1] (self.output_dir / filename).write_text(resp.text, encoding="utf-8") def scrape_writing_guides_from_rss(self, rss_feeds: list): """从 RSS 订阅持续采集写作技巧文章""" import feedparser for feed_url in rss_feeds: feed = feedparser.parse(feed_url) for entry in feed.entries[:20]: # 每个源取最新20篇 content = entry.get("summary", entry.get("description", "")) title = entry.get("title", "untitled") filename = f"{title[:50].replace('/', '_')}.md" doc = f"# {title}\n\n{content}\n\nSource: {entry.get('link', '')}" (self.output_dir / filename).write_text(doc, encoding="utf-8") def generate_style_guide(self, brand_name: str, sample_texts: list): """用 LLM 从样本文本中提取品牌风格指南""" prompt = f"""分析以下 {brand_name} 的文案样本，提取品牌调性和写作风格指南：样本：{chr(10).join(sample_texts)}请输出：1. 品牌调性关键词（3-5个）2. 语言风格特征3. 常用句式模板4. 禁用词/表达5. 典型文案结构""" # 调用本地 Qwen 模型 resp = requests.post("http://localhost:8000/v1/chat/completions", json={ "model": "qwen2.5-72b", "messages": [{"role": "user", "content": prompt}] }) guide = resp.json()["choices"][0]["message"]["content"] return guide1.3.3 图像/视频知识
# knowledge_scrapers/creative_scraper.pyclass CreativeKnowledgeScraper: """采集图像和视频创作知识""" def scrape_prompt_databases(self): """采集高质量 Prompt 数据库""" # 从 CivitAI / PromptHero 等平台采集优质 prompt # 分类存储：人像、风景、产品、抽象等 categories = { "portrait": "人像摄影 prompt 集合", "landscape": "风景 prompt 集合", "product": "产品图 prompt 集合", "illustration": "插画 prompt 集合", "ui_design": "UI 设计 prompt 集合", } for cat, desc in categories.items(): # 生成分类 prompt 指南 prompt = f"请为 FLUX/Stable Diffusion 模型生成 30 个高质量的{desc}，每个包含正向和负向 prompt，标注推荐参数（步数、CFG、尺寸）" resp = requests.post("http://localhost:8000/v1/chat/completions", json={ "model": "qwen2.5-72b", "messages": [{"role": "user", "content": prompt}] }) content = resp.json()["choices"][0]["message"]["content"] (self.output_dir / f"prompts_{cat}.md").write_text( f"# {desc}\n\n{content}", encoding="utf-8" ) def scrape_comfyui_workflows(self): """从社区采集 ComfyUI 工作流模板""" # OpenArt / Comfy.icu 等平台有大量共享工作流 workflow_categories = [ "text_to_image_basic", "image_to_image", "inpainting", "upscale", "text_to_video", "image_to_video", "style_transfer", ] # 将工作流 JSON + 说明文档存入知识库 for cat in workflow_categories: doc = f"# ComfyUI Workflow: {cat}\n\n" doc += f"工作流文件: workflows/{cat}.json\n" doc += f"使用说明: ...\n" (self.output_dir / f"workflow_{cat}.md").write_text(doc, encoding="utf-8")1.4 自动入库流水线
# knowledge_pipeline.py — 知识库自动更新流水线import hashlibimport jsonimport requestsfrom pathlib import Pathfrom datetime import datetimeclass DifyKnowledgeSync: """与 Dify 知识库 API 同步""" def __init__(self, dify_url: str, api_key: str): self.dify_url = dify_url.rstrip("/") self.api_key = api_key self.headers = {"Authorization": f"Bearer {api_key}"} self.state_file = Path("./knowledge_state.json") self.state = self._load_state() def _load_state(self) -> dict: if self.state_file.exists(): return json.loads(self.state_file.read_text()) return {"files": {}} def _save_state(self): self.state_file.write_text(json.dumps(self.state, indent=2, ensure_ascii=False)) def _file_hash(self, filepath: Path) -> str: return hashlib.sha256(filepath.read_bytes()).hexdigest() def sync_directory(self, local_dir: str, dataset_id: str): """ 将本地目录与 Dify 知识库同步 - 新文件 → 上传 - 已修改文件 → 删除旧版本 + 重新上传 - 已删除文件 → 从知识库删除 """ local_dir = Path(local_dir) current_files = {} # 扫描本地文件 for f in local_dir.rglob("*"): if f.is_file() and f.suffix in (".md", ".txt", ".pdf", ".docx", ".html"): rel_path = str(f.relative_to(local_dir)) file_hash = self._file_hash(f) current_files[rel_path] = {"hash": file_hash, "path": str(f)} # 对比变更 old_files = self.state.get("files", {}).get(dataset_id, {}) # 新增或修改的文件 for rel_path, info in current_files.items(): old_hash = old_files.get(rel_path, {}).get("hash") if old_hash != info["hash"]: print(f"[SYNC] Uploading: {rel_path}") # 如果是更新，先删除旧文档 old_doc_id = old_files.get(rel_path, {}).get("doc_id") if old_doc_id: self._delete_document(dataset_id, old_doc_id) # 上传新文档 doc_id = self._upload_document(dataset_id, info["path"]) current_files[rel_path]["doc_id"] = doc_id # 已删除的文件 for rel_path in set(old_files.keys()) - set(current_files.keys()): old_doc_id = old_files[rel_path].get("doc_id") if old_doc_id: print(f"[SYNC] Deleting: {rel_path}") self._delete_document(dataset_id, old_doc_id) # 保存状态 if "files" not in self.state: self.state["files"] = {} self.state["files"][dataset_id] = current_files self._save_state() def _upload_document(self, dataset_id: str, filepath: str) -> str: """上传文档到 Dify 知识库""" url = f"{self.dify_url}/v1/datasets/{dataset_id}/document/create_by_file" with open(filepath, "rb") as f: files = {"file": f} data = { "data": json.dumps({ "indexing_technique": "high_quality", "process_rule": { "mode": "automatic" } }) } resp = requests.post(url, headers=self.headers, files=files, data=data) if resp.status_code == 200: return resp.json().get("document", {}).get("id", "") return "" def _delete_document(self, dataset_id: str, document_id: str): """从知识库删除文档""" url = f"{self.dify_url}/v1/datasets/{dataset_id}/documents/{document_id}" requests.delete(url, headers=self.headers)# ========== 定时任务：每日自动更新 ==========# crontab: 0 3 * * * python knowledge_pipeline.pyif __name__ == "__main__": from knowledge_scrapers.coding_scraper import CodingKnowledgeScraper from knowledge_scrapers.writing_scraper import WritingKnowledgeScraper from knowledge_scrapers.creative_scraper import CreativeKnowledgeScraper # Step 1: 采集 print(f"[{datetime.now()}] Starting knowledge scrape...") CodingKnowledgeScraper("./knowledge/coding").scrape_github_awesome_lists() CodingKnowledgeScraper("./knowledge/coding").scrape_official_docs() WritingKnowledgeScraper("./knowledge/writing").scrape_copywriting_templates() CreativeKnowledgeScraper("./knowledge/creative").scrape_prompt_databases() CreativeKnowledgeScraper("./knowledge/creative").scrape_comfyui_workflows() # Step 2: 同步到 Dify print(f"[{datetime.now()}] Syncing to Dify...") syncer = DifyKnowledgeSync( dify_url="http://localhost/api", api_key="your-dify-api-key" ) # 每个领域对应一个 Dify dataset dataset_mapping = { "./knowledge/coding": "dataset-id-coding", "./knowledge/writing": "dataset-id-writing", "./knowledge/creative": "dataset-id-creative", "./knowledge/life": "dataset-id-life", } for local_dir, dataset_id in dataset_mapping.items(): syncer.sync_directory(local_dir, dataset_id) print(f"[{datetime.now()}] Done!")1.5 知识库质量保障
# knowledge_quality.py — 知识库质量检测class KnowledgeQualityChecker: """定期检查知识库质量""" def check_freshness(self, knowledge_dir: str, max_age_days: int = 90): """检查文档时效性，标记过期内容""" stale_files = [] for f in Path(knowledge_dir).rglob("*"): if f.is_file(): age_days = (datetime.now() - datetime.fromtimestamp(f.stat().st_mtime)).days if age_days > max_age_days: stale_files.append({"file": str(f), "age_days": age_days}) return stale_files def check_retrieval_quality(self, dataset_id: str, test_queries: list): """用测试查询验证检索质量""" results = [] for query in test_queries: # 调用 Dify 检索 API resp = requests.post( f"http://localhost/api/v1/datasets/{dataset_id}/retrieve", headers={"Authorization": "Bearer your-key"}, json={"query": query, "top_k": 5} ) records = resp.json().get("records", []) avg_score = sum(r.get("score", 0) for r in records) / max(len(records), 1) results.append({ "query": query, "hit_count": len(records), "avg_score": round(avg_score, 3), "quality": "GOOD" if avg_score > 0.7 else "NEEDS_IMPROVEMENT" }) return results def check_duplicates(self, knowledge_dir: str): """检测重复或高度相似的文档""" from difflib import SequenceMatcher files = list(Path(knowledge_dir).rglob("*.md")) duplicates = [] for i, f1 in enumerate(files): for f2 in files[i + 1:]: text1 = f1.read_text(encoding="utf-8")[:2000] text2 = f2.read_text(encoding="utf-8")[:2000] similarity = SequenceMatcher(None, text1, text2).ratio() if similarity > 0.8: duplicates.append({ "file1": str(f1), "file2": str(f2), "similarity": round(similarity, 2) }) return duplicates第二部分：Skill 发现与动态注册
2.1 Skill 体系设计
┌────────────────────────────────────────────────┐│ Skill Registry ││ ││ ┌──────────┐ ┌──────────┐ ┌──────────┐ ││ │ 内置Skills│ │ 社区Skills│ │ 自定义 │ ││ │ Built-in │ │ Community│ │ Skills │ ││ └─────┬────┘ └─────┬────┘ └─────┬────┘ ││ └──────────┬──┘─────────────┘ ││ ▼ ││ ┌──────────────┐ ││ │ Skill Router │ ← 意图匹配 + 能力发现││ └──────┬───────┘ ││ ▼ ││ ┌──────────────┐ ││ │ Skill Runner │ ← 执行 + 结果聚合 ││ └──────────────┘ │└────────────────────────────────────────────────┘2.2 Skill 定义标准
// skills/text2ppt.skill.json{ "name": "text2ppt", "version": "1.0.0", "display_name": "文本转PPT", "description": "将文本内容或大纲自动生成专业 PPT 文件", "category": "document", "tags": ["ppt", "presentation", "office", "演示文稿"], "trigger_patterns": [ "做(一个|份)?PPT", "制作演示文稿", "生成(PPT|幻灯片)", "text.*(to|转|变).*(ppt|slide)" ], "input_schema": { "type": "object", "properties": { "title": {"type": "string", "description": "PPT标题"}, "outline": {"type": "string", "description": "内容大纲或主题"}, "style": { "type": "string", "enum": ["business", "academic", "creative", "minimal"], "default": "business" }, "slide_count": {"type": "integer", "default": 10} }, "required": ["title"] }, "output_schema": { "type": "object", "properties": { "file_path": {"type": "string"}, "download_url": {"type": "string"}, "slide_count": {"type": "integer"} } }, "dependencies": { "tools": ["ppt_generator", "text2img"], "knowledge_bases": ["ppt_design"], "models": ["qwen2.5-72b"] }, "execution": { "type": "workflow", "steps": [ {"action": "llm_generate", "prompt_template": "ppt_outline"}, {"action": "tool_call", "tool": "text2img", "for_each": "slides"}, {"action": "tool_call", "tool": "ppt_generator"} ] }}2.3 Skill 自动发现引擎
# skill_discovery.py — Skill 自动发现与注册import jsonimport refrom pathlib import Pathfrom typing import Optionalclass SkillRegistry: """Skill 注册中心""" def __init__(self, skills_dir: str = "./skills"): self.skills_dir = Path(skills_dir) self.skills: dict = {} self._load_all_skills() def _load_all_skills(self): """启动时加载所有 Skill 定义""" for skill_file in self.skills_dir.rglob("*.skill.json"): with open(skill_file, "r", encoding="utf-8") as f: skill = json.load(f) self.skills[skill["name"]] = skill print(f"[SKILL] Loaded: {skill['name']} ({skill['display_name']})") def match_skill(self, user_input: str) -> Optional[dict]: """根据用户输入匹配最合适的 Skill""" matches = [] for name, skill in self.skills.items(): score = 0 # 1. 正则 trigger 匹配 for pattern in skill.get("trigger_patterns", []): if re.search(pattern, user_input, re.IGNORECASE): score += 10 # 2. tag 关键词匹配 for tag in skill.get("tags", []): if tag.lower() in user_input.lower(): score += 3 # 3. description 语义匹配（可用 Embedding 增强） if score > 0: matches.append({"skill": skill, "score": score}) matches.sort(key=lambda x: x["score"], reverse=True) return matches[0]["skill"] if matches else None def register_skill(self, skill_definition: dict): """动态注册新 Skill""" name = skill_definition["name"] skill_file = self.skills_dir / f"{name}.skill.json" with open(skill_file, "w", encoding="utf-8") as f: json.dump(skill_definition, f, ensure_ascii=False, indent=2) self.skills[name] = skill_definition print(f"[SKILL] Registered: {name}") def list_skills(self, category: str = None) -> list: """列出所有可用 Skill""" skills = list(self.skills.values()) if category: skills = [s for s in skills if s.get("category") == category] return [{"name": s["name"], "display_name": s["display_name"], "description": s["description"]} for s in skills]class SkillDiscoveryAgent: """ Skill 发现 Agent —— 当用户请求无法匹配现有 Skill 时， 自动搜索社区/市场，下载并注册新 Skill """ def __init__(self, registry: SkillRegistry): self.registry = registry # 社区 Skill 市场（可配置多个源） self.marketplaces = [ "https://marketplace.dify.ai/api/skills", # Dify 官方市场 "https://registry.mcp.run/api/tools", # MCP 社区 ] def discover_and_install(self, user_need: str) -> Optional[dict]: """ 1. 先在本地匹配 2. 匹配不到则搜索社区市场 3. 下载、验证、注册 """ # 本地匹配 skill = self.registry.match_skill(user_need) if skill: return skill # 搜索社区 print(f"[DISCOVERY] No local skill found for: {user_need}") print(f"[DISCOVERY] Searching community marketplaces...") for marketplace_url in self.marketplaces: try: resp = requests.get(marketplace_url, params={"q": user_need}, timeout=10) if resp.status_code == 200: results = resp.json().get("results", []) if results: best_match = results[0] print(f"[DISCOVERY] Found: {best_match['name']} from {marketplace_url}") # 下载 Skill 定义 skill_def = self._download_skill(best_match["download_url"]) if skill_def and self._validate_skill(skill_def): self.registry.register_skill(skill_def) return skill_def except Exception as e: print(f"[DISCOVERY] Marketplace error: {e}") return None def _download_skill(self, url: str) -> Optional[dict]: """下载 Skill 定义""" try: resp = requests.get(url, timeout=10) return resp.json() except Exception: return None def _validate_skill(self, skill_def: dict) -> bool: """验证 Skill 定义的完整性和安全性""" required_fields = ["name", "version", "description", "input_schema"] for field in required_fields: if field not in skill_def: print(f"[VALIDATION] Missing required field: {field}") return False # 安全检查：不允许执行任意代码的 Skill execution = skill_def.get("execution", {}) if execution.get("type") == "code" and "eval" in json.dumps(execution): print("[VALIDATION] Rejected: contains unsafe code execution") return False return True2.4 内置 Skill 清单
┌──────────────────────────────────────────────────────────────────┐│ 内置 Skills 清单 │├──────────┬───────────────┬───────────────────────────────────────┤│ 类别 │ Skill 名称 │ 功能描述 │├──────────┼───────────────┼───────────────────────────────────────┤│ 文案 │ text_rewrite │ 文案润色/改写/风格转换 ││ │ seo_optimize │ SEO 关键词优化 ││ │ copywriting │ 营销文案生成（多模板） ││ │ summary │ 长文摘要/提取要点 ││ │ translation │ 中英文翻译（保持风格） │├──────────┼───────────────┼───────────────────────────────────────┤│ 文档 │ text2ppt │ 文本/大纲 → PPT 生成 ││ │ text2doc │ 结构化文档生成（Word） ││ │ pdf_extract │ PDF 解析 + 内容提取 ││ │ markdown2doc │ Markdown → Word/PDF 转换 │├──────────┼───────────────┼───────────────────────────────────────┤│ 编码 │ code_generate │ 代码生成（多语言） ││ │ code_review │ 代码审查 + 改进建议 ││ │ code_explain │ 代码逐行解释 ││ │ debug_assist │ 错误诊断 + 修复建议 ││ │ test_generate │ 单元测试自动生成 ││ │ sql_generate │ 自然语言 → SQL │├──────────┼───────────────┼───────────────────────────────────────┤│ 图像 │ text2img │ 文本描述 → 图片生成 ││ │ img2img │ 图片风格转换 ││ │ img_upscale │ 图片超分辨率放大 ││ │ img_inpaint │ 图片局部修改/擦除 ││ │ img_describe │ 图片内容描述/分析 ││ │ remove_bg │ 智能抠图/去背景 │├──────────┼───────────────┼───────────────────────────────────────┤│ 视频 │ text2video │ 文本描述 → 视频生成 ││ │ img2video │ 静态图 → 动态视频 ││ │ video_trim │ 视频裁剪/拼接 ││ │ video_caption │ 视频字幕生成 ││ │ video_bgm │ 自动配乐/音频处理 │├──────────┼───────────────┼───────────────────────────────────────┤│ 生活 │ weather │ 天气查询 + 出行建议 ││ │ recipe │ 菜谱推荐（按食材/口味） ││ │ schedule │ 日程管理/提醒 ││ │ health_tip │ 健康建议/运动推荐 ││ │ web_search │ 联网搜索 + 结果摘要 │└──────────┴───────────────┴───────────────────────────────────────┘2.5 在 Dify 中配置 Skill → Tool 映射
# skill_to_dify_tool.py — 将 Skill 注册为 Dify 自定义工具import requestsimport jsonclass DifyToolRegistrar: """将 Skill 注册为 Dify 工具""" def __init__(self, dify_url: str, api_key: str): self.dify_url = dify_url self.headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} def register_skill_as_tool(self, skill: dict): """将 Skill 定义转换为 Dify OpenAPI Tool""" openapi_spec = { "openapi": "3.0.0", "info": { "title": skill["display_name"], "version": skill["version"] }, "paths": { f"/skill/{skill['name']}": { "post": { "summary": skill["description"], "operationId": skill["name"], "requestBody": { "content": { "application/json": { "schema": skill["input_schema"] } } }, "responses": { "200": { "content": { "application/json": { "schema": skill.get("output_schema", {}) } } } } } } }, "servers": [{"url": "http://localhost:8600"}] } return openapi_spec第三部分：可实施性与稳定性保障
3.1 分级部署策略（降低风险）
Level 0: 最小可行版（1天搞定）──────────────────────────────✅ Ollama + Qwen2.5-14B（单机 8GB 显存即可）✅ Dify 社区版（docker compose up -d）✅ 手动上传 5-10 个文档到知识库✅ 创建 1 个通用 Agent（纯对话 + 知识库检索）❌ 暂无图像/视频生成❌ 暂无自定义 Skill→ 验证：能对话、能检索知识库 = 成功Level 1: 基础多模态版（1周）──────────────────────────────✅ Level 0 全部✅ 升级模型到 Qwen2.5-32B（24GB 显存）✅ 部署 ComfyUI + SDXL/FLUX（文生图）✅ 创建 3-4 个 SubAgent（文案、编码、图像、生活）✅ 知识库扩展到 50+ 文档❌ 暂无视频生成❌ 暂无自动化采集→ 验证：能分领域对话、能生成图片 = 成功Level 2: 完整版（2-3周）──────────────────────────────✅ Level 1 全部✅ 升级到 Qwen2.5-72B（双卡/A100）✅ 视频生成（Wan2.1 / CogVideoX）✅ PPT/视频剪辑微服务✅ MCP Server 集成✅ 知识库自动采集流水线✅ 全部 6 个 SubAgent + 主 Orchestrator→ 验证：端到端完成一个"从文案到PPT到配图"的完整任务 = 成功Level 3: 生产级（4周+）──────────────────────────────✅ Level 2 全部✅ Skill 注册中心 + 自动发现✅ 用户自定义 Agent 界面✅ 监控告警（Prometheus + Grafana）✅ 自动扩缩容✅ 安全加固（鉴权、审计日志）→ 验证：多用户并发使用、7×24 稳定运行 = 成功3.2 稳定性设计
3.2.1 模型服务高可用
# docker-compose.ha.yml — 高可用模型服务version: '3.8'services: # Nginx 负载均衡 model-lb: image: nginx:alpine ports: - "8000:80" volumes: - ./nginx-model.conf:/etc/nginx/nginx.conf depends_on: - vllm-1 - vllm-2 # 模型服务实例 1 vllm-1: image: vllm/vllm-openai:latest runtime: nvidia environment: - CUDA_VISIBLE_DEVICES=0 command: > --model /models/qwen2.5-72b --served-model-name qwen2.5-72b --max-model-len 16384 --port 8000 # 模型服务实例 2（冗余） vllm-2: image: vllm/vllm-openai:latest runtime: nvidia environment: - CUDA_VISIBLE_DEVICES=1 command: > --model /models/qwen2.5-72b --served-model-name qwen2.5-72b --max-model-len 16384 --port 8000# nginx-model.confevents { worker_connections 1024; }http { upstream model_backend { least_conn; server vllm-1:8000 max_fails=3 fail_timeout=30s; server vllm-2:8000 max_fails=3 fail_timeout=30s; } server { listen 80; location / { proxy_pass http://model_backend; proxy_connect_timeout 60s; proxy_read_timeout 300s; # LLM 生成可能较慢 proxy_next_upstream error timeout http_502 http_503; } }}3.2.2 健康检查与自动恢复
# health_monitor.py — 服务健康监控import requestsimport subprocessimport timeimport loggingfrom datetime import datetimelogging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")class HealthMonitor: """监控各服务健康状态，自动重启故障服务""" def __init__(self): self.services = { "vllm": { "url": "http://localhost:8000/v1/models", "container": "vllm", "critical": True, "fail_count": 0, "max_fails": 3, }, "dify": { "url": "http://localhost/api/health", "container": "dify-api-1", "critical": True, "fail_count": 0, "max_fails": 3, }, "comfyui": { "url": "http://localhost:8188/system_stats", "container": "comfyui", "critical": False, "fail_count": 0, "max_fails": 5, }, "ollama": { "url": "http://localhost:11434/api/tags", "container": "ollama", "critical": False, "fail_count": 0, "max_fails": 3, }, } def check_all(self): """检查所有服务""" for name, svc in self.services.items(): try: resp = requests.get(svc["url"], timeout=10) if resp.status_code == 200: svc["fail_count"] = 0 logging.debug(f"{name}: OK") else: self._handle_failure(name, svc, f"HTTP {resp.status_code}") except requests.exceptions.RequestException as e: self._handle_failure(name, svc, str(e)) def _handle_failure(self, name: str, svc: dict, error: str): svc["fail_count"] += 1 logging.warning(f"{name}: FAIL ({svc['fail_count']}/{svc['max_fails']}) - {error}") if svc["fail_count"] >= svc["max_fails"]: logging.error(f"{name}: Max failures reached, restarting container...") self._restart_container(svc["container"]) svc["fail_count"] = 0 def _restart_container(self, container: str): try: subprocess.run(["docker", "restart", container], check=True, timeout=120) logging.info(f"Container {container} restarted successfully") except subprocess.CalledProcessError as e: logging.error(f"Failed to restart {container}: {e}") def run_loop(self, interval_seconds: int = 30): """持续监控循环""" logging.info("Health monitor started") while True: self.check_all() time.sleep(interval_seconds)if __name__ == "__main__": monitor = HealthMonitor() monitor.run_loop(interval_seconds=30)3.2.3 请求队列与限流
# request_queue.py — 防止模型过载import asynciofrom collections import dequefrom dataclasses import dataclass, fieldfrom typing import Anyimport time@dataclassclass QueuedRequest: priority: int # 0=highest timestamp: float = field(default_factory=time.time) payload: dict = field(default_factory=dict) future: asyncio.Future = field(default_factory=lambda: asyncio.get_event_loop().create_future())class RequestQueue: """ 带优先级的请求队列，防止模型服务过载 - 限制并发请求数 - 超时自动取消 - 优先级调度 """ def __init__(self, max_concurrent: int = 4, timeout: float = 120.0): self.max_concurrent = max_concurrent self.timeout = timeout self.active_count = 0 self.queue: list = [] # 优先级队列 async def submit(self, payload: dict, priority: int = 5) -> Any: req = QueuedRequest(priority=priority, payload=payload) self.queue.append(req) self.queue.sort(key=lambda r: (r.priority, r.timestamp)) # 等待处理 asyncio.create_task(self._process_queue()) return await asyncio.wait_for(req.future, timeout=self.timeout) async def _process_queue(self): while self.queue and self.active_count < self.max_concurrent: req = self.queue.pop(0) self.active_count += 1 try: result = await self._call_model(req.payload) req.future.set_result(result) except Exception as e: req.future.set_exception(e) finally: self.active_count -= 1 async def _call_model(self, payload: dict) -> dict: """实际调用模型 API""" import aiohttp async with aiohttp.ClientSession() as session: async with session.post( "http://localhost:8000/v1/chat/completions", json=payload, timeout=aiohttp.ClientTimeout(total=120) ) as resp: return await resp.json()3.2.4 监控看板
# docker-compose.monitoring.ymlservices: prometheus: image: prom/prometheus:latest ports: - "9090:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml grafana: image: grafana/grafana:latest ports: - "3000:3000" environment: GF_SECURITY_ADMIN_PASSWORD: admin123 volumes: - ./grafana_data:/var/lib/grafana# prometheus.ymlglobal: scrape_interval: 15sscrape_configs: - job_name: 'vllm' static_configs: - targets: ['vllm:8000'] - job_name: 'node' static_configs: - targets: ['node-exporter:9100'] - job_name: 'nvidia-gpu' static_configs: - targets: ['nvidia-gpu-exporter:9400']3.3 GPU 显存管理策略
# gpu_manager.py — 智能 GPU 显存调度class GPUMemoryManager: """ 多模型共享 GPU 时的显存管理策略 场景：单卡/双卡 4090 要同时跑 LLM + 图像 + 视频 策略：分时复用 —— 不同时加载所有模型 """ # 策略 1: 常驻 LLM + 按需加载多模态 STRATEGY_LLM_RESIDENT = { "resident": ["qwen2.5-72b"], # 常驻显存 "on_demand": ["flux", "wan2.1"], # 按需加载/卸载 "preemption": True, # 允许抢占 } # 策略 2: 分时段调度 STRATEGY_TIME_BASED = { "schedules": [ {"hours": "08:00-22:00", "models": ["qwen2.5-72b", "flux"]}, # 白天：对话 + 图像 {"hours": "22:00-08:00", "models": ["wan2.1"]}, # 夜间：批量视频生成 ] } # 策略 3: 队列优先级 STRATEGY_PRIORITY = { "priorities": { "chat": 1, # 最高：实时对话 "text2img": 2, # 中等：图像生成 "text2video": 3, # 最低：视频生成（耗时长） } }第四部分：用户自定义 Agent
4.1 Agent 创建工作流
用户自定义 Agent 创建流程：Step 1: 基本信息┌─────────────────────────────────────────┐│ Agent 名称: [我的法律助手 ] ││ 头像: [📚] ││ 描述: [专注劳动法咨询的AI助手 ] ││ 可见性: ○ 仅自己 ● 团队 ○ 公开 │└─────────────────────────────────────────┘Step 2: 选择基础能力┌─────────────────────────────────────────┐│ ☑ 对话能力（必选） ││ ☑ 知识库检索 ││ ☐ 文生图 ││ ☐ 代码执行 ││ ☑ 网页搜索 ││ ☐ 文件生成（PPT/Word/PDF） │└─────────────────────────────────────────┘Step 3: 配置 Persona（系统提示词）┌─────────────────────────────────────────┐│ 你是一名资深劳动法律师，具有10年执业经验。││ 你需要： ││ - 用通俗易懂的语言解释法律条文 ││ - 给出具体可操作的建议 ││ - 引用相关法律条款和判例 ││ - 在不确定时明确告知需要线下咨询 ││ ││ [使用AI帮我生成提示词] │└─────────────────────────────────────────┘Step 4: 挂载知识库┌─────────────────────────────────────────┐│ ☑ 劳动法全文.pdf ││ ☑ 劳动合同法.pdf ││ ☑ 最新劳动争议司法解释.pdf ││ ☑ 典型劳动纠纷案例100例.pdf ││ [+ 上传新文档] [+ 连接在线知识库] │└─────────────────────────────────────────┘Step 5: 选择 Skills┌─────────────────────────────────────────┐│ ☑ web_search（联网搜索最新判例） ││ ☑ text2doc（生成法律文书模板） ││ ☑ summary（长文摘要） ││ ☐ text2img ││ ☐ code_generate ││ [+ 浏览 Skill 市场] │└─────────────────────────────────────────┘Step 6: 测试 & 发布┌─────────────────────────────────────────┐│ [预览对话] [运行测试集] [发布] │└─────────────────────────────────────────┘4.2 Agent 配置格式（YAML DSL）
# agents/legal_assistant.agent.yamlapiVersion: v1kind: Agentmetadata: name: legal-assistant display_name: 法律助手 description: 专注劳动法咨询的AI助手 icon: "📚" author: user123 visibility: team # private | team | public tags: ["法律", "劳动法", "咨询"]spec: # 使用的模型 model: provider: local-vllm name: qwen2.5-72b temperature: 0.3 # 法律领域需要低温度，减少幻觉 max_tokens: 4096 # 系统提示词 persona: | 你是一名资深劳动法律师，具有10年执业经验。 回答问题时： 1. 先判断问题属于哪个法律领域 2. 引用具体法律条款（如《劳动合同法》第XX条） 3. 给出通俗易懂的解释 4. 提供具体可操作的建议 5. 在涉及复杂案件时，建议线下咨询专业律师 禁止： - 给出明确的案件胜败判断 - 替代律师做出法律决策 - 编造不存在的法律条文 # 知识库 knowledge_bases: - dataset_id: kb-labor-law description: 劳动法全文及司法解释 retrieval_mode: hybrid # vector + keyword top_k: 5 score_threshold: 0.6 - dataset_id: kb-cases description: 典型劳动纠纷案例库 retrieval_mode: vector top_k: 3 score_threshold: 0.7 # 挂载的 Skills / Tools skills: - name: web_search config: search_engine: bing max_results: 5 domains: ["court.gov.cn", "law.cn"] # 限制搜索域名 - name: text2doc config: templates: ["labor_contract", "resignation_letter", "arbitration_application"] - name: summary config: max_length: 500 # 对话设置 conversation: opening_message: "你好！我是法律助手，专注于劳动法领域。请描述你的问题，我会尽力帮你分析。" suggested_questions: - "公司不签劳动合同怎么办？" - "被无故辞退可以要求哪些赔偿？" - "加班费怎么计算？" # 对话记忆 memory: type: window # window | summary | full window_size: 20 # 保留最近20轮对话 # 工作流（可选，复杂 Agent 使用） workflow: - step: classify_intent type: llm prompt: "判断用户问题属于哪个法律类别：劳动合同、薪酬福利、工伤、离职、劳动仲裁、其他" - step: retrieve_knowledge type: knowledge_retrieval dataset: auto # 根据分类自动选择知识库 top_k: 5 - step: generate_answer type: llm prompt: "根据检索到的法律知识，回答用户问题。必须引用具体法条。" - step: safety_check type: llm prompt: "检查回答是否包含不当法律建议。如有，添加免责声明。" # 安全防护 guardrails: input_filters: - type: sensitive_words action: reject words: ["怎么钻法律空子", "如何逃避"] output_filters: - type: disclaimer trigger: "specific_legal_advice" message: "⚠️ 以上分析仅供参考，具体案件建议咨询专业律师。"4.3 Agent 管理后台 API
# agent_manager.py — Agent 生命周期管理from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom typing import Optionalimport yamlimport uuidfrom pathlib import Pathapp = FastAPI(title="Agent Manager")AGENTS_DIR = Path("./agents")AGENTS_DIR.mkdir(exist_ok=True)class AgentCreateRequest(BaseModel): name: str display_name: str description: str model: str = "qwen2.5-72b" temperature: float = 0.7 persona: str knowledge_base_ids: list[str] = [] skill_names: list[str] = [] visibility: str = "private" opening_message: Optional[str] = None suggested_questions: list[str] = []class AgentResponse(BaseModel): id: str name: str display_name: str status: str@app.post("/api/agents", response_model=AgentResponse)async def create_agent(req: AgentCreateRequest): """创建新 Agent""" agent_id = f"agent-{uuid.uuid4().hex[:8]}" # 构建 Agent 配置 config = { "apiVersion": "v1", "kind": "Agent", "metadata": { "id": agent_id, "name": req.name, "display_name": req.display_name, "description": req.description, "visibility": req.visibility, }, "spec": { "model": {"name": req.model, "temperature": req.temperature}, "persona": req.persona, "knowledge_bases": [{"dataset_id": kb_id} for kb_id in req.knowledge_base_ids], "skills": [{"name": s} for s in req.skill_names], "conversation": { "opening_message": req.opening_message or f"你好！我是{req.display_name}，有什么可以帮你的？", "suggested_questions": req.suggested_questions, }, }, } # 保存配置 config_file = AGENTS_DIR / f"{agent_id}.yaml" config_file.write_text(yaml.dump(config, allow_unicode=True, default_flow_style=False)) # 同步注册到 Dify（通过 Dify API 创建应用） await _register_in_dify(config) return AgentResponse(id=agent_id, name=req.name, display_name=req.display_name, status="active")@app.get("/api/agents")async def list_agents(visibility: Optional[str] = None): """列出所有 Agent""" agents = [] for f in AGENTS_DIR.glob("*.yaml"): config = yaml.safe_load(f.read_text()) meta = config.get("metadata", {}) if visibility and meta.get("visibility") != visibility: continue agents.append({ "id": meta.get("id"), "name": meta.get("name"), "display_name": meta.get("display_name"), "description": meta.get("description"), "visibility": meta.get("visibility"), }) return {"agents": agents}@app.put("/api/agents/{agent_id}")async def update_agent(agent_id: str, req: AgentCreateRequest): """更新 Agent 配置""" config_file = AGENTS_DIR / f"{agent_id}.yaml" if not config_file.exists(): raise HTTPException(status_code=404, detail="Agent not found") # ... 更新逻辑（类似 create） return {"status": "updated"}@app.delete("/api/agents/{agent_id}")async def delete_agent(agent_id: str): """删除 Agent""" config_file = AGENTS_DIR / f"{agent_id}.yaml" if config_file.exists(): config_file.unlink() return {"status": "deleted"}@app.post("/api/agents/{agent_id}/clone")async def clone_agent(agent_id: str, new_name: str): """克隆现有 Agent 作为模板""" config_file = AGENTS_DIR / f"{agent_id}.yaml" if not config_file.exists(): raise HTTPException(status_code=404, detail="Agent not found") config = yaml.safe_load(config_file.read_text()) new_id = f"agent-{uuid.uuid4().hex[:8]}" config["metadata"]["id"] = new_id config["metadata"]["name"] = new_name new_file = AGENTS_DIR / f"{new_id}.yaml" new_file.write_text(yaml.dump(config, allow_unicode=True)) return {"id": new_id, "status": "cloned"}async def _register_in_dify(config: dict): """将 Agent 配置注册到 Dify 平台""" # 通过 Dify API 创建对应的应用 # POST /api/v1/apps pass# 启动: uvicorn agent_manager:app --host 0.0.0.0 --port 87004.4 Agent 模板市场
┌─────────────────────────────────────────────────────────────┐│ Agent 模板市场 │├──────────────┬──────────────────────────────────────────────┤│ 🔥 热门模板 │ ││ │ 📝 新媒体运营助手 ││ │ 自动写推文、配图、排版，支持多平台风格 ││ │ Skills: copywriting, text2img, seo ││ │ ⭐ 4.8 | 1.2k 使用 ││ │ [使用此模板] ││ │ ││ │ 💻 全栈开发助手 ││ │ 代码生成、Review、测试、Debug 全流程 ││ │ Skills: code_generate, test_gen, debug ││ │ ⭐ 4.9 | 2.5k 使用 ││ │ [使用此模板] ││ │ ││ │ 🎨 设计师助手 ││ │ UI设计、Logo生成、配色方案、原型图 ││ │ Skills: text2img, img2img, remove_bg ││ │ ⭐ 4.6 | 800 使用 ││ │ [使用此模板] ││ │ ││ │ 📊 数据分析师 ││ │ SQL生成、图表绘制、报告撰写 ││ │ Skills: sql_gen, code_execute, text2doc ││ │ ⭐ 4.7 | 950 使用 ││ │ [使用此模板] ││ │ ││ │ 📚 学术研究助手 ││ │ 论文检索、文献综述、格式排版 ││ │ Skills: web_search, summary, text2doc ││ │ ⭐ 4.5 | 600 使用 ││ │ [使用此模板] │├──────────────┼──────────────────────────────────────────────┤│ 📂 分类 │ 🏢 办公效率 | 💻 软件开发 | 🎨 设计创意 ││ │ 📊 数据分析 | 📚 教育学习 | 🏥 健康医疗 ││ │ ⚖️ 法律顾问 | 💰 财务税务 | 🎮 娱乐休闲 │└──────────────┴──────────────────────────────────────────────┘4.5 Agent 间协作（Team 模式）
# teams/content_team.team.yamlapiVersion: v1kind: Teammetadata: name: content-production-team display_name: 内容生产团队 description: 从选题到发布的全流程内容生产spec: # 团队成员（都是 Agent） members: - agent: copywriter role: 文案撰写 description: 负责撰写文章正文 - agent: image-designer role: 配图设计 description: 为文章生成配图 - agent: video-creator role: 视频制作 description: 将文章内容转为短视频 - agent: editor role: 主编审核 description: 审核内容质量，给出修改意见 # 协作流程 workflow: - phase: "1. 选题策划" agent: editor action: "根据用户需求，确定文章选题、角度和大纲" output: topic_outline - phase: "2. 文案撰写" agent: copywriter action: "根据大纲撰写完整文章" input: topic_outline output: draft_article - phase: "3. 内容审核" agent: editor action: "审核文章质量，给出修改意见" input: draft_article output: review_feedback loop_until: "approved" # 循环直到审核通过 - phase: "4. 配图生成" agent: image-designer action: "根据文章内容生成 3-5 张配图" input: draft_article output: images parallel: true # 与下一步并行 - phase: "5. 视频制作" agent: video-creator action: "将文章核心内容制作为 60 秒短视频" input: draft_article output: video parallel: true # 与上一步并行 - phase: "6. 最终整合" agent: editor action: "整合文章、配图、视频，输出最终成品" input: [draft_article, images, video] output: final_package # 协作规则 rules: max_revision_rounds: 3 # 最多修改3轮 timeout_per_phase: 300 # 每阶段最长5分钟 fallback_on_failure: skip # 某环节失败时跳过（而非中断）4.6 用户自定义 Agent 的 Prompt 工程辅助
# prompt_engineer.py — 帮助用户生成高质量系统提示词class PromptEngineer: """辅助用户创建 Agent 的系统提示词""" TEMPLATE = """请根据以下信息，生成一个专业的 AI Agent 系统提示词：角色名称: {role_name}领域: {domain}目标用户: {target_users}核心功能: {core_functions}语气风格: {tone}限制条件: {restrictions}要求：1. 明确角色定位和专业背景2. 列出具体的行为准则（DO 和 DON'T）3. 定义输出格式偏好4. 包含错误处理策略（不确定时怎么做）5. 加入安全防护指令（防止 prompt 注入、越权回答）6. 总长度控制在 300-500 字""" def generate_persona(self, role_name: str, domain: str, target_users: str, core_functions: str, tone: str = "专业友好", restrictions: str = "不编造信息") -> str: prompt = self.TEMPLATE.format( role_name=role_name, domain=domain, target_users=target_users, core_functions=core_functions, tone=tone, restrictions=restrictions ) resp = requests.post("http://localhost:8000/v1/chat/completions", json={ "model": "qwen2.5-72b", "messages": [{"role": "user", "content": prompt}], "temperature": 0.7 }) return resp.json()["choices"][0]["message"]["content"] def optimize_persona(self, current_persona: str, user_feedback: str) -> str: """根据用户反馈优化提示词""" prompt = f"""当前系统提示词：{current_persona}用户反馈：{user_feedback}请优化系统提示词，解决用户反馈中提到的问题。保持原有优点，改进不足之处。""" resp = requests.post("http://localhost:8000/v1/chat/completions", json={ "model": "qwen2.5-72b", "messages": [{"role": "user", "content": prompt}], "temperature": 0.5 }) return resp.json()["choices"][0]["message"]["content"]第五部分：完整目录结构
ai-agent-platform/├── docker-compose.yml # 主部署文件├── docker-compose.ha.yml # 高可用配置├── docker-compose.monitoring.yml # 监控配置├── .env # 环境变量│├── models/ # 模型文件│ ├── qwen2.5-72b/│ ├── qwen2.5-coder-32b/│ └── bge-m3/│├── services/ # 工具微服务│ ├── ppt/│ │ ├── Dockerfile│ │ ├── requirements.txt│ │ └── ppt_service.py│ ├── video/│ │ ├── Dockerfile│ │ ├── requirements.txt│ │ └── video_service.py│ └── skill_runner/│ ├── Dockerfile│ └── skill_runner.py│├── mcp_servers/ # MCP Server│ ├── mcp_filesystem.py│ ├── mcp_comfyui.py│ └── mcp_tools.py│├── skills/ # Skill 定义│ ├── text2ppt.skill.json│ ├── text2img.skill.json│ ├── code_generate.skill.json│ ├── web_search.skill.json│ └── ...│├── agents/ # Agent 配置│ ├── copywriter.agent.yaml│ ├── coder.agent.yaml│ ├── designer.agent.yaml│ └── ...│├── teams/ # Team 配置│ └── content_team.team.yaml│├── knowledge/ # 知识库原始文件│ ├── coding/│ ├── writing/│ ├── creative/│ ├── life/│ └── custom/ # 用户自定义知识库│├── knowledge_scrapers/ # 知识采集器│ ├── coding_scraper.py│ ├── writing_scraper.py│ ├── creative_scraper.py│ └── __init__.py│├── core/ # 核心模块│ ├── knowledge_pipeline.py # 知识库同步流水线│ ├── knowledge_quality.py # 质量检测│ ├── skill_discovery.py # Skill 发现引擎│ ├── skill_to_dify_tool.py # Skill → Dify 工具转换│ ├── agent_manager.py # Agent 管理 API│ ├── prompt_engineer.py # Prompt 工程辅助│ ├── request_queue.py # 请求队列│ ├── gpu_manager.py # GPU 调度│ └── health_monitor.py # 健康监控│├── nginx/│ └── nginx-model.conf│├── prometheus/│ └── prometheus.yml│├── comfyui_models/ # ComfyUI 模型文件│ ├── checkpoints/│ ├── unet/│ └── loras/│├── comfyui_workflows/ # ComfyUI 工作流模板│ ├── text_to_image_flux.json│ ├── image_to_video_wan21.json│ └── ...│└── scripts/ # 运维脚本 ├── setup.sh # 一键安装 ├── download_models.sh # 模型下载 ├── backup.sh # 备份 └── update.sh # 更新第六部分：一键启动脚本
#!/bin/bash# scripts/setup.sh — 一键安装部署脚本set -eecho "========================================="echo " AI Agent Platform - One-Click Setup"echo "========================================="# 检查前置条件check_prerequisites() { echo "[1/8] Checking prerequisites..." # Docker if ! command -v docker &> /dev/null; then echo "Installing Docker..." curl -fsSL https://get.docker.com | sh sudo usermod -aG docker $USER fi echo " ✅ Docker: $(docker --version)" # Docker Compose if ! command -v docker compose &> /dev/null; then echo "ERROR: Docker Compose not found" exit 1 fi echo " ✅ Docker Compose: available" # NVIDIA Driver if command -v nvidia-smi &> /dev/null; then echo " ✅ NVIDIA Driver: $(nvidia-smi --query-gpu=driver_version --format=csv,noheader | head -1)" echo " ✅ GPU: $(nvidia-smi --query-gpu=name --format=csv,noheader)" echo " ✅ VRAM: $(nvidia-smi --query-gpu=memory.total --format=csv,noheader)" else echo " ⚠️ No NVIDIA GPU detected, will use CPU mode (slow)" fi # NVIDIA Container Toolkit if docker run --rm --gpus all nvidia/cuda:12.0-base nvidia-smi &> /dev/null; then echo " ✅ NVIDIA Container Toolkit: working" else echo " Installing NVIDIA Container Toolkit..." distribution=$(. /etc/os-release; echo $ID$VERSION_ID) curl -s -L https://nvidia.github.io/nvidia-docker/gpgkey | sudo apt-key add - curl -s -L "https://nvidia.github.io/nvidia-docker/$distribution/nvidia-docker.list" | \ sudo tee /etc/apt/sources.list.d/nvidia-docker.list sudo apt-get update && sudo apt-get install -y nvidia-container-toolkit sudo systemctl restart docker fi}# 创建目录结构create_directories() { echo "[2/8] Creating directory structure..." mkdir -p models services/ppt services/video mcp_servers skills agents teams mkdir -p knowledge/{coding,writing,creative,life,custom} mkdir -p knowledge_scrapers core scripts mkdir -p comfyui_models/{checkpoints,unet,loras} mkdir -p comfyui_workflows nginx prometheus mkdir -p ollama_data weaviate_data minio_data grafana_data outputs echo " ✅ Directories created"}# 下载模型download_models() { echo "[3/8] Downloading models (this may take a while)..." # 检测可用显存，选择合适的模型 if command -v nvidia-smi &> /dev/null; then VRAM=$(nvidia-smi --query-gpu=memory.total --format=csv,noheader,nounits | head -1) GPU_COUNT=$(nvidia-smi --query-gpu=count --format=csv,noheader | head -1) TOTAL_VRAM=$((VRAM * GPU_COUNT)) if [ "$TOTAL_VRAM" -ge 48000 ]; then MODEL="Qwen/Qwen2.5-72B-Instruct-GPTQ-Int4" echo " Using Qwen2.5-72B (${TOTAL_VRAM}MB VRAM detected)" elif [ "$TOTAL_VRAM" -ge 24000 ]; then MODEL="Qwen/Qwen2.5-32B-Instruct-AWQ" echo " Using Qwen2.5-32B (${TOTAL_VRAM}MB VRAM detected)" else MODEL="Qwen/Qwen2.5-14B-Instruct-AWQ" echo " Using Qwen2.5-14B (${TOTAL_VRAM}MB VRAM detected)" fi else MODEL="Qwen/Qwen2.5-7B-Instruct" echo " Using Qwen2.5-7B (CPU mode)" fi pip install modelscope -q modelscope download --model "$MODEL" --local_dir ./models/qwen2.5 echo " ✅ Language model downloaded"}# 启动核心服务start_services() { echo "[4/8] Starting core services..." docker compose up -d echo " ✅ Services starting..." # 等待服务就绪 echo " Waiting for services to be ready..." for i in {1..60}; do if curl -s http://localhost:8000/v1/models > /dev/null 2>&1; then echo " ✅ Model service ready" break fi sleep 5 done}# 初始化 Difyinit_dify() { echo "[5/8] Initializing Dify..." cd dify/docker && docker compose up -d && cd ../.. echo " ✅ Dify started at http://localhost" echo " 📌 Visit http://localhost/install to complete setup"}# 初始化知识库init_knowledge() { echo "[6/8] Initializing knowledge base..." python core/knowledge_pipeline.py --init echo " ✅ Knowledge base initialized"}# 注册 Skillsregister_skills() { echo "[7/8] Registering skills..." for skill_file in skills/*.skill.json; do echo " Registering: $skill_file" done echo " ✅ Skills registered"}# 完成finish() { echo "[8/8] Setup complete!" echo "" echo "=========================================" echo " 🎉 AI Agent Platform is ready!" echo "=========================================" echo "" echo " Services:" echo " - Dify Web UI: http://localhost" echo " - Model API: http://localhost:8000" echo " - ComfyUI: http://localhost:8188" echo " - Agent Manager: http://localhost:8700" echo " - MinIO Console: http://localhost:9001" echo " - Grafana: http://localhost:3000" echo "" echo " Next steps:" echo " 1. Visit http://localhost/install to set up Dify" echo " 2. Add your local model in Dify Settings" echo " 3. Create your first Agent!" echo ""}# 执行check_prerequisitescreate_directoriesdownload_modelsstart_servicesinit_difyinit_knowledgeregister_skillsfinish
