AI工程化实战：构建生产级AI编程助手与Harness架构

如果你正在寻找一个真正能提升日常开发效率的AI编程助手，而不是仅仅停留在Demo演示层面，那么今天讨论的AI工程化实战方案值得你认真了解。很多开发者尝试过各种AI编程工具，但往往遇到这样的困境：工具在简单任务上表现惊艳，一旦涉及复杂项目、长时间运行的编码任务，就会暴露出速率限制、响应延迟、工具链集成度低等问题，最终难以融入实际工作流。

本文要解决的核心问题，正是如何构建一个面向生产环境的AI辅助编程系统。我们将基于Claude Code、Codex等主流工具，深入探讨Harness工程化的实践路径。这不是简单的工具介绍，而是从工程角度分析如何让AI编程助手真正成为你每日开发的可靠伙伴。

从实际使用经验来看，很多开发者最初被Claude Code的代码生成质量所吸引，但在长期使用中会发现，Codex在速率限制、使用透明度、工具可靠性等方面的优势更加符合日常开发需求。这种从"演示效果"到"日常可用"的转变，正是AI工程化的核心价值所在。

1. 这篇文章真正要解决的问题

1.1 为什么AI编程助手难以融入实际工作流？

大多数AI编程工具在独立测试时表现良好，但在真实项目环境中往往遇到以下挑战：

• 上下文保持能力不足：长时间编码任务中，AI助手容易丢失之前的决策逻辑和代码结构
• 工具链集成度低：与现有开发环境、版本控制、测试框架的整合不够深入
• 速率限制和延迟问题：商业API的调用限制影响开发节奏
• 缺乏可预测性：相同提示词在不同时间可能产生差异较大的结果

1.2 Harness工程化如何解决这些问题？

Harness工程化的核心思想是将AI编程助手系统化地整合到开发流程中，而不是作为孤立工具使用。这包括：

• 建立标准化的工作流：定义清晰的AI助手调用模式和责任边界
• 设计可靠的错误处理机制：确保在API故障时开发流程不会完全中断
• 实现使用情况的可观测性：监控token消耗、响应时间等关键指标
• 构建模型无关的架构：避免过度依赖单一AI服务提供商

1.3 目标读者和预期收获

本文适合有以下需求的开发者：

• 已经体验过基础AI编程工具，希望提升到生产级应用
• 需要在团队中推广AI辅助编程的最佳实践
• 关心AI工具的长期可用性和成本效益
• 希望建立可维护、可扩展的AI开发基础设施

读完本文，你将掌握从工具选型、环境搭建到生产部署的完整实践路径。

2. 基础概念与核心原理

2.1 AI Agent在编程中的角色定位

AI编程Agent不是简单的代码补全工具，而是承担着更复杂的职责：

# 传统代码补全 vs AI编程Agent的对比 traditional_completion = { "scope": "单行或函数级建议", "context": "局部代码分析", "output": "语法正确的代码片段" } ai_programming_agent = { "scope": "完整功能模块开发", "context": "项目架构、业务逻辑、技术栈", "output": "可运行的功能代码+测试用例+文档" }

AI编程Agent的核心价值在于理解开发者的意图而非仅仅补全语法，这需要更深层次的代码库理解和任务分解能力。

2.2 Harness：AI工程化的基础设施

Harness可以理解为AI工具的运行环境和管控框架，主要包含以下组件：

Harness架构组成： ├── 模型路由层 (Model Routing) ├── 技能管理库 (Skill Management) ├── 工作流引擎 (Workflow Engine) ├── 状态持久化 (State Persistence) └── 监控指标 (Monitoring Metrics)

2.3 Claude Code vs Codex：技术特性对比

2.4 Agentic Coding的核心工作流

一个完整的Agentic Coding流程包含以下阶段：

1. 需求理解：将自然语言需求转化为技术任务
2. 方案设计：选择合适的技术栈和架构模式
3. 迭代开发：分步骤实现并持续验证
4. 代码审查：自动检查代码质量和最佳实践
5. 集成测试：确保新代码与现有系统兼容

3. 环境准备与前置条件

3.1 硬件和网络要求

• 操作系统：Windows 10/11, macOS 10.15+, Ubuntu 18.04+
• 内存：至少8GB，推荐16GB用于大型项目
• 网络：稳定访问OpenAI/Anthropic API的网络环境
• 存储：至少10GB可用空间用于缓存和模型数据

