Posts

延迟的两个关键指标 LLM服务的延迟分为两部分： TTFT（Time To First Token）：从请求到第一个token返回的时间 TPOT（Time Per Output Token）：每个后续token的生成时间 用户感知延迟 = TTFT + (输出token数 × TPOT)。优化需要分别针对这两个指标。 TTFT优化 预填充优化 TTFT主要由预填充（处理输入prompt）时间决定： # 1. 分块预填充：避免长prompt阻塞短请求 vllm serve model --enable-chunked-prefill --max-num-batched-tokens 4096 # 2. 前缀缓存：共享系统提示词的KV Cache vllm serve model --enable-prefix-caching # 3. 减少输入长度：精简系统提示词 # 差：500 token的系统提示词 # 好：150 token的精简系统提示词 模型预热 async def warmup_model(model_name): """服务启动时预热模型""" # 预加载模型到GPU dummy_input = "warmup" await llm.generate(dummy_input, max_tokens=1) # 预填充常见前缀 common_prompts = [ "你是一个专业助手", "请根据以下信息回答问题", ] for prompt in common_prompts: await llm.generate(prompt, max_tokens=1) TPOT优化 推测解码 # vLLM启用推测解码 vllm serve model \ --speculative-model /models/draft-model \ --num-speculative-tokens 5 \ --speculative-draft-tensor-parallel-size 1 量化 # INT8量化（2倍加速，精度损失<1%） vllm serve model --quantization awq --dtype float16 # INT4量化（3倍加速，精度损失3-5%） vllm serve model --quantization gptq --dtype float16 KV Cache优化 # FP8 KV Cache（减少显存带宽压力） vllm serve model --kv-cache-dtype fp8 网络层优化 流式响应 # 流式响应让用户更早看到输出 @app.post("/chat") async def chat(): async def stream(): yield "data: " # 立即发送头部 async for token in llm.astream(messages): yield json.dumps({"content": token}) + "\n" return StreamingResponse(stream()) 连接复用 # 使用HTTP/2或WebSocket减少连接建立开销 import httpx # 全局复用客户端 client = httpx.AsyncClient( http2=True, limits=httpx.Limits(max_connections=100, max_keepalive_connections=20), keepalive_expiry=30, ) 调度优化 优先级调度 class PriorityScheduler: """交互请求优先于批处理请求""" async def schedule(self, request): if request.type == "interactive": # 交互请求：立即处理 return await self.process_immediately(request) else: # 批处理请求：低峰期处理 return await self.queue_for_later(request) 请求预取 class RequestPrefetcher: """预测用户可能的下一步请求，提前计算""" async def on_user_typing(self, session_id): """用户正在输入时，预计算可能的请求""" likely_queries = await self.predict_queries(session_id) for query in likely_queries[:2]: # 预计算2个最可能的查询 cache_key = f"prefetch:{session_id}:{query}" if not await self.cache.exists(cache_key): result = await self.llm.generate(query) await self.cache.setex(cache_key, 30, result) 端到端优化清单 优化项 TTFT改善 TPOT改善 实现难度 分块预填充 30-50% — 低 前缀缓存 40-60% — 低 量化 10-20% 50-100% 中 推测解码 — 50-100% 中 流式响应 感知80%↓ — 低 模型分层 20-40% 30-50% 中 监控 # 延迟分解监控 class LatencyBreakdown: def record(self, request_start, first_token, request_end): ttft = first_token - request_start total = request_end - request_start tpot = (total - ttft) / max(output_tokens - 1, 1) metrics = { "ttft_ms": ttft * 1000, "tpot_ms": tpot * 1000, "total_ms": total * 1000, } # 告警 if ttft > 2.0: # TTFT > 2秒 alert("High TTFT") if tpot > 0.05: # TPOT > 50ms alert("High TPOT") 结语 LLM服务延迟优化需要从TTFT和TPOT两个维度系统推进。分块预填充和前缀缓存是最有效的TTFT优化手段，量化和推测解码是TPOT的核心优化手段。流式响应虽然不改变实际延迟，但显著改善用户感知。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 命令行是开发者最自然的工作环境。2026年，AI命令行工具已经从简单的"命令行聊天"发展为一个完整的工具生态，涵盖代码生成、文档查询、数据处理、系统管理等多个领域。本文将介绍2026年最实用的AI命令行工具。 工具一：aichat — 终端AI助手 # 安装 cargo install aichat # 基本使用 aichat "解释什么是Rust的所有权机制" # 多模型支持 aichat -m glm-5 "你好" aichat -m claude-4-opus "写一个函数" # 管道使用 cat error.log | aichat "分析这个错误日志" 配置 # ~/.config/aichat/config.yaml model: glm-5 api_base: http://localhost:11434/v1 temperature: 0.7 max_tokens: 2048 models: glm-5: provider: ollama model: glm-5:32b gpt-5: provider: openai model: gpt-5 工具二：shell-genie — 自然语言命令 # 安装 pip install shell-genie # 自然语言转命令 shell-genie ask "找出当前目录下最大的10个文件" # 输出: du -ah . | sort -rh | head -10 # 执行? [Y/n] # 解释命令 shell-genie explain "awk '{print $2}' file.txt | sort -u" # 输出: 提取file.txt的第二列，排序并去重 工具三：commit-assistant — AI提交助手 # 安装 npm install -g ai-commit # 自动生成提交信息 git add . ai-commit # 输出: "feat: 添加用户认证模块，支持JWT和OAuth2" # 指定风格 ai-commit --style conventional # Conventional Commits ai-commit --style zh # 中文提交 工具四：code-review-cli — 代码审查 # 安装 pip install ai-code-review # 审查当前改动 git diff | ai-code-review # 输出: # 🔴 严重问题：SQL注入风险（第23行） # 🟡 建议：添加错误处理（第45行） # 🟢 良好实践：使用了参数化查询 # 审查特定文件 ai-code-review src/auth.py 工具五：ai-docs — 文档生成 # 安装 pip install ai-docs # 从代码生成文档 ai-docs generate src/ # 输出: 自动生成API文档 # 从README生成API文档 ai-docs api --input README.md --output api-docs.md # 生成变更日志 ai-docs changelog --from v1.0.0 --to v2.0.0 工具六：translate-cli — 翻译工具 # 安装 pip install ai-translate # 翻译文本 translate "Hello, world" --to zh # 你好，世界 # 翻译文件 translate --file README.md --to zh --output README_ZH.md # 实时翻译管道 echo "Hello" | translate --to ja 工具七：ai-sql — SQL助手 # 安装 pip install ai-sql # 自然语言转SQL ai-sql "查询上个月销售额前10的产品" # SELECT product_name, SUM(quantity * price) as total_sales # FROM orders # WHERE created_at >= DATE_SUB(NOW(), INTERVAL 1 MONTH) # GROUP BY product_name # ORDER BY total_sales DESC # LIMIT 10; # 解释SQL ai-sql explain "SELECT ... FROM ..." # 这个查询的作用是... # 优化SQL ai-sql optimize "SELECT ... FROM ..." # 建议：添加索引 idx_user_id_created_at 工具八：ai-data — 数据分析 # 安装 pip install ai-data # 分析CSV ai-data analyze sales.csv # 输出: # 数据概览：1000行×15列 # 缺失值：3列有缺失 # 异常值：5个 # 趋势分析：销售额呈上升趋势 # 建议：考虑填充缺失的price列 # 数据可视化 ai-data plot sales.csv --x date --y amount --type line 工具九：ai-test — 测试生成 # 安装 npm install -g ai-test-gen # 生成单元测试 ai-test generate src/auth.py # 输出: tests/test_auth.py # 生成测试用例 ai-test cases "用户注册流程" # 输出: # 1. 正常注册 # 2. 重复用户名 # 3. 无效邮箱 # 4. 密码过短 # ... 工具十：ai-grep — 语义搜索 # 安装 pip install ai-grep # 语义搜索代码 ai-grep "处理用户认证的代码" # 输出: # src/auth/login.py:15 - def authenticate(username, password): # src/middleware/auth.py:8 - class AuthMiddleware: # src/api/users.py:32 - @require_auth # 语义搜索文档 ai-grep --type docs "如何部署" # 输出: # docs/deployment.md:1 - # 部署指南 # README.md:45 - ## 快速部署 工具十一：aichat-config — 多模型管理 # 模型切换 aichat config set model glm-5 aichat config set model gpt-5 # 查看可用模型 aichat models # 模型对比 aichat compare "写一个快排" --models glm-5,gpt-5,claude-4 工具十二：ai-explain — 代码解释 # 安装 pip install ai-explain # 解释代码 ai-explain src/complex_algorithm.py # 输出: # 这个文件实现了Dijkstra最短路径算法 # 主要函数：find_shortest_path(graph, start, end) # 时间复杂度：O((V+E)logV) # 解释命令 ai-explain "tar -xzvf archive.tar.gz" # 解压.tar.gz格式的压缩包 组合使用 工作流：代码开发 # 1. 用自然语言搜索代码 ai-grep "用户登录逻辑" # 2. 查看并解释代码 ai-explain src/auth.py # 3. 生成测试 ai-test generate src/auth.py # 4. 审查改动 git diff | ai-code-review # 5. 生成提交信息 ai-commit 工作流：数据分析 # 1. 分析数据 ai-data analyze sales.csv # 2. 生成SQL查询 ai-sql "查询月度销售趋势" # 3. 可视化 ai-data plot sales.csv --x month --y total --type bar 工作流：文档编写 # 1. 从代码生成文档 ai-docs generate src/ # 2. 翻译文档 translate --file docs/api.md --to en # 3. 生成变更日志 ai-docs changelog 自定义工具 # 创建自定义AI命令 #!/bin/bash # ~/.local/bin/ai-debug ERROR=$1 aichat "作为调试专家，分析以下错误并给出解决方案：$ERROR" 性能优化 # 使用本地模型（零成本） export AI_CHAT_MODEL=ollama:glm-5:32b # 使用缓存 export AI_CACHE_ENABLED=true # 异步处理 ai-data analyze big.csv --async 安全注意 # 不要将敏感数据发送到云端AI # 使用本地模型处理敏感代码 ai-grep "密码" --model local-only # 或使用数据脱敏 ai-data analyze customers.csv --anonymize 结语 2026年的AI命令行工具已经非常丰富，覆盖了开发的各个环节。这些工具让AI成为开发者的"第二大脑"，在终端中即可完成代码搜索、生成、审查、测试等工作。 ...

