企业官网建设流程全解析

最近在尝试将大语言模型应用到实际业务中，发现很多开发者虽然对单个工具使用熟练，但缺乏完整的全栈工程化思维。本文基于企业级项目经验，整合从提示词工程到智能体开发的完整技术栈，包含可落地的代码示例和避坑指南，无论你是想转型AI开发的全栈工程师，还是希望提升工程化能力的研究人员，都能获得体系化的实战方案。

1. LLM全栈工程师的技术体系解析

1.1 什么是LLM全栈开发

LLM全栈开发是指能够独立完成基于大语言模型的端到端应用开发，涵盖数据准备、模型调优、应用集成和部署运维全流程。与传统全栈开发相比，LLM全栈工程师需要掌握提示词工程、向量数据库、智能体框架等AI特有技术栈。

核心能力矩阵包括：

• 基础层：Python编程、API调用、数据处理
• 工具层：Prompt工程、RAG系统、Fine-tuning
• 框架层：Agent开发、Workflow编排
• 工程层：部署优化、监控运维

1.2 技术选型与生态分析

当前主流技术生态呈现多元化发展，企业级项目需要根据具体场景选择合适的技术组合：

• 开发工具：Cursor作为AI原生IDE，大幅提升编码效率；VS Code with Copilot为传统选择
• RAG框架：LangChain、LlamaIndex为成熟方案，新兴的MCP协议提供更灵活的扩展能力
• Agent框架：Coze、AutoGPT、ChatDev各有侧重，Coze在可视化工作流方面优势明显
• 部署平台：AWS Bedrock、Azure AI Studio、自建GPU集群等选择需考虑成本与控制权平衡

2. 环境准备与基础工具配置

2.1 开发环境搭建

推荐使用Python 3.9+作为基础环境，避免版本兼容性问题：

# 创建虚拟环境 python -m venv llm-env source llm-env/bin/activate # Linux/Mac # llm-env\Scripts\activate # Windows # 安装核心依赖 pip install openai langchain chromadb fastapi uvicorn

2.2 Cursor IDE配置优化

Cursor作为AI优先的代码编辑器，需要正确配置才能发挥最大效能：

// settings.json 配置示例 { "cursor.codeCompletionModel": "gpt-4", "cursor.inlineChatModel": "gpt-4", "editor.fontSize": 14, "python.analysis.autoImportCompletions": true }

中文界面设置技巧：

1. 通过Command Palette (Ctrl+Shift+P) 搜索"Configure Display Language"
2. 安装中文语言包后重启生效
3. 注意保持英文术语的专业性，避免翻译歧义

2.3 API密钥安全管理

所有LLM应用都需要妥善管理API密钥，推荐使用环境变量方式：

# config.py - 配置文件模板 import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY = os.getenv('OPENAI_API_KEY') ANTHROPIC_API_KEY = os.getenv('ANTHROPIC_API_KEY') SERPER_API_KEY = os.getenv('SERPER_API_KEY') # .env文件示例（切勿提交到版本库） # OPENAI_API_KEY=sk-your-key-here # ANTHROPIC_API_KEY=your-antropic-key # SERPER_API_KEY=your-serper-key

3. Prompt Engineering实战精要

3.1 结构化Prompt设计原则

有效的Prompt需要遵循明确的结构化原则，以下是一个企业级模板：

def create_structured_prompt(task_type, context, requirements): prompt_template = """ # 角色定义 你是一名专业的{role}，具有{expertise}领域经验。 # 任务目标 需要完成以下任务：{task_description} # 上下文信息 相关背景：{context} # 输出要求 - 格式：{format_requirements} - 长度：{length_constraints} - 风格：{style_guidelines} # 约束条件 {constraints} 请开始执行任务： """ return prompt_template.format( role=task_type, expertise=requirements.get('expertise', '相关'), task_description=requirements['description'], context=context, format_requirements=requirements.get('format', 'Markdown'), length_constraints=requirements.get('length', '适中'), style_guidelines=requirements.get('style', '专业'), constraints=requirements.get('constraints', '无特殊约束') ) # 使用示例 task_requirements = { 'description': '分析当前季度销售数据，识别关键趋势', 'expertise': '商业分析', 'format': '表格+文字分析', 'length': '500-800字', 'style': '数据驱动、见解深刻' } prompt = create_structured_prompt( task_type='商业分析师', context='Q3销售数据包含产品A、B、C的 regional销售情况', requirements=task_requirements )