3.2 软件依赖安装

# 1. 安装Python 3.8+环境 python --version # 应输出 Python 3.8.x 或更高版本 # 2. 安装必要的系统工具 # Ubuntu/Debian sudo apt update && sudo apt install -y git curl build-essential # macOS brew install git curl # 3. 创建虚拟环境 python -m venv ai-coding-env source ai-coding-env/bin/activate # Linux/macOS # ai-coding-env\Scripts\activate # Windows # 4. 安装核心Python包 pip install openai anthropic requests python-dotenv

3.3 API密钥配置

创建配置文件~/.aiconfig：

[openai] api_key = sk-your-openai-key-here base_url = https://api.openai.com/v1 [anthropic] api_key = your-anthropic-key-here [project] default_model = gpt-4 fallback_model = claude-3-sonnet cache_dir = ./ai_cache

3.4 开发环境设置

# 创建项目结构 mkdir -p ai-coding-harness/{src,tests,config,logs} cd ai-coding-harness # 初始化Git仓库 git init echo "ai_cache/" >> .gitignore echo "*.env" >> .gitignore echo "__pycache__/" >> .gitignore

4. Claude Code深度集成实战

4.1 Claude Code安装与配置

# 安装Claude Code CLI工具 pip install claude-code # 验证安装 claude-code --version # 配置Anthropic API claude-code config set anthropic.api_key $ANTHROPIC_API_KEY

4.2 基础代码生成示例

创建示例项目文件example_project.py：

# example_project.py def calculate_fibonacci(n): """计算斐波那契数列的第n项""" if n <= 1: return n else: return calculate_fibonacci(n-1) + calculate_fibonacci(n-2) def main(): # 测试斐波那契函数 for i in range(10): print(f"F({i}) = {calculate_fibonacci(i)}") if __name__ == "__main__": main()

使用Claude Code进行代码优化：

# 分析并优化代码 claude-code analyze example_project.py --suggestions # 生成测试用例 claude-code generate-test example_project.py --framework pytest

4.3 项目级别的代码生成

对于真实项目，Claude Code可以处理更复杂的任务：

# 为整个项目生成API文档 claude-code generate-docs ./src --format markdown # 代码重构建议 claude-code refactor ./src --pattern "**/*.py" --strategy "improve-readability"

4.4 Claude Code的局限性分析

在实际使用中，我们发现Claude Code存在以下需要注意的问题：

• 长任务稳定性：超过30分钟的任务容易因网络问题中断
• 复杂项目理解：对大型代码库的架构理解有限
• 自定义工具集成：扩展性相对受限
• 成本控制：缺乏细粒度的使用统计和预算提醒

5. Codex生产级集成方案

5.1 Codex环境配置

# config/codex_config.py import os from openai import OpenAI class CodexConfig: def __init__(self): self.api_key = os.getenv('OPENAI_API_KEY') self.base_url = os.getenv('OPENAI_BASE_URL', 'https://api.openai.com/v1') self.client = OpenAI(api_key=self.api_key, base_url=self.base_url) def get_available_models(self): """获取可用的Codex模型列表""" return [ "gpt-4o", "gpt-4-turbo", "gpt-4", "gpt-3.5-turbo" ] def get_model_config(self, model_name): """获取模型特定配置""" configs = { "gpt-4o": {"max_tokens": 4096, "temperature": 0.1}, "gpt-4-turbo": {"max_tokens": 4096, "temperature": 0.1}, "gpt-4": {"max_tokens": 2048, "temperature": 0.2}, "gpt-3.5-turbo": {"max_tokens": 2048, "temperature": 0.3} } return configs.get(model_name, configs["gpt-4o"])

5.2 基于Codex的智能代码补全