AI成本的结构 AI系统的成本主要由以下几部分构成： API调用费：按token计费的LLM API费用（通常占60-70%） 基础设施：GPU服务器/云服务费用（自部署场景） 向量数据库：存储和检索费用 数据标注：人工标注和评估成本 工程开发：系统开发和维护的人力成本 模型分层策略 class ModelTierRouter: """根据任务复杂度路由到不同级别的模型""" def __init__(self): self.tiers = { "simple": {"model": "qwen3-7b", "cost_per_1k": 0.0005}, "medium": {"model": "qwen3-32b", "cost_per_1k": 0.002}, "complex": {"model": "qwen3-72b", "cost_per_1k": 0.008}, } def route(self, query, context=None): # 简单规则路由 if len(query) < 50 and "?" in query: return self.tiers["simple"] # 基于意图分类 complexity = self.estimate_complexity(query, context) if complexity < 0.3: return self.tiers["simple"] elif complexity < 0.7: return self.tiers["medium"] else: return self.tiers["complex"] def estimate_complexity(self, query, context): """估计查询复杂度""" score = 0 # 长查询更复杂 score += min(len(query) / 1000, 0.3) # 需要推理的关键词 reasoning_words = ["分析", "比较", "为什么", "推导", "计算"] score += 0.2 * any(w in query for w in reasoning_words) # 多轮上下文更复杂 if context and len(context) > 5: score += 0.2 # 代码生成更复杂 if "代码" in query or "函数" in query: score += 0.3 return min(score, 1.0) 缓存策略 class SemanticCache: """语义缓存：相似查询命中缓存""" def __init__(self, vector_store, similarity_threshold=0.95): self.store = vector_store self.threshold = similarity_threshold async def get(self, query): """查询缓存""" query_embedding = await self.embed(query) results = await self.store.search( query_embedding, top_k=1 ) if results and results[0].score > self.threshold: # 缓存命中 return json.loads(results[0].document) return None async def set(self, query, response, ttl=3600): """写入缓存""" embedding = await self.embed(query) await self.store.add( id=generate_uuid(), embedding=embedding, document=json.dumps({ "query": query, "response": response, "timestamp": time.time(), }), ttl=ttl ) Token优化 class TokenOptimizer: """减少token消耗的各种策略""" def compress_history(self, messages): """压缩对话历史""" if len(messages) <= 4: return messages # 保留最近2轮 + 摘要 recent = messages[-4:] old = messages[:-4] summary = self.summarize(old) return [{"role": "system", "content": f"历史摘要：{summary}"}] + recent def optimize_prompt(self, prompt): """精简提示词""" # 移除冗余空格和换行 prompt = re.sub(r'\n{3,}', '\n\n', prompt) prompt = re.sub(r' {2,}', ' ', prompt) return prompt.strip() def truncate_context(self, documents, max_tokens=2000): """智能截断文档""" result = [] current_tokens = 0 for doc in documents: doc_tokens = len(doc) // 4 if current_tokens + doc_tokens > max_tokens: # 截断最后一个文档而非完全丢弃 remaining = max_tokens - current_tokens if remaining > 100: result.append(doc[:remaining * 4]) break result.append(doc) current_tokens += doc_tokens return result 批处理优化 class BatchProcessor: """合并多个请求降低API调用次数""" async def batch_generate(self, requests): """将多个独立请求合并为一个""" # 合并多个查询为一个prompt combined_prompt = "请依次回答以下问题：\n\n" for i, req in enumerate(requests): combined_prompt += f"问题{i+1}：{req['query']}\n" combined_prompt += "\n请按问题编号分别回答。" response = await self.llm.generate(combined_prompt) # 解析响应 answers = self.parse_batch_response(response, len(requests)) return answers 成本监控 class CostMonitor: def __init__(self): self.daily_costs = defaultdict(float) self.user_costs = defaultdict(float) self.model_costs = defaultdict(float) def record(self, user_id, model, input_tokens, output_tokens): pricing = { "qwen3-7b": {"input": 0.0003, "output": 0.0006}, "qwen3-32b": {"input": 0.001, "output": 0.002}, "qwen3-72b": {"input": 0.004, "output": 0.008}, } rates = pricing.get(model, {"input": 0.001, "output": 0.002}) cost = (input_tokens / 1000 * rates["input"] + output_tokens / 1000 * rates["output"]) today = datetime.now().strftime("%Y-%m-%d") self.daily_costs[today] += cost self.user_costs[user_id] += cost self.model_costs[model] += cost def alert_if_over_budget(self, daily_budget=100): today = datetime.now().strftime("%Y-%m-%d") if self.daily_costs[today] > daily_budget: self.send_alert( f"日成本超预算：${self.daily_costs[today]:.2f} / ${daily_budget}" ) 成本优化效果 策略 节省比例 实现难度 模型分层 30-50% 低 语义缓存 20-40% 中 Token优化 10-20% 低 批处理 15-25% 中 本地部署替代 50-80% 高 结语 AI成本优化是一个持续过程。模型分层让简单任务用小模型，语义缓存避免重复计算，Token优化减少浪费，批处理提升效率。组合使用这些策略，可以在不降低用户体验的前提下将AI成本降低50-70%。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 LocalAI是一个完全开源的AI服务框架，提供OpenAI兼容的API，让你可以在自己的硬件上运行AI服务。2026年，LocalAI已经成为替代OpenAI API最完整的方案。本文将详细介绍LocalAI的自托管部署。 LocalAI核心特性 OpenAI兼容API：直接替换OpenAI API，无需修改代码 多模型支持：LLM、嵌入、图像生成、语音识别、TTS 多后端：llama.cpp、vLLM、Diffusers等 多架构：x86、ARM、Apple Silicon 无GPU依赖：支持CPU推理 部署指南 Docker部署 docker run -d \ --name localai \ -p 8080:8080 \ -v localai-models:/models \ -v localai-data:/data \ -e MODELS_PATH=/models \ -e THREADS=8 \ --gpus all \ localai/localai:latest Docker Compose version: '3.8' services: localai: image: localai/localai:latest ports: - "8080:8080" volumes: - ./models:/models - ./data:/data environment: - MODELS_PATH=/models - THREADS=8 - GPU_TYPE=nvidia deploy: resources: reservations: devices: - driver: nvidia count: all capabilities: [gpu] restart: always 从源码编译 git clone https://github.com/mudler/LocalAI cd LocalAI make build 模型配置 LLM模型 # models/glm-5.yaml name: glm-5 backend: llama parameters: model: glm-5-32b.gguf temperature: 0.7 top_p: 0.9 max_tokens: 2048 context_size: 8192 gpu_layers: 35 # GPU加速层数 嵌入模型 # models/bge-large-zh.yaml name: bge-large-zh backend: bert parameters: model: bge-large-zh-v2.gguf 图像生成模型 # models/sd4.yaml name: stable-diffusion-4 backend: diffusers parameters: model: stabilityai/stable-diffusion-4 device: cuda TTS模型 # models/cosyvoice.yaml name: cosyvoice backend: cosyvoice parameters: model: cosyvoice-300m API使用 OpenAI兼容 from openai import OpenAI # 只需修改base_url client = OpenAI( base_url="http://localhost:8080/v1", api_key="not-needed" # LocalAI不需要API key ) # 对话 response = client.chat.completions.create( model="glm-5", messages=[{"role": "user", "content": "你好"}] ) # 嵌入 embedding = client.embeddings.create( model="bge-large-zh", input="要嵌入的文本" ) # 图像生成 image = client.images.generate( model="stable-diffusion-4", prompt="一只可爱的猫" ) 直接API调用 # 对话 curl http://localhost:8080/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "glm-5", "messages": [{"role": "user", "content": "你好"}] }' # 嵌入 curl http://localhost:8080/v1/embeddings \ -H "Content-Type: application/json" \ -d '{ "model": "bge-large-zh", "input": "要嵌入的文本" }' 集成方案 与LangChain集成 from langchain_openai import ChatOpenAI llm = ChatOpenAI( base_url="http://localhost:8080/v1", api_key="not-needed", model="glm-5" ) 与LlamaIndex集成 from llama_index.llms.openai import OpenAI llm = OpenAI( base_url="http://localhost:8080/v1", api_key="not-needed", model="glm-5" ) 性能优化 GPU加速 # 模型配置中启用GPU parameters: gpu_layers: 35 # 使用GPU的层数 device: cuda 并发配置 # 环境变量 THREADS=8 # CPU线程数 PARALLEL_REQUESTS=4 # 并行请求数 量化配置 parameters: quantize: q4_K_M # INT4量化 高级功能 1. 模型自动加载 # 启动时自动加载模型 PRELOAD_MODELS=glm-5,bge-large-zh 2. 请求队列 # 管理并发请求 queue: max_size: 100 timeout: 300 3. 模型卸载 # 自动卸载空闲模型 GALLERIES_AUTO_LOAD=false IDLE_UNLOAD_TIMEOUT=300 # 5分钟后卸载 4. 多GPU parameters: tensor_parallel_size: 4 gpu_memory_fraction: 0.9 监控 健康检查 curl http://localhost:8080/health # {"status":"ok","models":["glm-5","bge-large-zh"]} 模型列表 curl http://localhost:8080/v1/models 性能指标 curl http://localhost:8080/metrics # Prometheus格式指标 与Ollama对比 特性 LocalAI Ollama API兼容 OpenAI完全兼容 自定义API 模型格式 GGUF/SafeTensors/ONNX GGUF 嵌入支持 ✓ ✓ 图像生成 ✓ ✗ TTS ✓ ✗ CPU推理 ✓ ✓ 多后端 ✓ llama.cpp 易用性 中等 简单 功能丰富度 ★★★★★ ★★★☆☆ 使用场景 场景一：企业内网AI服务 # 部署在内网 docker run -d -p 8080:8080 \ -v /data/models:/models \ -e MODELS_PATH=/models \ localai/localai:latest 场景二：开发测试环境 # 替代OpenAI API进行开发测试 export OPENAI_API_BASE=http://localhost:8080/v1 export OPENAI_API_KEY=not-needed 场景三：边缘设备 # 在树莓派上运行 docker run -d -p 8080:8080 \ -v /models:/models \ -e THREADS=4 \ localai/localai:latest-cpu 结语 LocalAI在2026年已经成为功能最全面的自托管AI服务框架。它不仅兼容OpenAI API，还支持图像生成、TTS等多种AI能力。对于需要完全掌控AI基础设施的团队，LocalAI是最佳选择。 ...