3.2 常见Prompt模式与反模式

在实际项目中观察到的有效模式：

有效模式示例：

• 思维链（Chain-of-Thought）："让我们一步步推理这个问题..."
• 角色扮演（Role-Playing）："假设你是资深软件架构师..."
• 示例引导（Few-Shot）："参考以下示例的格式和深度..."

需要避免的反模式：

• 模糊指令："帮我分析一下数据" → 应具体说明分析维度和输出格式
• 矛盾要求："简洁但详细地说明" → 需要明确优先级
• 假设知识："用大家都知道的方法" → 应明确具体技术栈

3.3 Prompt版本管理与评估

企业级项目需要建立Prompt的版本管理机制：

class PromptManager: def __init__(self): self.prompt_versions = {} self.metrics = {} def save_prompt(self, name, prompt, version, metadata=None): if name not in self.prompt_versions: self.prompt_versions[name] = {} self.prompt_versions[name][version] = { 'prompt': prompt, 'metadata': metadata or {}, 'timestamp': datetime.now() } def evaluate_prompt(self, name, version, test_cases): """评估Prompt在不同测试用例上的表现""" results = [] for case in test_cases: # 执行测试并记录指标 result = self._execute_test(case) results.append({ 'case': case, 'result': result, 'score': self._calculate_score(result) }) return results

4. RAG系统构建与优化

4.1 企业级RAG架构设计

完整的RAG系统包含数据预处理、向量化、检索和生成四个核心模块：

# rag_system.py - 基础RAG系统实现 from langchain.document_loaders import PyPDFLoader, WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.chains import RetrievalQA from langchain.llms import OpenAI class EnterpriseRAGSystem: def __init__(self, embedding_model="text-embedding-ada-002"): self.embeddings = OpenAIEmbeddings(model=embedding_model) self.vector_store = None self.qa_chain = None def load_documents(self, document_paths): """加载多种格式的文档""" documents = [] for path in document_paths: if path.endswith('.pdf'): loader = PyPDFLoader(path) elif path.startswith('http'): loader = WebBaseLoader(path) else: continue documents.extend(loader.load()) # 文本分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) return text_splitter.split_documents(documents) def build_vector_store(self, documents, persist_directory="./chroma_db"): """构建向量数据库""" self.vector_store = Chroma.from_documents( documents=documents, embedding=self.embeddings, persist_directory=persist_directory ) return self.vector_store def create_qa_chain(self, llm_model="gpt-3.5-turbo"): """创建问答链""" llm = OpenAI(model_name=llm_model, temperature=0) self.qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=self.vector_store.as_retriever(), return_source_documents=True ) return self.qa_chain def query(self, question): """执行查询""" if not self.qa_chain: raise ValueError("请先构建QA链") return self.qa_chain({"query": question}) # 使用示例 rag_system = EnterpriseRAGSystem() documents = rag_system.load_documents(["document1.pdf", "https://example.com/doc"]) rag_system.build_vector_store(documents) rag_system.create_qa_chain() result = rag_system.query("什么是机器学习？") print(result['result'])

4.2 检索质量优化策略

提升RAG系统效果的关键技术点：

分块策略优化：

def adaptive_chunking(text, content_type): """根据内容类型自适应分块""" if content_type == "technical_doc": chunk_size = 800 chunk_overlap = 100 elif content_type == "legal_doc": chunk_size = 500 # 法律文档需要更精细的分块 chunk_overlap = 50 else: chunk_size = 1000 chunk_overlap = 200 splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", ". ", "! ", "? ", " ", ""] ) return splitter.split_text(text)

混合检索增强：

class HybridRetriever: def __init__(self, vector_store, keyword_retriever): self.vector_store = vector_store self.keyword_retriever = keyword_retriever def retrieve(self, query, top_k=5, alpha=0.7): # 向量检索 vector_results = self.vector_store.similarity_search(query, k=top_k) # 关键词检索 keyword_results = self.keyword_retriever.search(query, limit=top_k) # 结果融合 combined_results = self._rerank_results( vector_results, keyword_results, alpha ) return combined_results[:top_k]

4.3 RAG系统评估框架

