from langchain_core.prompts import PromptTemplate

# 客服郵件生成器
email_template = PromptTemplate.from_template(
    """
    為以下情況撰寫一封專業(yè)的客服郵件：
    
    客戶問題：{issue}
    客戶姓名：{customer_name}
    緊急程度：{urgency}
    
    語氣：共情且以解決方案為導(dǎo)向
    包含：問題確認(rèn)、解決方案步驟、后續(xù)跟進提議
    """
)
# 生成個性化回復(fù)
prompt = email_template.invoke({
    "issue": "支付后無法訪問高級功能",
    "customer_name": "Sarah",
    "urgency": "High"
})
print(prompt.to_string())

from langchain_core.prompts import ChatPromptTemplate

# AI 編程導(dǎo)師設(shè)置
coding_mentor_template = ChatPromptTemplate.from_messages([
    ("system", """你是 Python 編程專家，你的教學(xué)風(fēng)格是：
    ? 耐心且鼓勵
    ? 提供清晰的解釋和示例
    ? 提出引導(dǎo)性問題幫助學(xué)生思考
    ? 慶祝學(xué)生的進步和學(xué)習(xí)時刻
    """),
    ("user", "我在 {topic} 上有困難，能幫我理解 {specific_question} 嗎？")
])
# 創(chuàng)建學(xué)習(xí)會話
chat_prompt = coding_mentor_template.invoke({
    "topic": "列表推導(dǎo)式",
    "specific_question": "什么時候用它，什么時候用普通循環(huán)"
})

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

# 帶記憶的項目管理助手
project_template = ChatPromptTemplate.from_messages([
    ("system", "你是項目管理助手，跟蹤任務(wù)、截止日期和團隊進展。"),
    MessagesPlaceholder(variable_name="conversation_history"),
    ("user", "{current_request}")
])
# 模擬對話歷史
conversation_history = [
    HumanMessage(cnotallow="我們有個新項目：移動應(yīng)用開發(fā)"),
    AIMessage(cnotallow="好的！已記錄移動應(yīng)用開發(fā)項目。時間線是怎樣的？"),
    HumanMessage(cnotallow="我們需要在3個月內(nèi)上線"),
    AIMessage(cnotallow="已記錄3個月時間線。關(guān)鍵里程碑有哪些？")
]
# 繼續(xù)對話
current_prompt = project_template.invoke({
    "conversation_history": conversation_history,
    "current_request": "添加任務(wù)：設(shè)計用戶界面 mockups，下周五截止"
})

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 內(nèi)容創(chuàng)作流水線
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
output_parser = StrOutputParser()
# 博客文章生成鏈
blog_chain = (
    {"topic": lambda x: x, "audience": lambda x: "developers"}
    | PromptTemplate.from_template(
        "為 {audience} 撰寫一篇關(guān)于 {topic} 的吸引人博客文章，"
        "包含實用示例和可操作的建議。"
    )
    | llm
    | output_parser
)
result = blog_chain.invoke("API Rate Limiting Best Practices")
print(result)

from langchain_core.runnables import RunnablePassthrough

# 數(shù)據(jù)分析工作流
defextract_metrics(data):
    returnf"從 {data[:100]} 中提取的關(guān)鍵指標(biāo)..."
defgenerate_insights(context):
    returnf"基于 {context['metrics']} 的洞察：[分析內(nèi)容]"
# 多步驟分析鏈
analysis_chain = (
    RunnablePassthrough.assign(metrics=extract_metrics)
    | RunnablePassthrough.assign(insights=generate_insights)
    | PromptTemplate.from_template(
        "創(chuàng)建執(zhí)行摘要報告：\n"
        "數(shù)據(jù)：{input}\n"
        "指標(biāo)：{metrics}\n"
        "洞察：{insights}\n"
        "格式為專業(yè)商務(wù)報告。"
    )
    | llm
    | output_parser
)

from typing import TypedDict, List
from langgraph.graph import StateGraph, END

classCustomerSupportState(TypedDict):
    customer_message: str
    sentiment: str
    category: str
    priority: str
    resolution_steps: List[str]
    escalated: bool

# 定義工作流節(jié)點
defanalyze_sentiment(state: CustomerSupportState):
    # 情感分析邏輯
    message = state["customer_message"]
    # 簡化的情感檢測
    if"angry"in message.lower() or"frustrated"in message.lower():
        sentiment = "negative"
    elif"happy"in message.lower() or"great"in message.lower():
        sentiment = "positive"
    else:
        sentiment = "neutral"
    
    return {"sentiment": sentiment}

