pip install langchain openai python-dotenv

OPENAI_API_KEY=your_api_key_here

from dotenv import load_dotenv
import os

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

# 加载文档
loader = TextLoader("student_data.txt")
documents = loader.load()

# PII检测和分析器
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def anonymize_text(text):
    # 检测PII
    results = analyzer.analyze(text=text, language="en")
    # 匿名化处理
    anonymized_text = anonymizer.anonymize(text=text, analyzer_results=results)
    return anonymized_text.text

# 处理文档中的PII
safe_documents = [anonymize_text(doc.page_content) for doc in documents]
print(safe_documents[:1])  # 打印第一个处理后的文档示例

from langchain.llms import OpenAI

llm = OpenAI(
    openai_api_key=api_key,
    model="gpt-3.5-turbo",
    temperature=0,
    request_timeout=60,
    # 重要隐私设置
    headers={
        "OpenAI-Beta": "assistant=v2",
        "X-Data-Privacy": "strict" 
    }
)

response = llm("解释牛顿第一定律给高中生")
print(response)

import tempfile
import shutil
from contextlib import contextmanager

@contextmanager
def temp_cache_directory():
    """创建临时缓存目录上下文管理器"""
    temp_dir = tempfile.mkdtemp()
    try:
        yield temp_dir
    finally:
        # 无论是否发生异常都清理临时目录
        shutil.rmtree(temp_dir, ignore_errors=True)
        print(f"已清理临时目录: {temp_dir}")

# 使用示例
with temp_cache_directory() as cache_dir:
    # 在这里进行需要缓存的操作...
    print(f"正在使用临时目录: {cache_dir}")

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory
from langchain.callbacks import FileCallbackHandler
import logging

class SecureEduAIAssistant:
    def __init__(self):
        self.setup_logging()

        # 定义隐私友好的提示模板 (避免收集不必要的信息)
        self.prompt_template = PromptTemplate(
            input_variables=["topic", "grade_level"],
            template="""
            你是一位{grade_level}教师助手。请用适合该年龄段的方式解释{topic}。
            注意:
            1.不要询问或提及任何学生的个人信息 
            2.保持解释简洁专业 
            3.提供1个课堂活动建议"""
        )

        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
            input_key="topic"
        )

    def setup_logging(self):
        """配置安全的日志记录 (不记录敏感信息)"""
        self.logger = logging.getLogger(__name__)
        handler = logging.FileHandler('edu_ai.log')
        handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
        self.logger.addHandler(handler)

        # Redact sensitive info from logs (简化版)
        old_factory = logging.getLogRecordFactory()

        def secure_record_factory(*args, **kwargs):
            record = old_factory(*args, **kwargs)
            if "api_key" in record.getMessage().lower():
                record.msg = "[REDACTED]"
            return record

        logging.setLogRecordFactory(secure_record_factory)

    def explain_concept(self, topic, grade_level="高中"):
        """安全地解释教育概念"""
        with temp_cache_directory() as cache_dir:
            try:
                llm = OpenAI(
                    openai_api_key=os.getenv("OPENAI_API_KEY"),
                    cache=True,
                    cache_path=cache_dir,
                    model="gpt-3.5-turbo",
                    temperature=0.7,
                    max_tokens=500,
                    headers={"X-Data-Privacy": "strict"}
                )

                conversation = LLMChain(
                    llm=llm,
                    prompt=self.prompt_template,
                    memory=self.memory,
                    verbose=True,
                    callbacks=[FileCallbackHandler('edu_ai.log')]
                )

                response = conversation.run({
                    "topic": anonymize_text(topic),
                    "grade_level": grade_level  
                })

                return response

            except Exception as e:
                self.logger.error(f"教学助手错误: {str(e)}")
                return "抱歉，教学助手暂时不可用"

# 使用示例            
assistant = SecureEduAIAssistant()
physics_explanation = assistant.explain_concept("量子力学基础") 
print(physics_explanation)