建立科学的评估体系确保系统质量：

class RAGEvaluator: def __init__(self, test_dataset): self.test_dataset = test_dataset self.metrics = { 'accuracy': [], 'relevance': [], 'completeness': [] } def evaluate_retrieval(self, retriever, queries): """评估检索模块效果""" results = [] for query, expected_docs in queries: retrieved_docs = retriever.retrieve(query) precision = self._calculate_precision(retrieved_docs, expected_docs) recall = self._calculate_recall(retrieved_docs, expected_docs) results.append({'query': query, 'precision': precision, 'recall': recall}) return results def evaluate_generation(self, qa_chain, qa_pairs): """评估生成质量""" # 实现BLEU、ROUGE等指标计算 pass

5. Agentic AI开发实战

5.1 智能体架构设计模式

现代AI智能体通常采用分层架构：

# agent_framework.py - 基础智能体框架 from abc import ABC, abstractmethod from typing import List, Dict, Any import json class Tool: def __init__(self, name, description, parameters): self.name = name self.description = description self.parameters = parameters @abstractmethod def execute(self, **kwargs): pass class CalculatorTool(Tool): def __init__(self): super().__init__( name="calculator", description="执行数学计算", parameters={ "expression": {"type": "string", "description": "数学表达式"} } ) def execute(self, expression): try: result = eval(expression) return f"计算结果: {result}" except Exception as e: return f"计算错误: {str(e)}" class Agent: def __init__(self, name, tools: List[Tool], llm): self.name = name self.tools = {tool.name: tool for tool in tools} self.llm = llm self.memory = [] def process_query(self, query): """处理用户查询的完整流程""" # 1. 意图识别 intent = self._classify_intent(query) # 2. 工具选择 selected_tools = self._select_tools(intent) # 3. 规划执行 plan = self._create_execution_plan(query, selected_tools) # 4. 执行监控 results = self._execute_plan(plan) # 5. 结果整合 final_response = self._synthesize_results(results) self.memory.append({ 'query': query, 'plan': plan, 'results': results, 'response': final_response }) return final_response def _create_execution_plan(self, query, tools): """创建执行计划""" plan_prompt = f""" 基于查询"{query}"和可用工具{list(tools.keys())}，制定执行计划。 输出JSON格式： {{ "steps": [ {{ "tool": "工具名", "parameters": {{参数}}, "purpose": "步骤目的" }} ] }} """ response = self.llm.generate(plan_prompt) return json.loads(response)

5.2 多智能体协作系统

复杂任务需要多个智能体协同工作：

class MultiAgentSystem: def __init__(self): self.agents = {} self.coordinator = None def register_agent(self, agent): self.agents[agent.name] = agent def coordinate_task(self, task_description): """协调多个智能体完成任务""" # 任务分解 subtasks = self._decompose_task(task_description) # 智能体分配 assignments = self._assign_subtasks(subtasks) # 执行协调 results = {} for agent_name, tasks in assignments.items(): agent = self.agents[agent_name] agent_results = [] for task in tasks: result = agent.process_query(task) agent_results.append(result) results[agent_name] = agent_results # 结果整合 final_result = self._integrate_results(results) return final_result class SpecialistAgent(Agent): def __init__(self, name, specialty, tools, llm): super().__init__(name, tools, llm) self.specialty = specialty def _classify_intent(self, query): """专业领域意图识别""" # 实现领域特定的意图分类逻辑 pass

5.3 Coze工作流实战

Coze平台提供了可视化的智能体编排能力：

# coze_integration.py - Coze API集成示例 import requests import json class CozeWorkflowClient: def __init__(self, api_key, bot_id): self.api_key = api_key self.bot_id = bot_id self.base_url = "https://api.coze.com/v1" def execute_workflow(self, workflow_id, inputs): """执行Coze工作流""" headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } payload = { "bot_id": self.bot_id, "workflow_id": workflow_id, "inputs": inputs } response = requests.post( f"{self.base_url}/workflows/execute", headers=headers, json=payload ) if response.status_code == 200: return response.json() else: raise Exception(f"工作流执行失败: {response.text}") def create_conversation(self, user_message): """创建对话会话""" # Coze对话API集成 pass # 使用示例：客户服务自动化工作流 coze_client = CozeWorkflowClient("your-api-key", "your-bot-id") workflow_inputs = { "customer_query": "产品价格和优惠信息", "customer_tier": "premium", "language": "zh-CN" } result = coze_client.execute_workflow("customer_service_flow", workflow_inputs) print(result['response'])

