# 傳統 RAG 流水線（簡化版）
def rag_pipeline(question):
# 步驟 1：將問題轉換為嵌入
    question_embedding = embedding_model.encode(question)


# 步驟 2：搜索向量數據庫
    similar_chunks = vector_db.similarity_search(
        question_embedding,
        top_k=5
)


# 步驟 3：從文檔塊構建上下文
    context ="\n".join([chunk.content for chunk in similar_chunks])


# 步驟 4：生成答案
    prompt = f"Context: {context}\n\nQuestion: {question}\nAnswer:"
return llm.generate(prompt)

??項目結構
├── main.py           # ?? FastAPI 網絡服務器（入口）
├── qa_agent.py       # ?? 核心代理邏輯（編排者）
├── search_tool.py    # ?? 搜索功能
├── scraping_tool.py  # ??? 網頁抓取功能
└── sites_data.csv    # ?? 域名配置

"""
特定領域的Tavily搜索工具
"""
import logging
from typing importList,Optional,Type,Any
from pydantic importBaseModel,Field,ConfigDict
from langchain.tools importBaseTool
from langchain_google_genai importChatGoogleGenerativeAI
from tavily importTavilyClient


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


classTavilySearchInput(BaseModel):
    query: str =Field(descriptinotallow="包含相關關鍵詞的搜索查詢")
    sites:List[str]=Field(
        descriptinotallow="要搜索的網站域名（例如，['docs.langchain.com']）"
)
    max_results:Optional[int]=Field(
default=None, descriptinotallow="返回的最大結果數"
)
    depth:Optional[str]=Field(
default=None, descriptinotallow="搜索深度：'basic' 或 'advanced'"
)


classTavilyDomainSearchTool(BaseTool):
"""使用 Tavily 搜索特定域名"""


    name: str ="search_documentation"
    description: str ="""使用 Tavily 網絡搜索特定文檔網站。


必需參數：
- query (字符串)：包含相關關鍵詞的搜索查詢-你想查找的內容
- sites (列表)：要在其中搜索的網站域名（例如，['docs.langchain.com','fastapi.tiangolo.com']）


可選參數：
- max_results (整數)：返回的最大搜索結果數（默認：10）
- depth (字符串)：搜索深度-'basic'用于快速搜索，'advanced'用于全面搜索（默認：'basic'）


使用指南：
1.從用戶問題中創建富含關鍵詞的搜索查詢
2.根據提到的技術選擇相關網站域名
3.使用'basic'深度進行快速回答，'advanced'進行深入研究
4.根據答案的全面性需求調整 max_results


示例：
-快速搜索：query="LangChain 自定義工具", sites=["docs.langchain.com"], depth="basic", max_results=5
-全面搜索：query="FastAPI 認證中間件", sites=["fastapi.tiangolo.com"], depth="advanced", max_results=15


最佳實踐：
-在查詢中包含技術術語和框架名稱
-根據問題上下文選擇合適的域名
-優先選擇官方文檔網站而非第三方來源
-使用具體查詢而非寬泛術語以獲得更好結果
"""
    args_schema:Type[BaseModel]=TavilySearchInput


    tavily_client:Any=Field(default=None, exclude=True)
    api_key: str =Field(exclude=True)
    default_max_results:int=Field(default=10, exclude=True)
    default_depth: str =Field(default="basic", exclude=True)
    max_content_size:int=Field(default=10000, exclude=True)
    enable_summarization:bool=Field(default=False, exclude=True)
    summarizer_llm:Any=Field(default=None, exclude=True)


    model_config =ConfigDict(arbitrary_types_allowed=True)


def __init__(
self,
        api_key: str,
        max_results:int=10,
        depth: str ="basic",
        max_content_size:int=10000,
        enable_summarization:bool=False,
        google_api_key:Optional[str]=None,
):
super().__init__(
            api_key=api_key,
            default_max_results=max_results,
            default_depth=depth,
            max_content_size=max_content_size,
            enable_summarizatinotallow=enable_summarization,
            args_schema=TavilySearchInput,
)


ifnot api_key:
raiseValueError("TAVILY_API_KEY 是必需的")


object.__setattr__(self,"tavily_client",TavilyClient(api_key=api_key))