# src/codex_assistant.py import os import json from pathlib import Path from config.codex_config import CodexConfig class CodexCodingAssistant: def __init__(self, model="gpt-4o"): self.config = CodexConfig() self.model = model self.model_config = self.config.get_model_config(model) self.conversation_history = [] def generate_code(self, prompt, context_files=None): """生成代码的核心方法""" # 构建上下文丰富的提示词 full_prompt = self._build_prompt(prompt, context_files) try: response = self.config.client.chat.completions.create( model=self.model, messages=[{"role": "user", "content": full_prompt}], max_tokens=self.model_config["max_tokens"], temperature=self.model_config["temperature"] ) generated_code = response.choices[0].message.content self._log_interaction(prompt, generated_code) return generated_code except Exception as e: print(f"Codex API调用失败: {e}") return None def _build_prompt(self, prompt, context_files): """构建包含上下文的提示词""" context = "" if context_files: for file_path in context_files: if Path(file_path).exists(): with open(file_path, 'r', encoding='utf-8') as f: context += f"文件 {file_path} 的内容:\n```python\n{f.read()}\n```\n\n" return f"""{context} 根据以上代码上下文，请完成以下任务： {prompt} 要求： 1. 生成完整可运行的代码 2. 包含适当的注释 3. 遵循Python最佳实践 4. 如果涉及新功能，请包含基本的测试用例 """

5.3 实际项目集成示例

假设我们需要为一个Web应用添加用户认证功能：

# 使用Codex助手生成认证模块 assistant = CodexCodingAssistant() auth_prompt = """ 请创建一个完整的用户认证系统，包含以下功能： 1. 用户注册（用户名、邮箱、密码） 2. 用户登录（JWT token生成） 3. 密码加密存储 4. 基本的权限验证中间件 技术栈：FastAPI + SQLAlchemy + JWT """ context_files = [ "src/models.py", "src/database.py", "src/config.py" ] auth_code = assistant.generate_code(auth_prompt, context_files) print(auth_code)

5.4 Codex的高级功能：迭代式开发

# src/iterative_developer.py class IterativeCodexDeveloper: def __init__(self, assistant): self.assistant = assistant self.iteration_history = [] def develop_feature(self, feature_description, acceptance_criteria): """迭代式开发功能""" print(f"开始开发功能: {feature_description}") # 第一轮：生成基础架构 architecture_prompt = f""" 为以下功能设计代码架构： 功能描述：{feature_description} 验收标准：{acceptance_criteria} 请提供： 1. 模块划分建议 2. 主要类和方法设计 3. 数据库模型设计（如需要） """ architecture = self.assistant.generate_code(architecture_prompt) self.iteration_history.append(("架构设计", architecture)) # 第二轮：实现核心逻辑 implementation_prompt = f""" 基于以下架构实现核心逻辑： {architecture} 重点实现业务核心功能。 """ implementation = self.assistant.generate_code(implementation_prompt) self.iteration_history.append(("核心实现", implementation)) return self.iteration_history

6. Harness工程化架构设计

6.1 核心Harness架构

# src/core_harness.py from abc import ABC, abstractmethod from typing import Dict, List, Any import logging import json from datetime import datetime class BaseAIAgent(ABC): """AI代理基类""" @abstractmethod def execute_task(self, task_description: str, context: Dict[str, Any]) -> str: pass @abstractmethod def get_capabilities(self) -> List[str]: pass class CodingAgent(BaseAIAgent): """编程专用AI代理""" def __init__(self, model_provider: str = "openai"): self.model_provider = model_provider self.logger = logging.getLogger(__name__) self.task_history = [] def execute_task(self, task_description: str, context: Dict[str, Any]) -> str: """执行编程任务""" # 构建完整的编程任务提示词 prompt = self._build_programming_prompt(task_description, context) try: if self.model_provider == "openai": result = self._call_openai_api(prompt) elif self.model_provider == "anthropic": result = self._call_anthropic_api(prompt) else: raise ValueError(f"不支持的模型提供商: {self.model_provider}") self._log_task(task_description, result, "success") return result except Exception as e: self.logger.error(f"任务执行失败: {e}") self._log_task(task_description, str(e), "failed") raise def _build_programming_prompt(self, task: str, context: Dict) -> str: """构建编程任务提示词""" return f""" 你是一个经验丰富的软件工程师。请完成以下编程任务： 项目上下文： {json.dumps(context, indent=2, ensure_ascii=False)} 具体任务： {task} 请提供： 1. 完整的实现代码 2. 必要的注释说明 3. 如果有依赖变更，请说明 4. 基本的测试用例 """ class HarnessOrchestrator: """Harness编排器""" def __init__(self): self.agents = {} self.workflow_registry = {} def register_agent(self, name: str, agent: BaseAIAgent): """注册AI代理""" self.agents[name] = agent def execute_workflow(self, workflow_name: str, input_data: Dict) -> Dict: """执行工作流""" if workflow_name not in self.workflow_registry: raise ValueError(f"工作流未注册: {workflow_name}") workflow = self.workflow_registry[workflow_name] return workflow.execute(input_data)