6. MCP协议与系统集成

6.1 MCP协议核心概念

模型上下文协议（Model Context Protocol）为LLM应用提供了标准化的扩展机制：

# mcp_server.py - 基础MCP服务器实现 import asyncio from mcp import MCPServer, ClientSession from mcp.server.models import InitializationOptions class CustomMCPServer(MCPServer): def __init__(self): super().__init__("custom-tools-server") async def initialize_session(self, session: ClientSession) -> None: """初始化会话时注册可用工具""" await session.list_tools() # 注册自定义工具 tools = [ { "name": "search_database", "description": "在内部数据库中搜索信息", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer", "default": 10} } } } ] await session.register_tools(tools) async def handle_tool_call(self, session: ClientSession, tool_name: str, arguments: dict): """处理工具调用请求""" if tool_name == "search_database": return await self._search_database(arguments['query'], arguments.get('limit', 10)) else: raise ValueError(f"未知工具: {tool_name}") async def _search_database(self, query: str, limit: int): """模拟数据库搜索""" # 实际项目中连接真实数据库 results = [ {"id": 1, "title": "相关文档1", "content": "匹配的内容片段"}, {"id": 2, "title": "相关文档2", "content": "另一个匹配片段"} ] return results[:limit] # 启动服务器 async def main(): server = CustomMCPServer() await server.run() if __name__ == "__main__": asyncio.run(main())

6.2 MCP客户端集成

在应用中集成MCP客户端：

# mcp_client.py - MCP客户端实现 from mcp.client import ClientSession from mcp.client.stdio import stdio_client async def use_mcp_tools(): async with stdio_client("path/to/mcp/server") as (read, write): async with ClientSession(read, write) as session: # 初始化会话 init_result = await session.initialize() # 列出可用工具 tools = await session.list_tools() print("可用工具:", tools) # 调用工具 result = await session.call_tool( "search_database", {"query": "机器学习", "limit": 5} ) print("搜索结果:", result)

6.3 企业级MCP应用场景

MCP协议在企业环境中的典型应用：

统一工具平台：

class EnterpriseMCPServer: def __init__(self): self.tool_registry = {} def register_department_tools(self, department, tools): """按部门注册工具集""" self.tool_registry[department] = tools async def handle_department_request(self, department, tool_call): """处理部门特定的工具调用""" if department not in self.tool_registry: raise ValueError(f"未知部门: {department}") tool = self.tool_registry[department].get(tool_call.name) if not tool: raise ValueError(f"部门 {department} 中未找到工具 {tool_call.name}") return await tool.execute(tool_call.arguments)

7. 全栈项目实战：智能知识管理系统

7.1 系统架构设计

构建一个完整的企业知识管理系统，整合前面介绍的所有技术：

系统架构层次： 1. 数据层：文档存储 + 向量数据库 2. 服务层：RAG引擎 + Agent系统 3. 应用层：Web界面 + API接口 4. 集成层：MCP工具 + 第三方系统

7.2 后端核心实现

# app/main.py - FastAPI后端主程序 from fastapi import FastAPI, HTTPException from pydantic import BaseModel from rag_system import EnterpriseRAGSystem from agent_framework import MultiAgentSystem app = FastAPI(title="智能知识管理系统") # 初始化核心组件 rag_system = EnterpriseRAGSystem() agent_system = MultiAgentSystem() class QueryRequest(BaseModel): question: str context: dict = None class DocumentUploadRequest(BaseModel): documents: list metadata: dict = None @app.post("/api/query") async def query_knowledge_base(request: QueryRequest): """查询知识库""" try: result = rag_system.query(request.question) return { "answer": result['result'], "sources": [doc.metadata for doc in result['source_documents']], "confidence": 0.95 # 模拟置信度计算 } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/upload") async def upload_documents(request: DocumentUploadRequest): """上传文档到知识库""" try: documents = rag_system.load_documents(request.documents) rag_system.build_vector_store(documents) return {"message": f"成功上传 {len(documents)} 个文档片段"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/agent/task") async def submit_agent_task(task_description: str): """提交智能体任务""" result = agent_system.coordinate_task(task_description) return {"result": result} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