多轮对话的核心挑战 多轮对话不是简单的消息拼接——它需要管理对话状态、控制上下文窗口长度、处理话题切换、维护一致性。一个好的对话管理系统是LLM助手的"大脑"。 对话状态管理 from enum import Enum from dataclasses import dataclass, field class DialogState(Enum): GREETING = "greeting" INFORMATION_SEEKING = "information_seeking" PROBLEM_SOLVING = "problem_solving" CLARIFICATION = "clarification" CLOSING = "closing" @dataclass class ConversationContext: session_id: str user_id: str state: DialogState = DialogState.GREETING topic: str = "" entities: dict = field(default_factory=dict) # 提取的实体 history: list = field(default_factory=list) # 消息历史 user_preferences: dict = field(default_factory=dict) pending_clarification: str = "" class DialogManager: def __init__(self, llm): self.llm = llm self.sessions = {} # session_id -> ConversationContext def get_or_create_session(self, session_id, user_id): if session_id not in self.sessions: self.sessions[session_id] = ConversationContext( session_id=session_id, user_id=user_id ) return self.sessions[session_id] async def process_message(self, session_id, user_id, message): ctx = self.get_or_create_session(session_id, user_id) # 1. 状态检测 ctx.state = await self.detect_state(message, ctx) # 2. 实体提取 new_entities = await self.extract_entities(message) ctx.entities.update(new_entities) # 3. 构建上下文 context = self.build_context(ctx, message) # 4. 生成回复 response = await self.llm.generate(context) # 5. 更新历史 ctx.history.append({"role": "user", "content": message}) ctx.history.append({"role": "assistant", "content": response}) # 6. 上下文窗口管理 self.manage_window(ctx) return response 上下文窗口管理 class ContextWindowManager: def __init__(self, max_tokens=4096, reserve_tokens=1024): self.max_tokens = max_tokens self.reserve_tokens = reserve_tokens # 为生成预留 def manage(self, context): """管理上下文窗口大小""" available = self.max_tokens - self.reserve_tokens current_tokens = self.estimate_tokens(context.history) if current_tokens <= available: return # 在限制内，无需处理 # 策略1：摘要旧消息 while current_tokens > available and len(context.history) > 4: # 取最早的2条消息进行摘要 old_msgs = context.history[:2] summary = self.summarize(old_msgs) # 替换为摘要 context.history = [ {"role": "system", "content": f"[早期对话摘要]: {summary}"} ] + context.history[2:] current_tokens = self.estimate_tokens(context.history) # 策略2：如果仍然过长，截断 if current_tokens > available: # 保留system prompt和最近的消息 while current_tokens > available and len(context.history) > 2: context.history.pop(0 if context.history[0]["role"] != "system" else 1) current_tokens = self.estimate_tokens(context.history) def estimate_tokens(self, messages): return sum(len(m["content"]) // 4 for m in messages) def summarize(self, messages): """摘要旧消息""" text = "\n".join(f"{m['role']}: {m['content']}" for m in messages) # 使用LLM摘要 return f"[{len(messages)}条消息的摘要]" 话题管理 class TopicManager: def __init__(self, llm): self.llm = llm async def detect_topic_shift(self, current_message, history): """检测话题是否切换""" if len(history) < 2: return False, None recent_topic = await self.extract_topic(history[-2:]) current_topic = await self.extract_topic([{"content": current_message}]) similarity = await self.compute_similarity(recent_topic, current_topic) if similarity < 0.3: return True, current_topic # 话题切换 return False, recent_topic async def handle_topic_shift(self, ctx, new_topic): """处理话题切换""" # 保存当前话题的摘要 old_summary = await self.summarize_topic(ctx) ctx.topic_summaries = ctx.get("topic_summaries", []) ctx.topic_summaries.append({"topic": ctx.topic, "summary": old_summary}) # 切换到新话题 ctx.topic = new_topic ctx.state = DialogState.INFORMATION_SEEKING 意图跟踪 class IntentTracker: def __init__(self, llm): self.llm = llm self.intent_history = [] async def track(self, message, context): """跟踪用户意图""" prompt = f"""分析用户意图。 用户消息：{message} 对话历史摘要：{self.get_history_summary()} 当前状态：{context.state.value} 输出JSON： {{ "intent": "具体意图", "confidence": 0.0-1.0, "requires_clarification": true/false, "clarification_question": "如需澄清的问题" }}""" result = await self.llm.generate(prompt) intent = json.loads(result) self.intent_history.append(intent) return intent 会话持久化 import redis.asyncio as redis class SessionPersistence: def __init__(self, redis_url="redis://localhost:6379"): self.redis = redis.from_url(redis_url) async def save_session(self, session_id, context): """持久化会话""" await self.redis.setex( f"session:{session_id}", 86400, # 24小时过期 json.dumps({ "session_id": context.session_id, "user_id": context.user_id, "state": context.state.value, "topic": context.topic, "entities": context.entities, "history": context.history[-20:], # 只存最近20条 "user_preferences": context.user_preferences, }, ensure_ascii=False) ) async def load_session(self, session_id): """加载会话""" data = await self.redis.get(f"session:{session_id}") if data: d = json.loads(data) return ConversationContext( session_id=d["session_id"], user_id=d["user_id"], state=DialogState(d["state"]), topic=d["topic"], entities=d["entities"], history=d["history"], user_preferences=d["user_preferences"], ) return None 实践建议 会话超时：设置合理的会话过期时间（如30分钟无交互自动关闭） 上下文压缩：定期摘要旧对话，而非简单截断 意图确认：低置信度意图主动询问用户 多模态状态：不仅跟踪文本，还跟踪用户的情绪、满意度 A/B测试：对话策略的变更需要A/B测试验证效果 结语 多轮对话管理是LLM助手从"问答工具"升级为"对话伙伴"的关键。状态管理、上下文窗口控制、话题跟踪和会话持久化的协同工作，让Agent能够维持连贯、智能的多轮对话。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