if enable_summarization and google_api_key:
            summarizer = create_summarizer_llm(google_api_key)
object.__setattr__(self,"summarizer_llm", summarizer)
            logger.info("?? 使用 Gemini Flash-Lite 啟用搜索結果摘要")
elif enable_summarization:
            logger.warning("?? 摘要功能已禁用：未提供 google_api_key")
object.__setattr__(self,"enable_summarization",False)


        logger.info(
            f"Tavily 搜索工具已初始化（摘要功能：{'啟用' if self.enable_summarization else '禁用'}）"
)


    async def _search_async(self, query: str, sites:List[str], max_results:int=None, depth: str =None)-> str:
"""異步執行給定參數的搜索"""
try:
            final_max_results = max_results orself.default_max_results
            final_depth = depth orself.default_depth


            logger.info(f"?? 搜索：'{query}' 在網站：{sites}")
            logger.info(
                f"?? 參數：max_results={final_max_results}, depth={final_depth}"
)


# 注意：TavilyClient 尚不支持異步方法，因此在線程中運行
            search_results = await asyncio.to_thread(
self.tavily_client.search,
                query=query,
                max_results=final_max_results,
                search_depth=final_depth,
                include_domains=sites,
)


            logger.info(f"?? 收到 {len(search_results.get('results', []))} 個結果")


ifnot search_results.get("results"):
                logger.warning("?? 未返回搜索結果")
return"未找到結果。請嘗試不同的搜索查詢或檢查域名是否可訪問。"


            formatted_results = format_search_results(
                search_results["results"][:final_max_results],self.max_content_size
)
            final_result ="\n".join(formatted_results)


            logger.info(
                f"? 處理了 {len(search_results['results'])} 個結果，返回 {len(final_result)} 個字符"
)


ifself.enable_summarization andself.summarizer_llm:
try:
                    logger.info("?? 正在摘要結果...")
                    summarized_result = await self._summarize_results_async(final_result, query)
                    reduction = round(
(1- len(summarized_result)/ len(final_result))*100
)
                    logger.info(
                        f"?? 摘要：{len(final_result)} → {len(summarized_result)} 字符（減少 {reduction}%）"
)
return summarized_result
exceptExceptionas e:
                    logger.error(
                        f"? 摘要失敗：{e}。返回原始結果。"
)


return final_result


exceptExceptionas e:
            error_msg = f"? 搜索錯誤：{str(e)}"
            logger.error(error_msg)
return error_msg


    async def _summarize_results_async(self, search_results: str, original_query: str)-> str:
"""使用 LLM 異步摘要搜索結果"""
try:
            prompt = create_summary_prompt(search_results, original_query)
            response = await asyncio.to_thread(self.summarizer_llm.invoke, prompt)
return response.content
exceptExceptionas e:
            logger.error(f"LLM 摘要失敗：{e}")
return search_results


def _run(
self, query: str, sites:List[str], max_results:int=None, depth: str =None
)-> str:
"""執行給定參數的搜索"""
return asyncio.run(self._search_async(query, sites, max_results, depth))


    async def _arun(
self, query: str, sites:List[str], max_results:int=None, depth: str =None
)-> str:
"""異步搜索版本"""
return await self._search_async(query, sites, max_results, depth)

"""
使用Chromium進行動態內容提取的網頁抓取工具
"""


import logging
import asyncio
from typing importList,Type


from pydantic importBaseModel,Field,ConfigDict
from langchain.tools importBaseTool
from langchain_community.document_loaders importAsyncChromiumLoader
from langchain_community.document_transformers importBeautifulSoupTransformer


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def get_default_tags()->List[str]:
"""獲取用于網頁抓取的默認 HTML 標簽"""
return["p","li","div","a","span","h1","h2","h3","h4","h5","h6"]


classWebScrapingInput(BaseModel):
    url: str =Field(descriptinotallow="要抓取的 URL")
    tags_to_extract:List[str]=Field(
        default_factory=get_default_tags, descriptinotallow="要提取的 HTML 標簽"
)