6.2 TOML配置驱动的代理路由

创建config/agent_routing.toml：

[agent_registry] default_agent = "codex_primary" [[agent_mappings]] pattern = ".*refactor.*" agent = "claude_code" priority = 1 [[agent_mappings]] pattern = ".*algorithm.*|.*data_structure.*" agent = "codex_primary" priority = 2 [[agent_mappings]] pattern = ".*documentation.*|.*comment.*" agent = "claude_code" priority = 3 [agents.claude_code] provider = "anthropic" model = "claude-3-sonnet-20240229" max_tokens = 4096 temperature = 0.1 [agents.codex_primary] provider = "openai" model = "gpt-4o" max_tokens = 4096 temperature = 0.1 fallback_model = "gpt-4-turbo" [rate_limiting] requests_per_minute = 30 tokens_per_minute = 90000 retry_attempts = 3

6.3 实现配置解析器

# src/config_parser.py import toml from pathlib import Path from typing import Dict, Any class HarnessConfig: def __init__(self, config_path: str = "config/agent_routing.toml"): self.config_path = Path(config_path) self.config_data = self._load_config() def _load_config(self) -> Dict[str, Any]: """加载TOML配置文件""" if not self.config_path.exists(): raise FileNotFoundError(f"配置文件不存在: {self.config_path}") with open(self.config_path, 'r', encoding='utf-8') as f: return toml.load(f) def get_agent_for_task(self, task_description: str) -> Dict[str, Any]: """根据任务描述选择合适的代理""" default_agent = self.config_data['agent_registry']['default_agent'] for mapping in self.config_data['agent_mappings']: import re if re.search(mapping['pattern'], task_description, re.IGNORECASE): agent_name = mapping['agent'] return { 'agent': agent_name, 'config': self.config_data['agents'][agent_name], 'priority': mapping['priority'] } # 返回默认代理 return { 'agent': default_agent, 'config': self.config_data['agents'][default_agent], 'priority': 0 } def get_rate_limits(self) -> Dict[str, int]: """获取速率限制配置""" return self.config_data.get('rate_limiting', {})

7. 完整项目实战：构建AI辅助Web应用

7.1 项目需求分析

让我们构建一个真实的项目：个人知识管理系统（PKM），使用AI辅助开发：

功能需求：

• 文章管理和分类
• 智能标签生成
• 内容搜索和推荐
• Markdown编辑器支持
• AI辅助写作

7.2 项目初始化设置

# 创建项目结构 mkdir -p pkm-system/{app,models,routes,services,static,templates} cd pkm-system # 创建基础配置文件 cat > requirements.txt << EOF fastapi==0.104.1 uvicorn==0.24.0 sqlalchemy==2.0.23 alembic==1.12.1 python-dotenv==1.0.0 openai==1.3.0 anthropic==0.7.4 EOF # 创建环境配置 cat > .env.example << EOF DATABASE_URL=sqlite:///./pkm.db OPENAI_API_KEY=your_openai_key_here ANTHROPIC_API_KEY=your_anthropic_key_here SECRET_KEY=your_secret_key_here EOF

7.3 使用AI生成数据库模型

# 使用Codex生成SQLAlchemy模型 model_prompt = """ 请为个人知识管理系统创建SQLAlchemy数据模型，包含以下实体： 1. User: 用户信息（id, username, email, password_hash, created_at） 2. Article: 文章（id, title, content, user_id, created_at, updated_at） 3. Tag: 标签（id, name, color） 4. ArticleTag: 文章标签关联表 要求： 1. 使用SQLAlchemy 2.0语法 2. 包含适当的关系定义 3. 添加基本的约束和索引 4. 使用合适的字段类型 """ # 将生成的模型保存到 models.py generated_models = assistant.generate_code(model_prompt) with open('models.py', 'w', encoding='utf-8') as f: f.write(generated_models)

7.4 实现AI辅助功能