Agent记忆的三层架构 Agent需要像人类一样管理不同类型的记忆： 工作记忆：当前对话的上下文（短期） 情景记忆：过去交互的具体经历（中期） 语义记忆：从交互中提炼的知识（长期） 工作记忆：对话窗口管理 from collections import deque class WorkingMemory: def __init__(self, max_messages=20, max_tokens=4096): self.messages = deque(maxlen=max_messages) self.max_tokens = max_tokens self.token_counter = 0 def add(self, role, content): tokens = len(content) // 4 self.messages.append({"role": role, "content": content, "tokens": tokens}) self.token_counter += tokens # 超出token限制时移除最早的消息 while self.token_counter > self.max_tokens and len(self.messages) > 2: old = self.messages.popleft() self.token_counter -= old["tokens"] def get_context(self): return list(self.messages) def summarize_old_context(self): """当记忆过长时，摘要旧对话""" if len(self.messages) < 10: return old_messages = list(self.messages)[:len(self.messages)//2] summary = await self.summarize(old_messages) # 用摘要替换旧消息 for _ in range(len(old_messages)): self.messages.popleft() self.messages.appendleft({"role": "system", "content": f"之前的对话摘要：{summary}"}) 情景记忆：交互历史存储 class EpisodicMemory: def __init__(self, vector_store, llm): self.store = vector_store # 向量数据库 self.llm = llm async def save_interaction(self, user_id, user_message, assistant_response, metadata=None): """保存一次交互记录""" interaction = { "user_id": user_id, "user_message": user_message, "assistant_response": assistant_response, "timestamp": datetime.now().isoformat(), "metadata": metadata or {}, } # 生成摘要用于检索 summary = await self.llm.generate( f"用一句话概括这次交互：\n用户：{user_message}\n助手：{assistant_response}" ) # 向量化并存储 embedding = await self.llm.embed(summary) await self.store.add( id=generate_uuid(), embedding=embedding, document=json.dumps(interaction, ensure_ascii=False), metadata={"user_id": user_id, "summary": summary} ) async def recall(self, user_id, query, top_k=5): """检索相关的历史交互""" query_embedding = await self.llm.embed(query) results = await self.store.search( query_embedding, filter={"user_id": user_id}, top_k=top_k ) return [json.loads(r.document) for r in results] 语义记忆：知识图谱 class SemanticMemory: def __init__(self, graph_store): self.graph = graph_store # 图数据库（如Neo4j） async def learn(self, user_id, fact): """从交互中提取并存储知识""" # 使用LLM提取结构化知识 extracted = await self.extract_facts(fact) for fact_data in extracted: await self.graph.add_triple( subject=fact_data["subject"], predicate=fact_data["predicate"], object=fact_data["object"], metadata={"user_id": user_id, "source": "interaction"} ) async def query(self, user_id, entity): """查询关于某实体的知识""" triples = await self.graph.query( "MATCH (s)-[p]->(o) WHERE s.name = $entity RETURN s, p, o", {"entity": entity} ) knowledge = [] for triple in triples: knowledge.append(f"{triple['s']} {triple['p']} {triple['o']}") return knowledge async def extract_facts(self, text): """从文本中提取三元组""" prompt = f"""从以下文本中提取知识三元组（主语-谓语-宾语）： 文本：{text} 输出JSON数组： [{{"subject": "...", "predicate": "...", "object": "..."}}]""" result = await self.llm.generate(prompt) return json.loads(result) 记忆整合 class AgentMemorySystem: def __init__(self, working, episodic, semantic): self.working = working # 工作记忆 self.episodic = episodic # 情景记忆 self.semantic = semantic # 语义记忆 async def build_context(self, user_id, current_message): """构建完整的记忆上下文""" context_parts = [] # 1. 工作记忆（当前对话） working_ctx = self.working.get_context() context_parts.append({"type": "working", "messages": working_ctx}) # 2. 情景记忆（相关历史交互） episodic_results = await self.episodic.recall( user_id, current_message, top_k=3 ) if episodic_results: context_parts.append({ "type": "episodic", "memories": [r["summary"] for r in episodic_results] }) # 3. 语义记忆（相关知识） entities = await self.extract_entities(current_message) for entity in entities: knowledge = await self.semantic.query(user_id, entity) if knowledge: context_parts.append({ "type": "semantic", "entity": entity, "facts": knowledge }) return self.format_context(context_parts) def format_context(self, parts): """格式化记忆上下文""" context = "" for part in parts: if part["type"] == "working": context += "## 当前对话\n" for msg in part["messages"]: context += f"{msg['role']}: {msg['content']}\n" elif part["type"] == "episodic": context += "\n## 相关历史\n" for mem in part["memories"]: context += f"- {mem}\n" elif part["type"] == "semantic": context += f"\n## 关于{part['entity']}的知识\n" for fact in part["facts"]: context += f"- {fact}\n" return context 遗忘机制 class ForgettingMechanism: def __init__(self, decay_rate=0.01): self.decay_rate = decay_rate async def decay(self, memory_store): """时间衰减：降低旧记忆的重要性""" now = datetime.now() memories = await memory_store.get_all() for mem in memories: age_days = (now - mem["timestamp"]).days importance = mem.get("importance", 1.0) importance *= (1 - self.decay_rate) ** age_days if importance < 0.1: await memory_store.delete(mem["id"]) else: await memory_store.update(mem["id"], importance=importance) async def consolidate(self, memory_store): """记忆整合：将频繁出现的情景记忆转为语义记忆""" # 找出高频出现的事实 memories = await memory_store.get_all(min_importance=0.5) # 聚类相似记忆 clusters = self.cluster_memories(memories) for cluster in clusters: if len(cluster) >= 3: # 出现3次以上的模式转为知识 summary = await self.summarize_cluster(cluster) await self.semantic_memory.learn(summary) # 降低原始记忆的重要性 for mem in cluster: await memory_store.update(mem["id"], importance=mem["importance"] * 0.3) 存储选择 记忆类型 推荐存储 特点 工作记忆 内存（Redis） 快速读写，无需持久化 情景记忆 向量数据库 语义检索，按相似度召回 语义记忆 图数据库 关系查询，知识推理 结语 Agent记忆系统是构建长期智能助手的核心基础设施。工作记忆处理当前对话，情景记忆保存交互历史，语义记忆积累结构化知识。三层记忆的协同工作加上遗忘和整合机制，让Agent像人类一样"记得住该记的，忘得掉该忘的"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 Open WebUI是2026年最流行的自托管AI对话界面，提供了类似ChatGPT的体验，但完全在你的控制之下。配合Ollama等本地推理引擎，可以构建一个功能完整、数据私有的AI对话平台。本文将详细介绍Open WebUI的部署和使用。 为什么选择Open WebUI 核心优势 完全私有：数据不离开你的服务器 功能丰富：多模型支持、RAG、多用户、插件系统 界面友好：ChatGPT级别的用户体验 开源免费：无API费用 高度可定制：主题、提示、模型均可自定义 部署方案 方案一：Docker单机部署 # 最简单的部署 docker run -d -p 3000:8080 \ --add-host=host.docker.internal:host-gateway \ -v open-webui:/app/backend/data \ --name open-webui \ --restart always \ ghcr.io/open-webui/open-webui:main 方案二：Docker Compose version: '3.8' services: ollama: image: ollama/ollama:latest ports: - "11434:11434" volumes: - ollama_data:/root/.ollama deploy: resources: reservations: devices: - driver: nvidia count: all capabilities: [gpu] open-webui: image: ghcr.io/open-webui/open-webui:main ports: - "3000:8080" environment: - OLLAMA_BASE_URL=http://ollama:11434 - WEBUI_AUTH=true - ENABLE_RAG=true - ENABLE_OCR=true volumes: - webui_data:/app/backend/data depends_on: - ollama restart: always # 可选：向量数据库 chromadb: image: chromadb/chroma:latest ports: - "8000:8000" volumes: - chroma_data:/chroma/chroma volumes: ollama_data: webui_data: chroma_data: 方案三：Kubernetes部署 apiVersion: apps/v1 kind: Deployment metadata: name: open-webui spec: replicas: 1 template: spec: containers: - name: open-webui image: ghcr.io/open-webui/open-webui:main ports: - containerPort: 8080 env: - name: OLLAMA_BASE_URL value: "http://ollama-service:11434" - name: WEBUI_AUTH value: "true" resources: requests: memory: "2Gi" cpu: "1000m" --- apiVersion: v1 kind: Service metadata: name: open-webui spec: type: LoadBalancer ports: - port: 80 targetPort: 8080 核心功能配置 多模型管理 # 在WebUI中配置模型 models: - name: "GLM-5" ollama_model: "glm-5:32b" description: "中文最强模型" - name: "DeepSeek-V4" ollama_model: "deepseek-v4:671b" description: "开源代码之王" - name: "Qwen3 7B" ollama_model: "qwen3:7b" description: "轻量快速模型" RAG配置 # RAG设置 rag_config = { "embedding_model": "bge-large-zh", "chunk_size": 500, "chunk_overlap": 50, "top_k": 5, "reranker": "bge-reranker-v2", "vector_db": "chromadb" } 在WebUI中： ...