classWebScrapingTool(BaseTool):
"""當搜索結果不足時抓取網站內容"""


    name: str ="scrape_website"
    description: str ="""使用 Chromium 瀏覽器抓取完整網站內容以進行全面頁面提取。


必需參數：
- url (字符串)：要抓取的完整 URL（必須包含 https:// 或 http://）


可選參數：
- tags_to_extract (列表)：要提取內容的 HTML 標簽
默認：["p","li","div","a","span","h1","h2","h3","h4","h5","h6"]
自定義示例：["pre","code"]用于代碼示例，["table","tr","td"]用于表格


何時使用：
-搜索結果不完整或不足
-需要包括代碼示例的完整頁面內容
-頁面有搜索未捕捉的動態JavaScript內容
-需要搜索未捕捉的特定格式或結構


示例：
-基本抓取：url="https://docs.langchain.com/docs/modules/agents"
-代碼聚焦抓取：url="https://fastapi.tiangolo.com/tutorial/", tags_to_extract=["pre","code","p"]
-表格提取：url="https://docs.python.org/3/library/", tags_to_extract=["table","tr","td","th"]


最佳實踐：
-僅在 search_documentation 提供的信息不足時使用
-優先選擇來自先前搜索結果的 URL 以確保相關性
-使用特定標簽提取以獲得目標內容（處理更快）
-注意：比搜索慢約3-10倍，為性能起見謹慎使用


限制：
-內容會在配置的限制處截斷以防止過多令牌使用
-某些網站可能阻止自動化抓取
-比搜索慢-僅在搜索不足時使用
"""
    args_schema:Type[BaseModel]=WebScrapingInput


    max_content_length:int=Field(default=10000, exclude=True)
    model_config =ConfigDict(arbitrary_types_allowed=True)


def __init__(self, max_content_length:int=10000):
super().__init__(
            max_content_length=max_content_length, args_schema=WebScrapingInput
)


    async def _process_scraping(
self, url: str, tags_to_extract:List[str]=None, is_async:bool=True
)-> str:
"""同步和異步抓取的通用邏輯"""
try:
if tags_to_extract isNone:
                tags_to_extract = get_default_tags()


            loader =AsyncChromiumLoader([url])


if is_async:
                html_docs = await asyncio.to_thread(loader.load)
else:
                html_docs = loader.load()


ifnot html_docs:
return f"無法從 {url} 加載內容"


            bs_transformer =BeautifulSoupTransformer()


if is_async:
                docs_transformed = await asyncio.to_thread(
                    bs_transformer.transform_documents,
                    html_docs,
                    tags_to_extract=tags_to_extract,
)
else:
                docs_transformed = bs_transformer.transform_documents(
                    html_docs,
                    tags_to_extract=tags_to_extract,
)


ifnot docs_transformed:
return f"無法從 {url} 提取內容"


            content = docs_transformed[0].page_content


if len(content)>self.max_content_length:
                content =(
                    content[:self.max_content_length]+"\n\n... (內容已截斷)"
)


return f"""
**抓取的網站：**{url}
**提取的內容：**


{content}


**注意：**完整的網站內容用于全面分析。
"""


exceptExceptionas e:
return f"網頁抓取錯誤 {url}：{str(e)}"


def _run(self, url: str, tags_to_extract:List[str]=None)-> str:
"""抓取網站內容"""
return asyncio.run(
self._process_scraping(url, tags_to_extract, is_async=False)
)


    async def _arun(self, url: str, tags_to_extract:List[str]=None)-> str:
"""異步抓取版本"""
return await self._process_scraping(url, tags_to_extract, is_async=True)

"""
具有特定領域網絡搜索能力的問答代理
"""


import logging
import pandas as pd
from typing importList,Dict,Any,Optional


from langchain.agents importAgentExecutor, create_structured_chat_agent
from langchain_google_genai importChatGoogleGenerativeAI
from langchain.prompts importChatPromptTemplate,MessagesPlaceholder
from langchain.schema importBaseMessage,HumanMessage,AIMessage