# services/ai_assistant.py from src.codex_assistant import CodexCodingAssistant from src.config_parser import HarnessConfig import logging class PKMAIAssistant: def __init__(self): self.config = HarnessConfig() self.assistant = CodexCodingAssistant() self.logger = logging.getLogger(__name__) def generate_tags(self, article_content: str) -> list: """为文章内容生成智能标签""" prompt = f""" 请为以下文章内容生成5-8个相关标签： 文章内容： {article_content[:1000]} # 限制内容长度 要求： 1. 标签要具体且有代表性 2. 包含通用标签和特定内容标签 3. 返回JSON格式：{{"tags": ["tag1", "tag2", ...]}} """ try: result = self.assistant.generate_code(prompt) # 解析JSON结果 import json tags_data = json.loads(result) return tags_data.get("tags", []) except Exception as e: self.logger.error(f"标签生成失败: {e}") return [] def improve_writing(self, content: str, style: str = "professional") -> str: """改进写作风格""" prompt = f""" 请以下列风格改进这段文字： 目标风格：{style} 原文： {content} 要求： 1. 保持原意不变 2. 改进语法和表达 3. 增强可读性 4. 返回改进后的全文 """ return self.assistant.generate_code(prompt)

7.5 集成Harness到Web应用

# app/main.py from fastapi import FastAPI, Depends from services.ai_assistant import PKMAIAssistant from models import Article, Tag, User from sqlalchemy.orm import Session import databases app = FastAPI(title="AI辅助知识管理系统") ai_assistant = PKMAIAssistant() # 数据库依赖 async def get_db(): # 数据库连接逻辑 pass @app.post("/articles/") async def create_article(article_data: dict, db: Session = Depends(get_db)): """创建文章并自动生成标签""" # 保存基础文章信息 article = Article(**article_data) db.add(article) db.commit() # 使用AI生成标签 auto_tags = ai_assistant.generate_tags(article.content) for tag_name in auto_tags: tag = Tag(name=tag_name) db.add(tag) article.tags.append(tag) db.commit() return {"message": "文章创建成功", "article_id": article.id, "auto_tags": auto_tags} @app.put("/articles/{article_id}/improve") async def improve_article(article_id: int, style: str, db: Session = Depends(get_db)): """使用AI改进文章写作""" article = db.query(Article).filter(Article.id == article_id).first() if not article: return {"error": "文章不存在"} improved_content = ai_assistant.improve_writing(article.content, style) article.content = improved_content db.commit() return {"message": "文章改进完成", "improved_content": improved_content}

8. 监控与性能优化

8.1 使用情况监控

# src/monitoring.py import time import json from datetime import datetime from typing import Dict, List import psutil import requests class HarnessMonitor: def __init__(self, log_file: str = "logs/harness_metrics.log"): self.log_file = log_file self.metrics = { "api_calls": [], "token_usage": [], "response_times": [], "errors": [] } def log_api_call(self, agent: str, prompt_tokens: int, completion_tokens: int, response_time: float, success: bool = True): """记录API调用指标""" record = { "timestamp": datetime.now().isoformat(), "agent": agent, "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens, "response_time": response_time, "success": success } self.metrics["api_calls"].append(record) # 写入日志文件 with open(self.log_file, 'a', encoding='utf-8') as f: f.write(json.dumps(record) + '\n') def get_usage_statistics(self, hours: int = 24) -> Dict: """获取使用统计""" cutoff_time = datetime.now().timestamp() - (hours * 3600) recent_calls = [ call for call in self.metrics["api_calls"] if datetime.fromisoformat(call["timestamp"]).timestamp() > cutoff_time ] total_tokens = sum(call["total_tokens"] for call in recent_calls) avg_response_time = ( sum(call["response_time"] for call in recent_calls) / len(recent_calls) if recent_calls else 0 ) return { "total_calls": len(recent_calls), "total_tokens": total_tokens, "average_response_time": avg_response_time, "success_rate": ( sum(1 for call in recent_calls if call["success"]) / len(recent_calls) if recent_calls else 1 ) }

8.2 性能优化策略