引言 LlamaIndex是专注于"将私有数据连接到LLM"的框架。2026年的LlamaIndex已经从简单的RAG工具发展为一个完整的数据驱动LLM应用平台。本文将全面介绍LlamaIndex 2026的使用。 核心概念 数据连接器 from llama_index.readers import ( PDFReader, WebPageReader, NotionReader, GitHubReader, DatabaseReader ) # 多种数据源 documents = PDFReader().load_data("report.pdf") web_docs = WebPageReader().load_data(["https://example.com"]) db_docs = DatabaseReader(uri="postgresql://...").load_data("SELECT * FROM articles") 索引 from llama_index.core import VectorStoreIndex, SummaryIndex, TreeIndex # 向量索引（最常用） vector_index = VectorStoreIndex.from_documents(documents) # 摘要索引（适合长文档） summary_index = SummaryIndex.from_documents(documents) # 树索引（适合层次化数据） tree_index = TreeIndex.from_documents(documents) # 关键词索引 from llama_index.core import KeywordTableIndex keyword_index = KeywordTableIndex.from_documents(documents) 查询引擎 # 基本查询 query_engine = vector_index.as_query_engine(similarity_top_k=5) response = query_engine.query("什么是AI?") # 流式查询 streaming_engine = vector_index.as_query_engine(streaming=True) response = streaming_engine.query("什么是AI?") for text in response.response_gen: print(text, end="") # 子问题查询 from llama_index.core.tools import QueryEngineTool from llama_index.core.query_engine import SubQuestionQueryEngine tools = [ QueryEngineTool.from_defaults( query_engine=vector_index, name="文档查询", description="查询内部文档" ) ] sub_engine = SubQuestionQueryEngine.from_defaults(query_engine_tools=tools) response = sub_engine.query("比较文档A和文档B的观点") 2026年新特性 1. LlamaCloud from llama_index.cloud import LlamaCloud # 云端索引管理 cloud = LlamaCloud(api_key="...") index = cloud.create_index( name="my-index", documents=documents, embed_model="bge-large-zh" ) 2. Agent支持 from llama_index.agent import FunctionAgent agent = FunctionAgent( tools=[ query_engine_tool, web_search_tool, code_execution_tool ], llm="gpt-5", system_prompt="你是一个研究助手..." ) response = agent.chat("分析最新的AI趋势并生成报告") 3. 工作流 from llama_index.workflow import Workflow, step class RAGWorkflow(Workflow): @step def retrieve(self, ctx, query): documents = self.retriever.retrieve(query) ctx.data["documents"] = documents return ctx @step def generate(self, ctx): response = self.llm.complete( prompt=ctx.data["query"], context=ctx.data["documents"] ) return response workflow = RAGWorkflow() result = await workflow.run("什么是AI?") 4. 多模态 from llama_index.multi_modal import MultiModalIndex # 多模态索引 mm_index = MultiModalIndex.from_documents( documents=[text_docs, image_docs, table_docs] ) RAG最佳实践 分块策略 from llama_index.core.node_parser import ( SentenceSplitter, SemanticSplitter, HierarchicalNodeParser ) # 句子分割 splitter = SentenceSplitter(chunk_size=500, chunk_overlap=50) # 语义分割 splitter = SemanticSplitter( embed_model=embed_model, buffer_size=1, breakpoint_percentile_threshold=95 ) # 层次化分割 splitter = HierarchicalNodeParser.from_defaults( chunk_sizes=[2048, 512, 128] # 三级层次 ) 检索优化 from llama_index.core.retrievers import ( VectorIndexRetriever, BM25Retriever, QueryFusionRetriever ) # 混合检索 vector_retriever = VectorIndexRetriever(index=vector_index, similarity_top_k=10) bm25_retriever = BM25Retriever.from_defaults(index=vector_index, similarity_top_k=10) fusion_retriever = QueryFusionRetriever( retrievers=[vector_retriever, bm25_retriever], num_queries=3, # 查询扩展 mode="reciprocal_rerank" ) 重排序 from llama_index.core.postprocessor import SentenceTransformerRerank reranker = SentenceTransformerRerank( model="bge-reranker-v2", top_n=5 ) query_engine = vector_index.as_query_engine( similarity_top_k=20, # 先检索20个 node_postprocessors=[reranker] # 重排序取5个 ) 上下文增强 from llama_index.core.indices.query.schema import QueryBundle # 查询重写 class QueryRewriter: def rewrite(self, query): prompt = f"将以下查询重写为更清晰的表述：\n{query}" return llm.complete(prompt).text # 在查询前重写 rewritten = QueryRewriter().rewrite("AI怎么样") response = query_engine.query(QueryBundle(rewritten)) 评估 from llama_index.core.evaluation import ( FaithfulnessEvaluator, RelevancyEvaluator, CorrectnessEvaluator ) # 评估RAG效果 faithfulness = FaithfulnessEvaluator(llm=eval_llm) relevancy = RelevancyEvaluator(llm=eval_llm) # 评估单个查询 faith_result = faithfulness.evaluate_response( query=query, response=response ) # faith_result.passing: True/False 部署 API服务 from llama_index.core.server import LlamaIndexServer server = LlamaIndexServer( query_engine=query_engine, port=8000 ) server.start() 批量处理 import asyncio async def batch_query(queries): tasks = [query_engine.aquery(q) for q in queries] results = await asyncio.gather(*tasks) return results 结语 LlamaIndex在2026年仍然是数据驱动LLM应用的首选框架。它的数据连接器丰富、索引类型多样、查询引擎灵活，特别适合需要处理大量私有数据的场景。 ...