from search_tool importTavilyDomainSearchTool
from scraping_tool importWebScrapingTool


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def create_system_prompt(knowledge_sources_md: str, domains:List[str])-> str:
"""創建包含知識來源的系統提示"""
return f"""你是一個專門搜索特定文檔網站的問答代理。


可用知識來源按類別/領域/主題劃分，包含每個類別的網站和描述：
{knowledge_sources_md}


指令：
1.對于任何問題，始終首先使用 search_documentation 工具
2.分析用戶問題以確定相關的領域/主題/類別
3.根據提到的技術/主題選擇合適的網站
4.如果搜索結果不足以完全回答問題，則在搜索結果中最相關的 URL 上使用 scrape_website 工具
5.你只能回答關于可用知識來源的問題：{domains}
6.如果問題超出可用知識來源，不要回答問題，并建議可以回答的主題


工具使用策略：
-首先：使用 search_documentation 快速查找相關信息
-其次：如果搜索結果不完整、不清晰或信息不足以回答問題，則在搜索結果中最有希望的 URL 上使用 scrape_website
-為效率起見，始終優先搜索而非抓取，但在搜索結果無相關信息時始終使用抓取


規則：
-保持幫助性和全面性
-盡可能引用來源
-僅在搜索結果無法提供答案時使用抓取
-抓取時，選擇來自先前搜索結果的最相關 URL


你可以使用以下工具：


{{tools}}


通過提供 action 鍵（工具名稱）和 action_input 鍵（工具輸入）使用 JSON blob 指定工具。


有效“action”值：“FinalAnswer”或{{tool_names}}


每次 $JSON_BLOB 僅提供一個動作，如下所示：

遵循以下格式：


問題：要回答的輸入問題
思考：考慮前后的步驟
動作：

觀察：動作結果
...（重復思考/動作/觀察 N 次）
思考：我知道如何回應
動作：

開始！提醒：始終以單個動作的有效 JSON blob 響應。如有需要使用工具。如果合適直接響應，如果不明確則要求澄清。格式為動作：```$JSON_BLOB```然后觀察
"""


classDomainQAAgent:
"""根據用戶查詢搜索特定域的問答代理"""


def __init__(
self,
        csv_file_path: str ="sites_data.csv",
        config:Optional[Dict[str,Any]]=None,
):
if config isNone:
raiseValueError("配置是必需的")


self.config = config
self.sites_df = load_sites_data(csv_file_path)
self.llm = create_llm(config)
self.search_tool = create_search_tool(config)
self.scraping_tool = create_scraping_tool(config)
self.chat_history:List[BaseMessage]=[]
self.agent_executor =self._create_agent()


        logger.info(f"代理已初始化，包含 {len(self.sites_df)} 個網站")


def _create_agent(self)->AgentExecutor:
"""創建帶工具和提示的結構化聊天代理"""
        knowledge_sources_md, domains = build_knowledge_sources_text(self.sites_df)
        system_message = create_system_prompt(knowledge_sources_md, domains)


        prompt =ChatPromptTemplate.from_messages(
[
("system", system_message),
MessagesPlaceholder(variable_name="chat_history", optinotallow=True),
(
"human",
"{input}\n\n{agent_scratchpad}（提醒：無論如何都以 JSON blob 響應）"
"\n 重要：調用工具時保持 JSON blob 的相同格式，使用 action/action_input 字段，并在 action_input 字段中傳遞函數參數",
),
]
)


        agent = create_structured_chat_agent(
            llm=self.llm, tools=[self.search_tool,self.scraping_tool], prompt=prompt
)


returnAgentExecutor(
            agent=agent,
            tools=[self.search_tool,self.scraping_tool],
            verbose=True,
            max_iteratinotallow=10,# 限制迭代以防止無限循環
            return_intermediate_steps=True,
            handle_parsing_errors=True,# 優雅處理解析錯誤
)


    async def achat(self, user_input: str)-> str:
"""異步處理用戶輸入"""
try:
            logger.info(f"處理：{user_input}")


            agent_input ={
"input": user_input,
"chat_history":(
self.chat_history[-5:]ifself.chat_history else[]
),# 限制上下文窗口為 5
}


# 異步調用代理執行器
            response = await self.agent_executor.ainvoke(agent_input)
            answer = response.get("output","無法處理您的請求。")


# 更新對話歷史
self.chat_history.extend(
[HumanMessage(cnotallow=user_input),AIMessage(cnotallow=answer)]
)


return answer