# src/optimization.py from functools import lru_cache from typing import Dict, Any import hashlib import json class PromptOptimizer: def __init__(self): self.prompt_cache = {} @lru_cache(maxsize=1000) def optimize_prompt(self, prompt: str, context: Dict[str, Any]) -> str: """优化提示词以提高响应质量""" # 生成提示词指纹用于缓存 prompt_fingerprint = self._generate_fingerprint(prompt, context) if prompt_fingerprint in self.prompt_cache: return self.prompt_cache[prompt_fingerprint] # 提示词优化逻辑 optimized = self._apply_optimization_rules(prompt) self.prompt_cache[prompt_fingerprint] = optimized return optimized def _generate_fingerprint(self, prompt: str, context: Dict) -> str: """生成提示词指纹""" content = prompt + json.dumps(context, sort_keys=True) return hashlib.md5(content.encode()).hexdigest() def _apply_optimization_rules(self, prompt: str) -> str: """应用提示词优化规则""" optimization_rules = [ # 移除冗余描述 (r"请请", "请"), (r"非常非常", "非常"), # 标准化技术术语 (r"python代码", "Python代码"), (r"json格式", "JSON格式"), # 增强明确性 (r"写一个函数", "编写一个完整的函数实现"), (r"举个例子", "提供一个具体可运行的示例"), ] optimized = prompt for pattern, replacement in optimization_rules: import re optimized = re.sub(pattern, replacement, optimized) return optimized class RateLimitManager: """速率限制管理""" def __init__(self, limits: Dict[str, int]): self.limits = limits self.call_timestamps = [] self.token_usage = [] def can_make_request(self) -> bool: """检查是否可以发起新请求""" current_time = time.time() # 清理过期记录 one_minute_ago = current_time - 60 self.call_timestamps = [t for t in self.call_timestamps if t > one_minute_ago] # 检查调用频率限制 if len(self.call_timestamps) >= self.limits.get("requests_per_minute", 30): return False # 检查token使用限制 recent_tokens = sum( usage for timestamp, usage in self.token_usage if timestamp > one_minute_ago ) if recent_tokens >= self.limits.get("tokens_per_minute", 90000): return False return True def record_request(self, tokens_used: int): """记录请求使用情况""" current_time = time.time() self.call_timestamps.append(current_time) self.token_usage.append((current_time, tokens_used))

9. 生产环境部署与运维

9.1 Docker容器化部署

创建Dockerfile：

FROM python:3.11-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ git \ curl \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建日志目录 RUN mkdir -p logs # 设置环境变量 ENV PYTHONPATH=/app ENV PYTHONUNBUFFERED=1 # 启动命令 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

创建docker-compose.yml：

version: '3.8' services: ai-pkm-system: build: . ports: - "8000:8000" environment: - DATABASE_URL=sqlite:///./pkm.db - OPENAI_API_KEY=${OPENAI_API_KEY} - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY} volumes: - ./data:/app/data - ./logs:/app/logs restart: unless-stopped # 可选：添加监控服务 monitor: image: grafana/grafana:latest ports: - "3000:3000" environment: - GF_SECURITY_ADMIN_PASSWORD=admin volumes: - grafana-storage:/var/lib/grafana restart: unless-stopped volumes: grafana-storage:

9.2 持续集成配置

创建.github/workflows/deploy.yml：

name: Deploy AI PKM System on: push: branches: [ main ] jobs: test-and-deploy: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.11' - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install pytest - name: Run tests run: | pytest tests/ -v - name: Build and push Docker image if: success() run: | docker build -t my-registry/ai-pkm-system:latest . docker push my-registry/ai-pkm-system:latest - name: Deploy to production if: success() run: | # 部署脚本 echo "Deploying to production..."

9.3 安全最佳实践

# src/security.py import os import secrets from typing import Optional from fastapi import HTTPException, Header class SecurityManager: def __init__(self): self.api_keys = self._load_api_keys() def _load_api_keys(self) -> dict: """加载API密钥配置""" return { "openai": os.getenv('OPENAI_API_KEY', ''), "anthropic": os.getenv('ANTHROPIC_API_KEY', '') } def validate_api_key(self, provided_key: Optional[str] = Header(None)) -> bool: """验证API密钥""" if not provided_key: raise HTTPException(status_code=401, detail="API密钥缺失") # 在实际项目中，这里应该查询数据库验证密钥 expected_key = os.getenv('INTERNAL_API_KEY') if not secrets.compare_digest(provided_key, expected_key or ''): raise HTTPException(status_code=401, detail="无效的API密钥") return True def sanitize_prompt(self, prompt: str) -> str: """清理提示词中的潜在安全风险""" import html # 转义HTML特殊字符 sanitized = html.escape(prompt) # 移除潜在的危险模式 dangerous_patterns = [ r"系统命令", r"文件操作", r"网络请求", r"数据库查询" ] import re for pattern in dangerous_patterns: sanitized = re.sub(pattern, "[已过滤]", sanitized, flags=re.IGNORECASE) return sanitized

10. 常见问题与排查指南

10.1 安装与配置问题

10.2 运行时性能问题