不是非此即彼 RAG和微调不是互斥的选择，而是互补的技术。很多场景下，两者结合使用效果最佳。关键在于理解各自的优势和局限，根据具体需求做出合理选择。 决策矩阵 ┌──────────────────────────────────┐ │ 知识更新频率 │ │ 低 中 高 │ ┌──────────┼──────────┬──────────┬────────────┤ 知识 │ 私有 │ 微调 │ RAG │ RAG │ 特有性 │ 公开 │ 微调 │ RAG │ Prompt │ 量 │ 大量 │ RAG │ RAG │ RAG │ │ 少量 │ 微调 │ 微调 │ Prompt │ └──────────┴──────────┴──────────┴────────────┘ 何时用RAG 适合场景： 知识库频繁更新（如产品文档、新闻） 需要精确引用来源 大量私有文档（数千篇以上） 需要多跳推理 # RAG的典型应用：企业知识库问答 class EnterpriseQABot: def __init__(self): self.retriever = VectorRetriever(documents=company_docs) self.llm = LLM(model="qwen3-32b") async def answer(self, question): docs = await self.retriever.search(question, top_k=5) context = "\n".join(d.content for d in docs) prompt = f"""基于以下参考资料回答问题。 参考资料： {context} 问题：{question} 要求标注引用来源。""" return await self.llm.generate(prompt) RAG优势： ...