from langchain.chains import TransformChain, SequentialChain 
from keycloak import KeycloakOpenID

def auth_check(inputs: dict) -> dict:
    """验证用户权限的转换链"""

    # Keycloak配置 (从环境变量获取)
    keycloak_openid = KeycloakOpenID(
        server_url=os.getenv("KEYCLOAK_URL"),
        client_id=os.getenv("EDU_CLIENT_ID"),
        realm_name=os.getenv("KEYCLOAK_REALM"),
        client_secret_key=os.getenv("CLIENT_SECRET")
    )

    token = inputs["token"]

    try:
        # Token验证和权限检查 (实际项目应更详细)
        userinfo = keycloak_openid.userinfo(token)

        if "educator" not in userinfo.get("roles", []):
            raise ValueError("无教育工作者权限")

        inputs["user_id"] = userinfo["sub"]  

    except Exception as e:
        raise ValueError(f"认证失败: {str(e)}")

    return inputs

auth_chain = TransformChain(
    input_variables=["token"], 
    output_variables=["user_id"],
    transform=auth_check,
)

# Combine with LLM chain...
full_secure_chain = SequentialChain(
   chains=[auth_chain, assistant.conversation],
   input_variables=["token", "topic", "grade_level"],
   verbose=True  
) 

def delete_user_data(user_id):
   """删除特定用户的全部对话历史"""

   # Step1:从数据库中删除结构化数据 
   db.execute(f"DELETE FROM chat_history WHERE user_id='{user_id}'")

   # Step2:删除向量存储中的嵌入 (如果使用了RAG)
   vectorstore.delete(filter={"user_id": user_id})

   # Step3:确认删除完成并记录审计日志  
   logger.info(f"用户{user_id}数据已按GDPR要求删除") 

# AWS S3加密存储对话历史 (CLI示例) 
aws s3api put-bucket-encryption \
 --bucket edu-ai-data-bucket \
 --server-side-encryption-configuration '{
   "Rules": [{
     "ApplyServerSideEncryptionByDefault": {
       "SSEAlgorithm": "aws:kms"
     }
   }]
 }'

from langsmith.run_trees import RunTree 

def log_run_safely(inputs, outputs):
   """安全地记录运行日志到LangSmith"""

   run_tree = RunTree(
       name="Secure Edu AI Run",
       inputs={k:v for k,v in inputs.items() if k != 'token'},
       outputs={
           'response': outputs[:200] + "...[TRUNCATED]" if len(outputs)>200 else outputs,
           'status': 'success'  
       },
       project_name="edu-ai-prod",
       metadata={
           'privacy_level': 'high',
           'data_protection': 'gdpr'
       }
   )

   run_tree.post() 

# Then add this callback to your chain...

name: Security Scan 

on: [push]

jobs:
 secretscan:
   runs-on: ubuntu-latest  

   steps:
     - uses: actions/checkout@v3

     - name: Detect secrets 
       uses: gitleaks/gitleaks-action@v8  
       with:  
         config-path: .gitleaks.toml

     - name: Dependency check  
       uses: pyupio/safety@v1  

     - name: Bandit SAST  
       run: pip install bandit && bandit -r . -ll      

title = "Edu AI Leaks Config"

[[rules]]
description = "OpenAI API Key"
regex = '''sk-[a-zA-Z0-9]{32,}'''
tags = ["key", "openai"]

def postprocess_response(text):
   """过滤LLM响应中的潜在PII"""  

   patterns_to_redact = [
      r'\b\d{3}-\d{2}-\d{4}\b',      # SSN模式  
      r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', # Email  
      r'\b\d{10}\b'                  # Phone-like numbers   
   ]

   for pattern in patterns_to_redact:
      text = re.sub(pattern, '[REDACTED]', text)     

   return text 

# Usage:
safe_response = postprocess_response(llm_response)  

bandit -r . -ll --skip B101,B404,B603   
safety check --full-report   

caldera-cli adversary emulate 'Educator Impersonation' --target http://localhost:8000/api/edu/ask  