7.3 前端界面集成

<!-- templates/index.html - 知识管理界面 --> <!DOCTYPE html> <html> <head> <title>智能知识管理系统</title> <script src="https://cdn.jsdelivr.net/npm/axios/dist/axios.min.js"></script> </head> <body> <div class="container"> <h1>企业知识智能助手</h1> <!-- 知识查询模块 --> <div class="query-section"> <textarea id="questionInput" placeholder="输入你的问题..."></textarea> <button onclick="submitQuery()">智能搜索</button> <div id="results"></div> </div> <!-- 文档上传模块 --> <div class="upload-section"> <input type="file" id="documentUpload" multiple> <button onclick="uploadDocuments()">上传文档</button> </div> </div> <script> async function submitQuery() { const question = document.getElementById('questionInput').value; const response = await axios.post('/api/query', { question: question }); document.getElementById('results').innerHTML = ` <h3>答案：</h3> <p>${response.data.answer}</p> <h4>参考来源：</h4> <ul> ${response.data.sources.map(source => `<li>${source.title || source.source}</li>` ).join('')} </ul> `; } async function uploadDocuments() { const files = document.getElementById('documentUpload').files; const formData = new FormData(); for (let file of files) { formData.append('documents', file); } await axios.post('/api/upload', formData, { headers: {'Content-Type': 'multipart/form-data'} }); alert('文档上传成功！'); } </script> </body> </html>

8. 性能优化与生产部署

8.1 向量检索性能优化

大规模知识库需要优化检索性能：

# optimization.py - 性能优化策略 import time from functools import lru_cache from typing import List class OptimizedRetriever: def __init__(self, vector_store, cache_size=1000): self.vector_store = vector_store self.cache = {} self.cache_size = cache_size @lru_cache(maxsize=1000) def cached_retrieval(self, query: str, k: int) -> List: """带缓存的检索""" cache_key = f"{query}_{k}" if cache_key in self.cache: return self.cache[cache_key] results = self.vector_store.similarity_search(query, k=k) self.cache[cache_key] = results # 维护缓存大小 if len(self.cache) > self.cache_size: oldest_key = next(iter(self.cache)) del self.cache[oldest_key] return results def batch_retrieve(self, queries: List[str], k: int = 5): """批量检索优化""" # 实现批量处理的逻辑 pass # 索引优化策略 def create_optimized_index(vector_store, index_type="HNSW"): """创建优化索引""" if index_type == "HNSW": # 分层可导航小世界图索引 index_config = { "m": 16, # 构建时每个节点的连接数 "ef_construction": 200, # 索引构建参数 "ef_search": 100 # 搜索时动态候选集大小 } elif index_type == "IVF": # 倒排文件索引 index_config = { "nlist": 100, # 聚类中心数量 "nprobe": 10 # 搜索时探查的聚类数量 } return vector_store.create_index(index_config)

8.2 模型推理优化

减少API调用成本和延迟：

class ModelOptimizer: def __init__(self, llm, cache_enabled=True): self.llm = llm self.cache_enabled = cache_enabled self.response_cache = {} def optimized_generate(self, prompt, max_tokens=500, temperature=0.7): """优化生成过程""" if self.cache_enabled: cache_key = hash(prompt) # 简化示例 if cache_key in self.response_cache: return self.response_cache[cache_key] # 实现提示词压缩、结果缓存等优化 response = self.llm.generate( prompt, max_tokens=max_tokens, temperature=temperature ) if self.cache_enabled: self.response_cache[cache_key] = response return response def batch_process(self, prompts): """批量处理提示词""" # 实现批量API调用优化 pass

8.3 生产环境部署配置

Docker化部署方案：

# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ gcc \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml version: '3.8' services: knowledge-app: build: . ports: - "8000:8000" environment: - OPENAI_API_KEY=${OPENAI_API_KEY} - DATABASE_URL=postgresql://user:pass@db:5432/knowledge depends_on: - db - redis db: image: postgres:13 environment: POSTGRES_DB: knowledge POSTGRES_USER: user POSTGRES_PASSWORD: pass volumes: - postgres_data:/var/lib/postgresql/data redis: image: redis:6-alpine volumes: - redis_data:/data volumes: postgres_data: redis_data:

9. 安全与权限管理

9.1 API安全防护

# security.py - 安全防护措施 from fastapi import Security, HTTPException from fastapi.security import APIKeyHeader from starlette.status import HTTP_403_FORBIDDEN api_key_header = APIKeyHeader(name="X-API-Key") class SecurityManager: def __init__(self, valid_api_keys): self.valid_api_keys = set(valid_api_keys) async def validate_api_key(self, api_key: str = Security(api_key_header)): """验证API密钥""" if api_key not in self.valid_api_keys: raise HTTPException( status_code=HTTP_403_FORBIDDEN, detail="无效的API密钥" ) return api_key def sanitize_input(self, user_input: str) -> str: """输入清洗与防护""" # 移除潜在危险字符 dangerous_patterns = [ "<script>", "javascript:", "onload=", "onerror=", "eval(", "document.cookie" ] sanitized = user_input for pattern in dangerous_patterns: sanitized = sanitized.replace(pattern, "") return sanitized.strip()

9.2 数据权限控制

class DataPermissionManager: def __init__(self): self.user_permissions = {} def set_user_permissions(self, user_id, permissions): """设置用户权限""" self.user_permissions[user_id] = permissions def check_document_access(self, user_id, document_id): """检查文档访问权限""" user_perm = self.user_permissions.get(user_id, {}) document_perm = user_perm.get('documents', []) return document_id in document_perm def filter_sensitive_content(self, content, user_role): """根据用户角色过滤敏感内容""" if user_role == 'guest': # 对访客隐藏敏感信息 sensitive_keywords = ['机密', '内部', '薪资'] for keyword in sensitive_keywords: content = content.replace(keyword, '***') return content

10. 监控与运维体系

10.1 系统监控指标

# monitoring.py - 监控系统实现 import time import logging from dataclasses import dataclass from typing import Dict, List @dataclass class SystemMetrics: response_time: float error_rate: float cache_hit_rate: float active_connections: int class MonitoringSystem: def __init__(self): self.metrics_history: Dict[str, List] = { 'response_time': [], 'error_rate': [], 'throughput': [] } self.logger = logging.getLogger('monitoring') def record_metric(self, metric_name: str, value: float): """记录指标数据""" if metric_name not in self.metrics_history: self.metrics_history[metric_name] = [] self.metrics_history[metric_name].append({ 'timestamp': time.time(), 'value': value }) # 保持最近1000个数据点 if len(self.metrics_history[metric_name]) > 1000: self.metrics_history[metric_name].pop(0) def generate_health_report(self) -> Dict: """生成系统健康报告""" return { 'status': self._calculate_system_status(), 'metrics': { 'avg_response_time': self._calculate_average('response_time'), 'error_rate': self._calculate_error_rate(), 'system_uptime': self._get_uptime() }, 'alerts': self._check_alerts() }

10.2 日志管理策略

# logging_config.py - 日志配置 import logging import json from datetime import datetime def setup_logging(): """配置结构化日志""" logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('app.log'), logging.StreamHandler() ] ) class StructuredLogger: def __init__(self, name): self.logger = logging.getLogger(name) def log_query(self, query: str, response: str, metadata: dict): """记录查询日志""" log_entry = { 'timestamp': datetime.now().isoformat(), 'query': query, 'response_preview': response[:100] + '...' if len(response) > 100 else response, 'metadata': metadata, 'type': 'query' } self.logger.info(json.dumps(log_entry, ensure_ascii=False)) def log_error(self, error: Exception, context: dict): """记录错误日志""" error_entry = { 'timestamp': datetime.now().isoformat(), 'error_type': type(error).__name__, 'error_message': str(error), 'context': context, 'type': 'error' } self.logger.error(json.dumps(error_entry, ensure_ascii=False))

实际项目中还会遇到各种环境配置问题，比如CUDA版本兼容性、依赖冲突等。建议建立完善的测试流程，每个组件都要有对应的单元测试和集成测试。

对于生产环境部署，要考虑灰度发布策略，先在小范围验证系统稳定性。监控方面除了技术指标，还要关注业务指标，比如用户满意度、问题解决率等。

持续学习是这个领域最重要的能力，新的模型、框架、工具不断涌现，需要保持技术敏感度，同时也要深入理解业务需求，让技术真正创造价值。