defcategorize_issue(state: CustomerSupportState):
    message = state["customer_message"].lower()
    if"billing"in message or"payment"in message:
        category = "billing"
    elif"technical"in message or"bug"in message:
        category = "technical"
    else:
        category = "general"
    
    return {"category": category}

defdetermine_priority(state: CustomerSupportState):
    if state["sentiment"] == "negative"and state["category"] == "billing":
        priority = "high"
    elif state["category"] == "technical":
        priority = "medium"
    else:
        priority = "low"
    
    return {"priority": priority}

def should_escalate(state: CustomerSupportState):
    """決定是否需要升級工單"""
    if state["priority"] == "high"or state["sentiment"] == "negative":
        return"escalate"
    else:
        return"resolve"

defescalate_ticket(state: CustomerSupportState):
    return {
        "escalated": True,
        "resolution_steps": ["已升級至高級支持團隊", "經(jīng)理將在1小時內(nèi)聯(lián)系"]
    }

defauto_resolve(state: CustomerSupportState):
    steps = {
        "billing": ["檢查支付狀態(tài)", "發(fā)送賬單說明", "提供支付計劃"],
        "technical": ["收集系統(tǒng)信息", "應(yīng)用常見修復(fù)", "安排后續(xù)跟進"],
        "general": ["提供相關(guān)文檔", "提供額外協(xié)助"]
    }
    
    return {
        "escalated": False,
        "resolution_steps": steps.get(state["category"], ["已提供一般協(xié)助"])
    }

# 構(gòu)建工作流圖
workflow = StateGraph(CustomerSupportState)
# 添加節(jié)點
workflow.add_node("analyze_sentiment", analyze_sentiment)
workflow.add_node("categorize_issue", categorize_issue)
workflow.add_node("determine_priority", determine_priority)
workflow.add_node("escalate_ticket", escalate_ticket)
workflow.add_node("auto_resolve", auto_resolve)
# 定義流程
workflow.set_entry_point("analyze_sentiment")
workflow.add_edge("analyze_sentiment", "categorize_issue")
workflow.add_edge("categorize_issue", "determine_priority")
# 條件路由
workflow.add_conditional_edges(
    "determine_priority",
    should_escalate,
    {
        "escalate": "escalate_ticket",
        "resolve": "auto_resolve"
    }
)
workflow.add_edge("escalate_ticket", END)
workflow.add_edge("auto_resolve", END)
# 編譯并運行
app = workflow.compile()

from langchain.tools import Tool
from langchain.agents import create_react_agent

# 定義自定義工具
@tool
defsearch_knowledge_base(query: str) -> str:
    """搜索內(nèi)部知識庫獲取信息"""
    knowledge_db = {
        "api limits": "標(biāo)準(zhǔn)計劃：每小時1000次請求，高級計劃：每小時10000次",
        "password reset": "使用忘記密碼鏈接，檢查郵件，遵循說明",
        "billing cycle": "按注冊日期月度計費，提供年度折扣"
    }
    
    for key, value in knowledge_db.items():
        if key in query.lower():
            return value
    return"知識庫中未找到信息"

@tool
defcreate_support_ticket(issue: str, priority: str) -> str:
    """為復(fù)雜問題創(chuàng)建支持工單"""
    ticket_id = f"TICKET-{hash(issue) % 10000}"
    returnf"支持工單 {ticket_id} 已創(chuàng)建，優(yōu)先級：{priority}"

@tool
defsend_email_notification(recipient: str, subject: str, message: str) -> str:
    """向客戶發(fā)送郵件通知"""
    returnf"郵件已發(fā)送至 {recipient}，主題：{subject}"

# 創(chuàng)建帶工具的代理
tools = [search_knowledge_base, create_support_ticket, send_email_notification]
agent_prompt = """
你是一個樂于助人的客服代理，你可以：
1. 搜索知識庫獲取答案
2. 為復(fù)雜問題創(chuàng)建支持工單
3. 向客戶發(fā)送郵件通知
始終保持幫助性和專業(yè)性，必要時使用工具。
客戶問題：{input}
{agent_scratchpad}
"""

# 代理執(zhí)行示例
defrun_support_agent(customer_query: str):
    # 這里將與實際代理執(zhí)行整合
    print(f"正在處理：{customer_query}")
    print("代理思考中...")
    print("工具使用：search_knowledge_base")
    print("回復(fù)已生成！")