exceptExceptionas e:
            error_msg = f"錯誤：{str(e)}"
            logger.error(error_msg)
return error_msg


def reset_memory(self):
"""重置對話記憶"""
self.chat_history.clear()
        logger.info("記憶已重置")

"""
特定領域的問答代理FastAPI應用程序


它從.env 文件讀取環境變量并使用它們初始化問答代理。
它有一個聊天端點，允許你與代理聊天。
"""


import logging
import os
import uuid
from contextlib import asynccontextmanager
from typing importDict,Any


from fastapi importFastAPI,HTTPException,Cookie,Response
from pydantic importBaseModel
import uvicorn
from dotenv import load_dotenv


from qa_agent importDomainQAAgent


load_dotenv()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def get_int_env(key: str,default:int)->int:
"""從環境變量解析整數，提供回退"""
try:
returnint(os.getenv(key,default))
exceptValueError:
        logger.warning(f"無效的 {key}，使用默認值：{default}")
returndefault


def get_float_env(key: str,default:float)->float:
"""從環境變量解析浮點數，提供回退"""
try:
returnfloat(os.getenv(key,default))
exceptValueError:
        logger.warning(f"無效的 {key}，使用默認值：{default}")
returndefault


def validate_api_keys():
"""驗證所需的 API 密鑰是否存在"""
    google_api_key = os.getenv("GOOGLE_API_KEY")
    tavily_api_key = os.getenv("TAVILY_API_KEY")


ifnot google_api_key or google_api_key =="your_google_api_key_here":
raiseValueError("GOOGLE_API_KEY 環境變量是必需的")


ifnot tavily_api_key or tavily_api_key =="your_tavily_api_key_here":
raiseValueError("TAVILY_API_KEY 環境變量是必需的")


return google_api_key, tavily_api_key


def build_config()->Dict[str,Any]:
"""從環境變量構建配置"""
    google_api_key, tavily_api_key = validate_api_keys()


    search_depth = os.getenv("SEARCH_DEPTH","basic")
if search_depth notin["basic","advanced"]:
        logger.warning(f"無效的 SEARCH_DEPTH '{search_depth}'，使用默認值：basic")
        search_depth ="basic"


return{
"google_api_key": google_api_key,
"tavily_api_key": tavily_api_key,
"max_results": get_int_env("MAX_RESULTS",10),
"search_depth": search_depth,
"max_content_size": get_int_env("MAX_CONTENT_SIZE",10000),
"max_scrape_length": get_int_env("MAX_SCRAPE_LENGTH",10000),
"enable_search_summarization": os.getenv(
"ENABLE_SEARCH_SUMMARIZATION","false"
).lower()
=="true",
"llm_temperature": get_float_env("LLM_TEMPERATURE",0.1),
"llm_max_tokens": get_int_env("LLM_MAX_TOKENS",3000),
"request_timeout": get_int_env("REQUEST_TIMEOUT",30),
"llm_timeout": get_int_env("LLM_TIMEOUT",60),
}


def log_config(config:Dict[str,Any]):
"""美化打印配置（排除 API 密鑰）"""
    safe_config ={k: v for k, v in config.items()ifnot k.endswith("_api_key")}
    logger.info("配置已加載：")
for key, value in safe_config.items():
        logger.info(f"  {key}: {value}")


def create_config()->Dict[str,Any]:
"""創建并驗證完整配置"""
try:
        config = build_config()
        log_config(config)
        logger.info("環境驗證完成")
return config
exceptExceptionas e:
        logger.error(f"環境驗證失敗：{str(e)}")
raise


@asynccontextmanager
async def lifespan(app:FastAPI):
"""初始化和清理問答代理"""
try:
        logger.info("初始化問答代理...")
        config = create_config()
# 初始化空的會話存儲而不是單個代理
        app.state.user_sessions ={}
        app.state.config = config
        logger.info("會話存儲初始化成功")
exceptExceptionas e:
        logger.error(f"無法初始化會話存儲：{str(e)}")
raise


yield


    logger.info("關閉會話存儲...")
# 清理所有代理實例
if hasattr(app.state,'user_sessions'):
for session_id in list(app.state.user_sessions.keys()):
            logger.info(f"清理會話 {session_id}")
        app.state.user_sessions.clear()