pip install compliance-checker   
compliance-checker --framework gdpr --level high ./src/    

// React组件中的数据安全处理 

function EduQuestionForm() {
 const [question, setQuestion] = useState('');
 const [response, setResponse] = useState(null);
 const [isProcessing, setIsProcessing] = useState(false);

 const handleSubmitSecureQuestion async () => {
     setIsProcessing(true);

     try {
         // Step1:前端PII过滤 (基本防护) 
         const sanitizedQuestion =
             question.replace(/\b\w+@\w+\.\w+\b/g,'[EMAIL]')
                     .replace(/\d{10}/g,'[PHONE]');

         // Step2:HSTS加密传输       
         const res await fetch('/api/edu/ask', {
             method:'POST',
             headers:{
                 'Content-Type':'application/json',
                 'Authorization': `Bearer ${getAuthToken()}`
             },
             body JSON.stringify({
                 question sanitizedQuestion,
                 nonce window.crypto.randomUUID() // CSRF防护                
             })
         });

         // Step3:Safe渲染(XSS防护)         
         setResponse(DOMPurify.sanitize(await res.text()));

     } catch(err) {
         console.error('[SECURE] Question error:', err);     
     } finally {
         setIsProcessing(false);    
     }
 };
}

version '3.8'

services edu ai app build context ./app dockerfile Dockerfile.prod environment OPENAI_API_KEY ${SECRET_API_KEY} ports - "127.0.0..18000..8000" security opt no new privileges true cap drop ALL volumes ./app/data /var/lib/app..ro healthcheck test ["CMD", curl f http://localhost..8000/health || exit..1"] restart unless stopped deploy resources limits cpus '0..5' memory..512M networks edu net 

networks edu net driver bridge attachable false internal true 

x security headers labels traefik http middleware secure headers headers sslRedirect true stsIncludeSubdomains true stsPreload true stsSeconds..31536000 contentType nosniff frameDeny true browserXssFilter true referrerPolicy strict origin when cross origin featurePolicy camera none microphone none geolocation none midi none sync xhr none magnetometer none gyroscope none speaker none vibrate none fullscreen none payment none usb none autoplay none document domain self plugin types application/pdf application/x shockwave flash script src self unsafe inline style src self unsafe inline img src self data blob font src self connect src self ws localhost *.edu.example.com media src self object src none child src frame ancestors none form action self base uri self worker src blob frame src sameorigin manifest src self prefetch src self default src none require trusted types for script style script nonce {{random nonce}} style nonce {{random nonce}} block all mixed content upgrade insecure requests expect ct enforce max age..30 report uri https example.com report endpoint require sri for script style sandbox allow forms allow same origin allow scripts allow popups allow modals allow downloads allow top navigation allow pointer lock allow presentation allow popups to escape sandbox allow top navigation by user activation strict transport security max age..31536000 include subdomains preload x content type options nosniff x frame options DENY x xss protection..1 mode block referrer policy strict origin when cross origin permissions policy geolocation () camera () microphone () usb () fullscreen () payment () autoplay () document domain () plugin types application pdf application x shockwave flash script src https example.com unsafe inline unsafe eval style src https example.com unsafe inline img src https example.com data blob connect src https example.com wss example.com media src https example.com object src none child src https example.com frame ancestors https example.com form action https example.com base uri self worker src blob frame src sameorigin manifest src https example.com prefetch src https example.com default src none require trusted types default report only trusted types default enforce trusted types default policy names default allowed trusted types default disallowed trusted types default fallback policy names default report only policy names default allowed fallback policy names default disallowed fallback policy names default sandbox allow forms allow same origin allow scripts allow popups allow modals allow downloads allow top navigation allow pointer lock allow presentation allow popups to escape sandbox allow top navigation by user activation strict transport security max age..31536000 include subdomains preload x content type options nosniff x frame options DENY x xss protection..1 mode block referrer policy strict origin when cross origin permissions policy geolocation () camera () microphone () usb () fullscreen () payment () autoplay ()