引言 Haystack是deepset开发的企业级NLP框架，在RAG领域有着深厚积累。2026年的Haystack已经发展成为一个完整的RAG解决方案框架。本文将分享Haystack在RAG实践中的经验。 Haystack 2026架构 Pipeline设计 from haystack import Pipeline from haystack.components.embedders import OllamaEmbedder from haystack.components.retrievers import ChromaRetriever from haystack.components.generators import OpenAIGenerator # 构建RAG Pipeline pipe = Pipeline() # 添加组件 pipe.add_component("embedder", OllamaEmbedder(model="bge-large-zh")) pipe.add_component("retriever", ChromaRetriever(top_k=5)) pipe.add_component("generator", OpenAIGenerator(model="gpt-5")) # 连接组件 pipe.connect("embedder.embedding", "retriever.query_embedding") pipe.connect("retriever.documents", "generator.documents") 文档处理 from haystack.components.converters import PDFToDocument, MarkdownToDocument from haystack.components.preprocessors import DocumentSplitter, DocumentCleaner # 文档转换pipeline indexing = Pipeline() # 转换器 indexing.add_component("pdf_converter", PDFToDocument()) indexing.add_component("md_converter", MarkdownToDocument()) # 预处理 indexing.add_component("cleaner", DocumentCleaner()) indexing.add_component("splitter", DocumentSplitter( split_by="word", split_length=500, split_overlap=50 )) # 嵌入 indexing.add_component("embedder", OllamaEmbedder(model="bge-large-zh")) indexing.add_component("writer", ChromaDocumentWriter()) # 连接 indexing.connect("pdf_converter.documents", "cleaner.documents") indexing.connect("cleaner.documents", "splitter.documents") indexing.connect("splitter.documents", "embedder.documents") indexing.connect("embedder.documents", "writer.documents") RAG优化实践 实践一：混合检索 from haystack.components.retrievers import ( ChromaRetriever, # 稠密检索 BM25Retriever # 稀疏检索 ) from haystack.components.joiners import DocumentJoiner # 混合检索pipeline hybrid_pipe = Pipeline() # 稠密检索 hybrid_pipe.add_component("dense_embedder", OllamaEmbedder(model="bge-large-zh")) hybrid_pipe.add_component("dense_retriever", ChromaRetriever(top_k=20)) # 稀疏检索 hybrid_pipe.add_component("sparse_retriever", BM25Retriever(top_k=20)) # 融合 hybrid_pipe.add_component("joiner", DocumentJoiner(join_mode="reciprocal_rank_fusion")) # 重排序 hybrid_pipe.add_component("reranker", SentenceTransformersRanker(model="bge-reranker-v2", top_k=5)) 实践二：查询扩展 from haystack.components.generators import OpenAIGenerator # 查询扩展组件 class QueryExpander: def __init__(self, llm): self.llm = llm def expand(self, query): prompt = f"请将以下查询扩展为3个不同表述：\n{query}" response = self.llm.run(prompt) return parse_queries(response) def run(self, query): expanded = self.expand(query) return {"queries": expanded} # 在pipeline中使用 pipe.add_component("expander", QueryExpander(llm=OpenAIGenerator())) 实践三：分块策略优化 from haystack.components.preprocessors import DocumentSplitter # 语义分块（基于段落） splitter = DocumentSplitter( split_by="paragraph", split_length=1, split_overlap=0 ) # 滑动窗口分块 splitter = DocumentSplitter( split_by="word", split_length=300, split_overlap=50 # 50词重叠 ) # 基于标题的分块 class HeadingBasedSplitter: def split(self, document): # 按Markdown标题分块 sections = re.split(r'^#+\s', document.content, flags=re.MULTILINE) return [Document(content=s.strip()) for s in sections if s.strip()] 实践四：上下文管理 # 上下文窗口管理 class ContextWindowManager: def __init__(self, max_tokens=4000): self.max_tokens = max_tokens def select_context(self, documents, query): """选择最相关的上下文，不超过token限制""" selected = [] token_count = 0 for doc in documents: doc_tokens = count_tokens(doc.content) if token_count + doc_tokens > self.max_tokens: # 截断最后一个文档 remaining = self.max_tokens - token_count if remaining > 100: # 至少100 token才包含 doc.content = doc.content[:remaining] selected.append(doc) break selected.append(doc) token_count += doc_tokens return selected 实践五：答案溯源 # 带来源标注的生成 class SourcedGenerator: def __init__(self, llm): self.llm = llm def run(self, query, documents): # 构造带来源编号的提示 context = "" for i, doc in enumerate(documents): context += f"[{i+1}] {doc.content}\n\n" prompt = f""" 基于以下参考信息回答问题。在回答中标注信息来源。 参考信息： {context} 问题：{query} 回答格式：答案内容[来源编号] """ response = self.llm.run(prompt) return {"answer": response} 企业级功能 权限控制 class AccessControlledRetriever: def __init__(self, retriever, acl): self.retriever = retriever self.acl = acl # 访问控制列表 def run(self, query, user_id): # 检索 documents = self.retriever.run(query) # 过滤：只返回用户有权限的文档 accessible = [ doc for doc in documents if self.acl.has_access(user_id, doc.metadata.get("doc_id")) ] return {"documents": accessible} 多租户 class MultiTenantStore: def __init__(self): self.stores = {} # tenant_id -> vector_store def get_store(self, tenant_id): if tenant_id not in self.stores: self.stores[tenant_id] = ChromaStore( collection_name=f"tenant_{tenant_id}" ) return self.stores[tenant_id] 缓存 from haystack.components.cachers import CacheChecker pipe.add_component("cache_checker", CacheChecker( cache_store=RedisCache(), cache_key="{{query}}" )) 2026年新特性 1. 多模态RAG from haystack.components.embedders import CLIPEmbedder # 图文混合RAG pipe.add_component("image_embedder", CLIPEmbedder()) pipe.add_component("text_embedder", OllamaEmbedder(model="bge-large-zh")) 2. 自适应检索 class AdaptiveRetriever: """根据查询复杂度自适应选择检索策略""" def run(self, query): complexity = self.assess_complexity(query) if complexity == "simple": return self.simple_retrieve(query) elif complexity == "medium": return self.hybrid_retrieve(query) else: return self.multi_hop_retrieve(query) 3. 评估集成 from haystack.components.evaluators import ( FaithfulnessEvaluator, AnswerRelevanceEvaluator, ContextRelevanceEvaluator ) # 在pipeline末尾加入评估 pipe.add_component("faithfulness", FaithfulnessEvaluator()) pipe.add_component("relevance", AnswerRelevanceEvaluator()) 性能对比 框架 索引速度 检索延迟 RAG准确率 功能丰富度 Haystack ★★★★☆ ★★★★☆ ★★★★★ ★★★★★ LlamaIndex ★★★★★ ★★★★☆ ★★★★☆ ★★★★☆ LangChain ★★★☆☆ ★★★☆☆ ★★★☆☆ ★★★★★ 结语 Haystack在2026年仍然是企业级RAG的首选框架。其Pipeline架构清晰、组件丰富、可扩展性强，特别适合需要精细控制RAG流程的企业应用。 ...