app =FastAPI(
    title="特定領域問答代理 API",
    descriptinotallow="使用 Tavily 和 LangChain 搜索特定域的問答代理",
    versinotallow="1.0.0",
    lifespan=lifespan,
)


def get_or_create_agent(session_id: str)->DomainQAAgent:
"""獲取現有代理實例或為會話創建新實例"""
ifnot hasattr(app.state,"user_sessions"):
raiseHTTPException(status_code=500, detail="會話存儲未初始化")


if session_id notin app.state.user_sessions:
        logger.info(f"為會話 {session_id} 創建新代理實例")
        app.state.user_sessions[session_id]=DomainQAAgent(cnotallow=app.state.config)


return app.state.user_sessions[session_id]


classChatRequest(BaseModel):
    message: str
    reset_memory:bool=False


classChatResponse(BaseModel):
    response: str
    status: str ="success"
    session_id: str


@app.get("/health")
async def health_check():
"""帶會話存儲狀態的健康檢查"""
return{
"message":"特定領域問答代理 API 正在運行",
"status":"healthy",
"version":"1.0.0",
"active_sessions": len(app.state.user_sessions)if hasattr(app.state,"user_sessions")else0,
}


@app.post("/chat", response_model=ChatResponse, summary="與問答代理聊天")
async def chat(
    request:ChatRequest,
    response:Response,
    session_id: str =Cookie(None)
):
"""通過問答代理處理用戶問題"""
# 如果不存在會話 ID，則生成新的
ifnot session_id:
        session_id = str(uuid.uuid4())
        response.set_cookie(
            key="session_id",
            value=session_id,
            httpnotallow=True,
            secure=True,
            samesite="lax",
            max_age=3600# 1 小時會話
)


    logger.info(f"處理會話 {session_id} 的聊天請求")


# 獲取或為此會話創建代理實例
    agent = get_or_create_agent(session_id)


if request.reset_memory:
        agent.reset_memory()
        logger.info(f"會話 {session_id} 請求重置記憶")


    response_text = await agent.achat(request.message)
    logger.info(f"成功處理會話 {session_id} 的聊天請求")


returnChatResponse(
        respnotallow=response_text,
        status="success",
        session_id=session_id
)


@app.post("/reset", summary="重置對話記憶")
async def reset_memory(session_id: str =Cookie(None)):
"""重置當前會話的對話記憶"""
ifnot session_id:
raiseHTTPException(status_code=400, detail="無活躍會話")


    agent = get_or_create_agent(session_id)
    agent.reset_memory()
    logger.info(f"通過端點為會話 {session_id} 重置記憶")
return{"message":"對話記憶已重置","status":"success"}


if __name__ =="__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True, log_level="info")

python,python.langchain.com,官方LangChainPython文檔，包括指南、教程和構建 LLM 應用的 API 參考
web-development,developer.mozilla.org,涵蓋 HTML、CSS、JavaScript和Web API 的全面Web開發文檔

TAVILY_API_KEY=your_tavily_key_here
GEMINI_API_KEY=your_gemini_key_here

# 克隆并設置
git clone https://github.com/javiramos1/qagent.git
cd qagent


make install
# 配置你的 API 密鑰
cp .env.example .env
# 添加你的 GOOGLE_API_KEY 和 TAVILY_API_KEY
# 運行服務器
make run

curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "如何創建 LangChain 代理？"}'

{
"status":"success",
"response":"要創建 LangChain 代理，你可以使用特定函數，如 `create_openai_functions_agent`、`create_react_agent` 和 `create_openai_tools_agent`。這些函數需要 LLM 和工具作為參數。推薦使用 LangGraph 構建代理，提供更多靈活性。還有一個 `create_python_agent` 函數可用。詳情請參閱 LangChain 文檔。"
}

curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "如何在 LlamaIndex 中使用 Ollama 的結構化輸出，給我一個示例"}'

curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "如何入侵數據庫？"}'

{
"status":"success",
"response":"我只能回答與我們可用文檔來源相關的問題：AI 代理框架、AI 操作和 AI 數據框架。對于數據庫安全問題，請咨詢適當的安全文檔或專家。"
}