class MultiAgentSystem:
    def__init__(self):
        self.agents = {
            "researcher": self.create_researcher_agent(),
            "writer": self.create_writer_agent(),
            "reviewer": self.create_reviewer_agent()
        }
    
    defcreate_researcher_agent(self):
        return {
            "role": "研究專家",
            "tools": ["web_search", "data_analysis", "fact_checking"],
            "prompt": "收集給定主題的全面信息"
        }
    
    defcreate_writer_agent(self):
        return {
            "role": "內(nèi)容創(chuàng)作者",
            "tools": ["content_generation", "style_formatting"],
            "prompt": "基于研究結(jié)果創(chuàng)建吸引人的內(nèi)容"
        }
    
    defcreate_reviewer_agent(self):
        return {
            "role": "質(zhì)量保證",
            "tools": ["grammar_check", "fact_verification", "style_review"],
            "prompt": "審查并提升內(nèi)容質(zhì)量"
        }
    
    defcoordinate_workflow(self, task: str):
        """協(xié)調(diào)多代理協(xié)作"""
        # 研究階段
        research_data = self.agents["researcher"]["process"](task)
        
        # 寫作階段
        draft_content = self.agents["writer"]["process"](research_data)
        
        # 審查階段
        final_content = self.agents["reviewer"]["process"](draft_content)
        
        return final_content

# 使用示例
multi_agent = MultiAgentSystem()
result = multi_agent.coordinate_workflow("撰寫關(guān)于 AI 倫理的綜合指南")

cot_template = PromptTemplate.from_template("""
一步步解決這個問題：
問題：{problem}
讓我仔細(xì)思考：
步驟1：理解問題要求
步驟2：識別關(guān)鍵信息
步驟3：應(yīng)用相關(guān)原則/公式
步驟4：計算/推理解決方案
步驟5：驗證答案是否合理
解決方案：
""")

few_shot_template = PromptTemplate.from_template("""
將以下客戶反饋分類為 Positive、Negative 或 Neutral：
示例：
輸入："產(chǎn)品很好用，運輸也很快！"
輸出：Positive
輸入："客服很差，非常失望"
輸出：Negative
輸入："產(chǎn)品按時到達，符合預(yù)期"
輸出：Neutral
現(xiàn)在分類這個：
輸入：{feedback}
輸出：
""")

def self_correcting_agent(initial_response: str, validation_criteria: str):
    """實現(xiàn)自我糾正機制"""
    
    correction_template = PromptTemplate.from_template("""
    原始回復(fù)：{response}
    
    驗證標(biāo)準(zhǔn)：{criteria}
    
    根據(jù)標(biāo)準(zhǔn)審查你的回復(fù)。如果需要改進：
    1. 識別具體問題
    2. 提供修正版本
    3. 解釋改進的內(nèi)容
    
    最終回復(fù)：
    """)
    
    return correction_template.format(
        respnotallow=initial_response,
        criteria=validation_criteria
    )

from langchain_core.runnables import RunnableLambda
import logging

defsafe_llm_call(prompt: str, fallback_response: str = "抱歉，我在處理你的請求時遇到問題。"):
    """安全的 LLM 調(diào)用包裝器，帶錯誤處理"""
    try:
        # 這里是你的 LLM 調(diào)用
        response = llm.invoke(prompt)
        return response
    except Exception as e:
        logging.error(f"LLM 調(diào)用失?。簕e}")
        return fallback_response

# 帶回退機制的韌性鏈
resilient_chain = (
    prompt_template
    | RunnableLambda(safe_llm_call)
    | output_parser
)

from langchain.memory import ConversationBufferWindowMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory

# 持久化對話內(nèi)存
def create_persistent_memory(session_id: str):
    return ConversationBufferWindowMemory(
        chat_memory=RedisChatMessageHistory(
            session_id=session_id,
            url="redis://localhost:6379"
        ),
        memory_key="chat_history",
        k=10  # 保留最近10次交流
    )

import asyncio
from langchain_core.runnables import RunnableParallel

asyncdefparallel_processing_example():
    """并發(fā)處理多個任務(wù)"""
    
    parallel_chain = RunnableParallel(
        summary=summarization_chain,
        sentiment=sentiment_chain,
        keywords=keyword_extraction_chain
    )
    
    results = await parallel_chain.ainvoke({"text": document})
    return results

# 緩存重復(fù)查詢
from functools import lru_cache
@lru_cache(maxsize=1000)
defcached_llm_call(prompt_hash: str):
    """緩存相同提示的 LLM 回復(fù)"""
    return llm.invoke(prompt_hash)