import openai

client = openai.OpenAI(api_key="your-api-key")

# 文本对话
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "解释什么是Transformer架构"}]
)
print(response.choices[0].message.content)

# 图像理解
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "分析这张网络架构图"},
            {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}}
        ]
    }]
)
print(response.choices[0].message.content)

# Windows安装（下载安装包或winget）
winget install Ollama.Ollama

# 拉取模型
ollama pull qwen2.5:7b
ollama pull deepseek-r1:7b
ollama pull llama3.1:8b

# 运行模型
ollama run qwen2.5:7b

# 查看已安装模型
ollama list

# 启动API服务（默认11434端口）
ollama serve

import requests

# 调用Ollama的OpenAI兼容API
response = requests.post(
    "http://localhost:11434/v1/chat/completions",
    json={
        "model": "qwen2.5:7b",
        "messages": [{"role": "user", "content": "什么是5G-A？"}],
        "temperature": 0.7
    }
)
print(response.json()["choices"][0]["message"]["content"])

# 构建自定义模型
ollama create telecom-bot -f Modelfile
# 运行
ollama run telecom-bot

# 安装vLLM
pip install vllm

# 启动API服务（单GPU）
python -m vllm.entrypoints.openai.api_server \
  --model Qwen/Qwen2.5-7B-Instruct \
  --port 8000 \
  --max-model-len 32768

# 多GPU张量并行
python -m vllm.entrypoints.openai.api_server \
  --model Qwen/Qwen2.5-72B-Instruct \
  --tensor-parallel-size 4 \
  --port 8000

# 使用vLLM的Python SDK
from vllm import LLM, SamplingParams

llm = LLM(model="Qwen/Qwen2.5-7B-Instruct")
sampling = SamplingParams(temperature=0.7, max_tokens=500)

prompts = ["解释什么是RAG技术", "用Python写一个快速排序"]
outputs = llm.generate(prompts, sampling)
for output in outputs:
    print(output.outputs[0].text)

import openai

# 使用TokenHub作为网关
client = openai.OpenAI(
    api_key="your-tokenhub-api-key",
    base_url="https://tokenhub.tencentmaas.com/v1"
)

# 调用混元模型
response = client.chat.completions.create(
    model="hy3",  # 腾讯混元Hy3
    messages=[{"role": "user", "content": "解释什么是大模型Agent"}],
    temperature=0.7
)
print(response.choices[0].message.content)

# 切换到DeepSeek模型（只需改model参数）
response = client.chat.completions.create(
    model="deepseek-v3",  # DeepSeek V3
    messages=[{"role": "user", "content": "解释什么是大模型Agent"}],
)
print(response.choices[0].message.content)

# 切换到通义千问
response = client.chat.completions.create(
    model="qwen-max",
    messages=[{"role": "user", "content": "解释什么是大模型Agent"}],
)
print(response.choices[0].message.content)

import anthropic

# 使用Anthropic SDK通过TokenHub调用
client = anthropic.Anthropic(
    api_key="your-tokenhub-api-key",
    base_url="https://tokenhub.tencentmaas.com/anthropic"
)

message = client.messages.create(
    model="hy3",
    max_tokens=1024,
    messages=[{"role": "user", "content": "用Python写一个二分查找"}]
)
print(message.content[0].text)

# 配置环境变量
export ANTHROPIC_API_KEY="your-tokenhub-api-key"
export ANTHROPIC_BASE_URL="https://tokenhub.tencentmaas.com/anthropic"

# 启动Claude Code，使用TokenHub网关的模型
claude --model hy3

# 监控每次调用的Token使用量
response = client.chat.completions.create(
    model="deepseek-v3",
    messages=[{"role": "user", "content": "你好"}]
)

# 查看Token使用统计
usage = response.usage
print(f"输入Token: {usage.prompt_tokens}")
print(f"输出Token: {usage.completion_tokens}")
print(f"总Token: {usage.total_tokens}")

from langchain_openai import OpenAIEmbeddings

# 使用Embedding模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    api_key="your-api-key"
)

# 将文本转为向量
vector = embeddings.embed_query("5G-A技术有什么特点？")
print(f"向量维度: {len(vector)}")  # 通常1536维

# 批量嵌入
texts = ["5G-A介绍", "6G展望", "光纤通信"]
vectors = embeddings.embed_documents(texts)

import chromadb

# 创建Chroma客户端
client = chromadb.PersistentClient(path="./vector_db")

# 创建集合
collection = client.create_collection(
    name="telecom_kb",
    metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

# 添加文档
collection.add(
    documents=[
        "5G-A是5G的增强版，下行峰值速率可达10Gbps",
        "5G-A新增了通感一体能力，可实现低空通信",
        "千兆宽带套餐月费128元，包含500分钟通话"
    ],
    metadatas=[
        {"source": "5ga-brochure.pdf", "page": 1},
        {"source": "5ga-brochure.pdf", "page": 3},
        {"source": "broadband-pricing.xlsx", "page": 1}
    ],
    ids=["doc1", "doc2", "doc3"]
)

# 查询
results = collection.query(
    query_texts=["5G-A的速率是多少"],
    n_results=2
)
print(results["documents"])

from langchain_community.document_loaders import (
    PyPDFLoader, TextLoader, Docx2txtLoader, CSVLoader
)

# 加载PDF
pdf_loader = PyPDFLoader("product_manual.pdf")
pdf_docs = pdf_loader.load()

# 加载Word
docx_loader = Docx2txtLoader("report.docx")
docx_docs = docx_loader.load()

# 加载CSV
csv_loader = CSVLoader("data.csv")
csv_docs = csv_loader.load()

# 合并所有文档
all_docs = pdf_docs + docx_docs + csv_docs
print(f"总文档数: {len(all_docs)}")

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 递归字符分块
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块最大500字符
    chunk_overlap=50,      # 块间重叠50字符（保持上下文）
    separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]  # 中文优先
)

chunks = splitter.split_documents(all_docs)
print(f"分块数: {len(chunks)}")
print(f"第一块内容: {chunks[0].page_content[:100]}")

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 第一步：加载文档
loader = PyPDFLoader("telecom_products.pdf")
docs = loader.load()

# 第二步：文本分块
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50
)
chunks = splitter.split_documents(docs)

# 第三步：向量化并存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 第四步：创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}  # 检索Top-4相关块
)

# 第五步：构造RAG链
prompt_template = """
你是一个电信业务知识助手。请基于以下参考资料回答用户问题。
如果参考资料中没有相关信息，请说"根据现有资料无法回答该问题"。

参考资料：
{context}

用户问题：{question}

回答：
"""
PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# 第六步：问答
result = qa_chain({"query": "5G-A套餐有哪些？"})
print(f"回答: {result['result']}")
print(f"引用来源: {[doc.metadata for doc in result['source_documents']]}")

from langchain.retrievers import MultiQueryRetriever

# 多查询检索器：自动生成多个查询变体
multi_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)
# 用户提问"5G-A快吗" → 自动生成:
# "5G-A的传输速率是多少"
# "5G-A相比5G速度提升多少"
# "5G-A峰值速率"
# 分别检索后合并去重

# 克隆Dify仓库
git clone https://github.com/langgenius/dify.git
cd dify/docker

# 复制环境变量
cp .env.example .env

# 启动Docker Compose
docker compose up -d

# 访问 http://localhost:80

# 在项目目录启动Claude Code
claude

# VibeCoding对话
> 帮我创建一个Express.js API，提供以下接口：
  - GET /api/users 获取用户列表
  - POST /api/users 创建用户
  - GET /api/users/:id 获取用户详情
  - PUT /api/users/:id 更新用户
  - DELETE /api/users/:id 删除用户
  使用SQLite，包含输入验证和错误处理

# Claude Code会自动：
# 1. 规划文件结构
# 2. 创建所有代码文件
# 3. 安装依赖
# 4. 运行测试
# 5. 生成README

pip install langchain langchain-openai langchain-community
pip install chromadb pypdf docx2txt  # 文档处理和向量库

from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0.7,
    api_key="your-api-key"
)

# 基本调用
response = llm.invoke("什么是大模型？")
print(response.content)

# 流式输出
for chunk in llm.stream("讲一个关于AI的笑话"):
    print(chunk.content, end="", flush=True)

from langchain.prompts import ChatPromptTemplate

# 创建提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，请用{style}的风格回答问题。"),
    ("human", "{question}")
])

# 格式化
messages = prompt.invoke({
    "role": "电信技术专家",
    "style": "通俗易懂",
    "question": "什么是5G网络切片？"
})
response = llm.invoke(messages)
print(response.content)

from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

# 定义输出结构
class ProductInfo(BaseModel):
    name: str = Field(description="产品名称")
    price: float = Field(description="月费（元）")
    features: list = Field(description="主要特性")

# 创建解析器
parser = PydanticOutputParser(pydantic_object=ProductInfo)

# 在提示词中使用
prompt = ChatPromptTemplate.from_messages([
    ("system", "根据用户描述提取产品信息。\n{format_instructions}"),
    ("human", "{product_description}")
]).partial(format_instructions=parser.get_format_instructions())

chain = prompt | llm | parser
result = chain.invoke({
    "product_description": "5G尊享套餐，月费199元，包含100GB流量、1000分钟通话、千兆宽带"
})
print(f"产品: {result.name}, 价格: {result.price}, 特性: {result.features}")

# 基础链：Prompt → LLM → Parser
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"question": "什么是RAG？"})

from langchain.chains import LLMChain

# 链1：生成技术概念解释
prompt1 = ChatPromptTemplate.from_template(
    "简要解释什么是{concept}，50字以内。"
)
chain1 = prompt1 | llm | StrOutputParser()

# 链2：基于解释生成示例
prompt2 = ChatPromptTemplate.from_template(
    "基于以下解释，给出一个{concept}的实际应用示例：\n{explanation}"
)
chain2 = prompt2 | llm | StrOutputParser()

# 组合成顺序链
from langchain.chains import SimpleSequentialChain
overall_chain = SimpleSequentialChain(
    chains=[chain1, chain2],
    verbose=True
)
# 注意：SimpleSequentialChain是旧版API，新版推荐用LCEL

# 新版LCEL方式
from langchain_core.runnables import RunnablePassthrough

concept_chain = (
    {"explanation": prompt1 | llm | StrOutputParser(),
     "concept": RunnablePassthrough()}
    | prompt2
    | llm
    | StrOutputParser()
)

result = concept_chain.invoke("网络切片")
print(result)

from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferWindowMemory

# 创建带记忆的对话链
memory = ConversationBufferWindowMemory(k=5)  # 保留最近5轮
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 多轮对话
print(conversation.predict(input="我叫张三，我是电信工程师"))
print(conversation.predict(input="我负责5G基站的维护"))
print(conversation.predict(input="我刚才说我是做什么的？"))  # 能回忆起

from langchain.tools import tool

@tool
def search_telecom_info(query: str) -> str:
    """搜索电信业务信息。输入搜索关键词。"""
    # 实际实现调用搜索API
    return f"搜索结果：关于'{query}'的电信信息..."

@tool
def calculate_plan_cost(monthly_fee: float, months: int) -> str:
    """计算套餐总费用。输入月费和月数。"""
    total = monthly_fee * months
    return f"套餐{monthly_fee}元/月，{months}个月总费用：{total}元"

@tool
def create_trouble_ticket(issue: str, phone: str) -> str:
    """创建故障工单。输入问题描述和联系电话。"""
    return f"工单已创建，问题描述：{issue}，联系电话：{phone}，工单号：T20260723001"

tools = [search_telecom_info, calculate_plan_cost, create_trouble_ticket]

from langchain.agents import create_tool_calling_agent, AgentExecutor

# 创建Agent
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是电信客服助手，使用提供的工具帮助用户解决问题。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行Agent
result = agent_executor.invoke({
    "input": "199元套餐用一年多少钱？另外我家里宽带断了，帮我报修，电话13800138000"
})
print(result["output"])

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# RAG检索工具
@tool
def search_knowledge_base(query: str) -> str:
    """从电信知识库中检索信息。输入问题关键词。"""
    docs = retriever.invoke(query)
    return "\n".join([doc.page_content for doc in docs])

# 组合RAG + 其他工具的Agent
all_tools = [search_knowledge_base, calculate_plan_cost, create_trouble_ticket]
rag_agent = create_tool_calling_agent(llm, all_tools, prompt)

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# 定义工具
@tool
def search_web(query: str) -> str:
    """搜索网络信息"""
    return f"搜索结果：{query}的相关信息..."

@tool
def calculate(expression: str) -> str:
    """数学计算"""
    try:
        return str(eval(expression))
    except:
        return "计算失败"

tools = [search_web, calculate]
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

# 定义节点函数
def agent_node(state: AgentState):
    """Agent决策节点：决定下一步做什么"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState):
    """条件判断：是否调用工具"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode(tools))

# 设置边
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")  # 工具执行后回到Agent

# 编译并运行
app = workflow.compile()

# 调用
result = app.invoke({
    "messages": [("user", "搜索5G-A的速度，然后计算10Gbps等于多少Mbps")]
})
for msg in result["messages"]:
    print(f"{msg.type}: {msg.content}")

from langgraph.graph import StateGraph, END

class TeamState(TypedDict):
    task: str
    research_result: str
    analysis_result: str
    report: str
    next: str

# 研究Agent
def research_agent(state: TeamState):
    """负责信息收集和研究"""
    result = llm.invoke(f"作为研究员，请收集关于'{state['task']}'的信息")
    return {"research_result": result.content, "next": "analyst"}

# 分析Agent
def analysis_agent(state: TeamState):
    """负责数据分析和洞察"""
    result = llm.invoke(
        f"作为分析师，请分析以下研究内容并给出洞察：\n{state['research_result']}"
    )
    return {"analysis_result": result.content, "next": "writer"}

# 报告Agent
def writer_agent(state: TeamState):
    """负责生成最终报告"""
    result = llm.invoke(
        f"作为报告撰写者，基于以下研究和分析，撰写一份报告：\n"
        f"研究：{state['research_result']}\n"
        f"分析：{state['analysis_result']}"
    )
    return {"report": result.content, "next": END}

# 构建多Agent图
workflow = StateGraph(TeamState)
workflow.add_node("researcher", research_agent)
workflow.add_node("analyst", analysis_agent)
workflow.add_node("writer", writer_agent)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "analyst")
workflow.add_edge("analyst", "writer")
workflow.add_edge("writer", END)

team_app = workflow.compile()
result = team_app.invoke({"task": "2026年AI对电信行业的影响"})
print(result["report"])

# 在关键决策点加入人工审核
from langgraph.checkpoint.memory import MemorySaver

def human_review(state: AgentState):
    """人工审核节点 - 暂停等待人工确认"""
    pass  # 实际执行时会暂停

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("review", human_review)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")
workflow.add_edge("agent", "review")  # Agent决策后审核
workflow.add_conditional_edges("review", lambda x: "tools" if x.get("approve") else "agent")
workflow.add_edge("tools", "agent")

# 编译时启用中断
app = workflow.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["review"]  # 在review前暂停
)

from langgraph.checkpoint.sqlite import SqliteSaver

# 使用SQLite持久化状态
config = {"configurable": {"thread_id": "conversation-1"}}
app_with_checkpoint = workflow.compile(
    checkpointer=SqliteSaver.from_conn_string("checkpoints.db")
)

# 第一次执行（可能中断）
result = app_with_checkpoint.invoke(
    {"messages": [("user", "帮我分析网络故障")]},
    config=config
)

# 恢复执行（从断点继续）
result = app_with_checkpoint.invoke(
    {"messages": [("user", "继续")]},
    config=config  # 相同thread_id恢复状态
)

from deepagents import DeepAgent
from langchain_openai import ChatOpenAI

# 创建DeepAgent
agent = DeepAgent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[search_web, calculate, create_trouble_ticket],
    system_prompt="""
    你是电信运维智能助手。你可以：
    1. 查询网络设备状态
    2. 分析告警日志
    3. 创建和管理工单
    4. 生成运维报告
    
    对于复杂任务，请先制定计划，然后逐步执行。
    """,
    max_steps=20,           # 最大执行步数
    enable_memory=True,      # 启用长期记忆
    enable_file_context=True # 启用文件上下文
)

# 执行复杂任务
result = agent.run(
    "检查昨晚所有5G基站的告警，分析根因，为重大告警创建工单，"
    "最后生成运维分析报告"
)
print(result.output)
print(f"执行步骤: {result.steps}")
print(f"使用工具: {result.tools_used}")

# 安装FastMCP（2025-2026年最流行的MCP开发框架）
pip install fastmcp

# FastMCP理念类似FastAPI
# 自动分析Python函数的类型注解和文档字符串
# 编译为大模型可识别的JSON Schema

from fastmcp import FastMCP

# 创建MCP Server实例
mcp = FastMCP("电信运维助手")

# 使用装饰器注册Tool
@mcp.tool()
def query_alarm(device_id: str, hours: int = 24) -> dict:
    """查询指定设备的告警信息
    
    Args:
        device_id: 设备ID，如"SW-001"
        hours: 查询最近多少小时的告警，默认24小时
    
    Returns:
        包含告警列表的字典
    """
    # 模拟查询告警
    alarms = [
        {"time": "2026-07-22 08:30", "level": "critical", "msg": "光功率异常"},
        {"time": "2026-07-22 09:15", "level": "warning", "msg": "CPU使用率>80%"},
    ]
    return {"device_id": device_id, "count": len(alarms), "alarms": alarms}

# 注册Resource
@mcp.resource("config://device/{device_id}")
def get_device_config(device_id: str) -> str:
    """获取设备配置文件"""
    return f"# 设备 {device_id} 配置\ninterface eth0\n  ip address 192.168.1.1/24"

# 注册Prompt模板
@mcp.prompt()
def alarm_analysis_prompt(device_id: str) -> str:
    """生成告警分析提示词"""
    return f"""请分析设备 {device_id} 的告警信息：
1. 查询最近24小时的告警
2. 按严重程度排序
3. 分析可能的根因
4. 给出处理建议"""

# 启动Server
if __name__ == "__main__":
    mcp.run(transport="stdio")  # 本地使用stdio传输

from fastmcp import FastMCP
from pydantic import BaseModel, Field
import asyncio

mcp = FastMCP("进阶运维工具")

# 使用Pydantic模型定义复杂参数
class TroubleTicket(BaseModel):
    """工单创建参数"""
    device_id: str = Field(description="故障设备ID")
    alarm_type: str = Field(description="告警类型：critical/warning/info")
    description: str = Field(description="故障描述")
    priority: int = Field(default=3, ge=1, le=5, description="优先级1-5")

@mcp.tool()
async def create_trouble_ticket(ticket: TroubleTicket) -> dict:
    """创建运维工单
    
    根据告警信息自动创建工单，并通知相关负责人。
    """
    ticket_id = f"TKT-{ticket.device_id}-{hash(ticket.description) % 10000:04d}"
    return {
        "ticket_id": ticket_id,
        "status": "created",
        "priority": ticket.priority,
        "message": f"工单已创建，优先级{ticket.priority}"
    }

@mcp.tool()
async def batch_check_devices(device_ids: list[str]) -> dict:
    """批量检查设备状态
    
    Args:
        device_ids: 设备ID列表
    """
    results = {}
    for did in device_ids:
        await asyncio.sleep(0.1)
        results[did] = {"status": "online", "latency_ms": 12}
    return {"total": len(device_ids), "results": results}

# 远程传输模式（生产环境推荐）
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

{
  "mcpServers": {
    "telecom-ops": {
      "command": "python",
      "args": ["path/to/telecom_mcp_server.py"],
      "env": {
        "DB_HOST": "192.168.1.100",
        "DB_PORT": "5432"
      }
    },
    "web-search": {
      "command": "npx",
      "args": ["-y", "@anthropic/mcp-web-search"]
    }
  }
}

{
  "mcpServers": {
    "telecom-ops": {
      "url": "http://localhost:8080/mcp",
      "transport": "http"
    }
  }
}

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

async def main():
    # 配置MCP Server连接
    server_params = StdioServerParameters(
        command="python",
        args=["telecom_mcp_server.py"],
        env={"DB_HOST": "192.168.1.100"}
    )
    
    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            
            # 列出可用工具
            tools = await session.list_tools()
            print("可用工具:", [t.name for t in tools.tools])
            
            # 调用工具
            result = await session.call_tool(
                "query_alarm",
                arguments={"device_id": "SW-001", "hours": 48}
            )
            print("查询结果:", result.content)
            
            # 读取资源
            resource = await session.read_resource("config://device/SW-001")
            print("设备配置:", resource.contents[0].text)
            
            # 获取提示词模板
            prompt = await session.get_prompt(
                "alarm_analysis_prompt",
                arguments={"device_id": "SW-001"}
            )
            print("分析提示词:", prompt.messages)

asyncio.run(main())

from langchain_mcp_adapters import create_mcp_tools
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent
import asyncio

async def create_agent_with_mcp():
    # 从MCP Server加载工具
    tools = await create_mcp_tools(
        command="python",
        args=["telecom_mcp_server.py"]
    )
    
    # 创建Agent
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    agent = create_tool_calling_agent(llm, tools, 
        system_prompt="你是电信运维助手，可以查询设备告警和管理工单。")
    
    # 执行任务
    result = await agent.ainvoke(
        "查询设备SW-001最近24小时的告警，如果有关键告警请创建工单"
    )
    print(result)
    return result

asyncio.run(create_agent_with_mcp())

---
name: telecom-ops-analyst
version: 1.0.0
description: 电信运维分析技能，支持告警分析、故障诊断和工单管理
author: Telecom Team
tags: [telecom, ops, alarm, troubleshooting]
tools:
  - query_alarm
  - create_trouble_ticket
  - get_device_config
---

# 电信运维分析技能

## 概述
本技能提供电信网络运维分析能力，包括告警查询、故障根因分析、
工单创建和运维报告生成。

## 使用场景
- 设备告警批量分析
- 故障根因定位
- 自动化工单创建
- 运维报告生成

## 工作流程

### 步骤1：告警查询
1. 使用 `query_alarm` 工具查询设备告警
2. 按严重程度排序：critical > warning > info
3. 统计告警数量和类型分布

### 步骤2：根因分析
1. 分析告警关联性
2. 查询设备配置（`get_device_config`）
3. 结合网络拓扑判断故障范围
4. 参考 `references/troubleshooting_guide.md`

### 步骤3：工单创建
- 仅对critical级别告警创建工单
- 工单优先级规则：
  - 核心网设备故障 → 优先级1
  - 接入网设备故障 → 优先级2
  - 一般告警 → 优先级3

### 步骤4：报告生成
- 使用 `scripts/report.py` 生成运维报告
- 报告模板见 `resources/template.xlsx`
- 包含：告警概览、根因分析、处理建议、工单列表

## 约束
- 不修改任何设备配置
- 仅查询和分析，不执行变更操作
- 工单创建前需确认告警真实性

#!/usr/bin/env python3
"""告警分析脚本 - Skills辅助工具"""
import json
from datetime import datetime
from collections import Counter

def analyze_alarm_patterns(alarms: list[dict]) -> dict:
    """分析告警模式"""
    level_dist = Counter(a["level"] for a in alarms)
    hour_dist = Counter(
        datetime.fromisoformat(a["time"]).hour for a in alarms
    )
    type_dist = Counter(a.get("type", "unknown") for a in alarms)
    
    return {
        "total": len(alarms),
        "by_level": dict(level_dist),
        "by_hour": dict(sorted(hour_dist.items())),
        "by_type": dict(type_dist),
        "critical_count": level_dist.get("critical", 0),
        "recommendation": "建议优先处理critical告警"
            if level_dist.get("critical", 0) > 0
            else "当前无关键告警"
    }

def generate_summary(analysis: dict) -> str:
    """生成告警摘要"""
    return f"""
告警分析摘要
============
总告警数: {analysis['total']}
关键告警: {analysis['critical_count']}
告警级别分布: {analysis['by_level']}
建议: {analysis['recommendation']}
"""

if __name__ == "__main__":
    sample_alarms = [
        {"time": "2026-07-22T08:30:00", "level": "critical", "type": "光功率异常"},
        {"time": "2026-07-22T09:15:00", "level": "warning", "type": "CPU高"},
        {"time": "2026-07-22T10:00:00", "level": "warning", "type": "CPU高"},
    ]
    analysis = analyze_alarm_patterns(sample_alarms)
    print(generate_summary(analysis))

# 将Skill放入项目目录
mkdir -p .claude/skills/alarm-analysis
cp -r alarm-analysis-skill/* .claude/skills/alarm-analysis/

# Claude Code会自动发现并加载SKILL.md
# 在对话中直接使用：
# "帮我分析SW-001设备的告警"
# Claude会自动加载alarm-analysis技能，按SKILL.md中定义的流程执行

"""
场景：电信运维数字员工
- Agent：基于Claude Code的决策大脑
- MCP：连接监控系统、工单系统、配置数据库
- Skills：封装运维分析流程和诊断知识
"""

# 1. MCP Server提供工具能力
from fastmcp import FastMCP

mcp = FastMCP("电信运维")

@mcp.tool()
def query_alarms(area: str, level: str = "all") -> dict:
    """查询区域告警"""
    return {"area": area, "alarms": [...]}

@mcp.tool()
def get_topology(device_id: str) -> dict:
    """获取网络拓扑"""
    return {"device": device_id, "upstream": "...", "downstream": "..."}

@mcp.tool()
def create_ticket(device_id: str, desc: str, priority: int) -> dict:
    """创建工单"""
    return {"ticket_id": "TKT-001", "status": "created"}

# 2. Skills定义工作流程（.claude/skills/ops-diagnosis/SKILL.md）
# SKILL.md中定义：
# - 告警分类规则
# - 根因分析步骤
# - 工单创建标准
# - 报告生成模板

# 3. Agent整体协调
# Claude Code加载Skills -> 连接MCP Server -> 按Skills流程执行
# 用户："分析北京区域所有critical告警，创建工单并生成报告"
# Agent自动：
#   1. 加载ops-diagnosis技能
#   2. 调用MCP的query_alarms查询告警
#   3. 按Skills定义的分类规则分析
#   4. 调用MCP的create_ticket创建工单
#   5. 按Skills模板生成报告

from fastmcp import FastMCP
from datetime import datetime
import random

mcp = FastMCP("设备巡检系统")

DEVICES = {
    "SW-001": {"name": "核心交换机-A", "ip": "10.0.0.1", "area": "北京"},
    "SW-002": {"name": "汇聚交换机-B", "ip": "10.0.0.2", "area": "北京"},
    "RT-001": {"name": "核心路由器-A", "ip": "10.0.1.1", "area": "上海"},
}

@mcp.tool()
def list_devices(area: str = "all") -> list[dict]:
    """列出所有设备或指定区域的设备"""
    if area == "all":
        return [{"id": k, **v} for k, v in DEVICES.items()]
    return [{"id": k, **v} for k, v in DEVICES.items() if v["area"] == area]

@mcp.tool()
def inspect_device(device_id: str) -> dict:
    """巡检指定设备，返回设备健康状态"""
    if device_id not in DEVICES:
        return {"error": f"设备{device_id}不存在"}
    return {
        "device_id": device_id,
        "timestamp": datetime.now().isoformat(),
        "cpu_usage": random.uniform(10, 85),
        "memory_usage": random.uniform(20, 90),
        "interfaces": {
            "eth0": {"status": "up", "traffic_in": random.randint(100, 1000)},
            "eth1": {"status": "up", "traffic_in": random.randint(100, 1000)},
            "eth2": {"status": "down", "traffic_in": 0},
        },
        "temperature": random.uniform(25, 55),
    }

@mcp.tool()
def check_connectivity(source: str, target: str) -> dict:
    """检查两台设备之间的连通性"""
    return {
        "source": source,
        "target": target,
        "reachable": random.choice([True, True, False]),
        "latency_ms": random.randint(1, 100),
        "packet_loss": random.uniform(0, 5),
    }

@mcp.tool()
def generate_inspection_report(results: list[dict]) -> str:
    """生成巡检报告（Markdown格式）"""
    report = f"# 设备巡检报告\n\n"
    report += f"巡检时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n"
    report += f"巡检设备数: {len(results)}\n\n"
    report += "| 设备ID | CPU | 内存 | 接口状态 | 温度 | 状态 |\n"
    report += "|--------|-----|------|---------|------|------|\n"
    for r in results:
        cpu = f"{r.get('cpu_usage', 0):.1f}%"
        mem = f"{r.get('memory_usage', 0):.1f}%"
        ifaces = r.get('interfaces', {})
        up_count = sum(1 for v in ifaces.values() if v['status'] == 'up')
        total = len(ifaces)
        temp = f"{r.get('temperature', 0):.1f}C"
        status = "正常" if float(cpu[:-1]) < 80 and float(mem[:-1]) < 85 else "告警"
        report += f"| {r.get('device_id', '?')} | {cpu} | {mem} | {up_count}/{total} | {temp} | {status} |\n"
    return report

if __name__ == "__main__":
    mcp.run(transport="stdio")

---
name: device-inspection
version: 1.0.0
description: 网络设备巡检技能
---

# 网络设备巡检

## 工作流程

### 1. 获取设备列表
- 使用 `list_devices` 获取指定区域所有设备
- 记录设备ID、名称、IP地址

### 2. 逐设备巡检
- 对每台设备调用 `inspect_device`
- 判断标准：
  - CPU > 80% → 告警
  - 内存 > 85% → 告警
  - 温度 > 50°C → 告警
  - 接口down → 告警

### 3. 连通性检查
- 对核心设备之间调用 `check_connectivity`
- 丢包率 > 1% → 告警

### 4. 生成报告
- 使用 `generate_inspection_report` 汇总结果
- 报告包含：巡检概览、异常设备列表、处理建议

# 数字员工：电信运维助手
name: "运维管家小智"
description: "7x24小时电信网络运维智能助手"

system_prompt: |
  你是"运维管家小智"，一位专业的电信网络运维数字员工。
  
  ## 你的职责
  1. 监控网络设备告警，实时响应
  2. 分析告警根因，提供诊断建议
  3. 自动创建和跟踪运维工单
  4. 生成运维分析报告
  5. 协助值班人员进行故障处理
  
  ## 工作规范
  - critical告警：5分钟内响应，立即创建工单并通知
  - warning告警：15分钟内响应，分析原因并记录
  - info告警：汇总分析，定期报告
  - 所有操作需记录日志，确保可追溯
  
  ## 语言风格
  - 专业简洁，使用标准运维术语
  - 重要信息用【】标注
  - 数字数据精确到小数点后1位
  
  ## 安全约束
  - 不执行任何设备配置变更操作
  - 不泄露客户隐私数据
  - 工单创建前需二次确认

tools:
  - query_alarms        # 查询告警
  - analyze_root_cause  # 根因分析
  - create_ticket       # 创建工单
  - send_notification   # 发送通知
  - generate_report     # 生成报告
  - search_kb           # 知识库检索

knowledge_base:
  - "运维知识库"        # 故障处理手册、设备文档
  - "告警代码库"        # 告警代码对照表

# Dify自定义工具配置示例（通过API扩展）

# 工具1：查询告警
{
    "name": "query_alarms",
    "description": "查询指定区域和级别的网络告警",
    "parameters": {
        "type": "object",
        "properties": {
            "area": {"type": "string", "description": "区域：北京/上海/广州"},
            "level": {"type": "string", "enum": ["critical", "warning", "info", "all"]}
        },
        "required": ["area"]
    },
    "api": {
        "url": "http://monitor-api.internal/alarms",
        "method": "GET",
        "headers": {"Authorization": "Bearer ${API_KEY}"}
    }
}

# 工具2：创建工单
{
    "name": "create_ticket",
    "description": "创建运维工单",
    "parameters": {
        "type": "object",
        "properties": {
            "device_id": {"type": "string"},
            "description": {"type": "string"},
            "priority": {"type": "integer", "minimum": 1, "maximum": 5}
        },
        "required": ["device_id", "description"]
    }
}

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import Tool
from langchain.prompts import ChatPromptTemplate

# 1. 定义工具
def query_alarms(area: str, level: str = "all") -> str:
    """查询指定区域的网络告警"""
    # 模拟查询
    return f"区域{area}共有{level}级别告警3条：\n1. [critical] SW-001 光功率异常\n2. [warning] SW-002 CPU>80%\n3. [warning] RT-001 温度偏高"

def create_ticket(device_id: str, description: str, priority: int = 3) -> str:
    """创建运维工单"""
    return f"工单已创建：TKT-{device_id}-{priority}，描述：{description}，优先级：{priority}"

def search_kb(query: str) -> str:
    """检索运维知识库"""
    return f"知识库匹配结果：光功率异常通常由光纤连接松动或光模块故障引起，建议：1.检查光纤接头 2.更换光模块 3.联系线路维护"

def send_notification(channel: str, message: str) -> str:
    """发送通知"""
    return f"通知已发送至{channel}：{message}"

tools = [
    Tool(name="query_alarms", func=lambda **kw: query_alarms(**kw), description="查询网络告警"),
    Tool(name="create_ticket", func=lambda **kw: create_ticket(**kw), description="创建运维工单"),
    Tool(name="search_kb", func=lambda **kw: search_kb(**kw), description="检索知识库"),
    Tool(name="send_notification", func=lambda **kw: send_notification(**kw), description="发送通知"),
]

# 2. 创建Agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是"运维管家小智"，电信网络运维数字员工。
    
    工作规范：
    - critical告警：立即创建优先级1的工单并通知值班人员
    - 工单创建前查阅知识库获取处理建议
    - 所有操作简洁专业
    
    可用工具：{tool_names}
    {tools}"""),
    ("human", "{input}"),
    ("assistant", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=10)

# 3. 执行任务
result = agent_executor.invoke({
    "input": "查询北京区域所有告警，critical级别的创建工单并通知值班人员"
})
print(result["output"])

from langchain.memory import ConversationSummaryMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 对话记忆（摘要式）
memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="chat_history",
    return_messages=True
)

# 长期记忆（向量检索）
vectorstore = Chroma(
    collection_name="employee_memory",
    embedding_function=OpenAIEmbeddings()
)

# 存储经验
def save_experience(task: str, solution: str, outcome: str):
    """保存处理经验到长期记忆"""
    vectorstore.add_texts([
        f"任务: {task}\n方案: {solution}\n结果: {outcome}"
    ], metadatas=[{"type": "experience", "date": "2026-07-22"}])

# 检索经验
def recall_experience(query: str, k: int = 3) -> str:
    """从长期记忆中检索相关经验"""
    docs = vectorstore.similarity_search(query, k=k)
    return "\n---\n".join(d.page_content for d in docs)

# 将记忆工具加入Agent
memory_tools = tools + [
    Tool(name="recall_experience", func=recall_experience, 
         description="检索过往处理经验"),
    Tool(name="save_experience", func=lambda **kw: (save_experience(**kw), "经验已保存")[1],
         description="保存处理经验")
]

from functools import wraps
import logging

# 操作审计日志
audit_logger = logging.getLogger("digital_employee_audit")
audit_logger.setLevel(logging.INFO)
handler = logging.FileHandler("audit.log", encoding="utf-8")
audit_logger.addHandler(handler)

def audit_log(action: str):
    """操作审计装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            audit_logger.info(f"[{action}] 参数: {kwargs}")
            result = func(*args, **kwargs)
            audit_logger.info(f"[{action}] 结果: {result[:100]}")
            return result
        return wrapper
    return decorator

# 权限控制
class PermissionManager:
    def __init__(self):
        self.permissions = {
            "viewer": ["query_alarms", "search_kb", "recall_experience"],
            "operator": ["query_alarms", "search_kb", "create_ticket", "send_notification", "recall_experience"],
            "admin": ["*"]  # 全部权限
        }
    
    def check_permission(self, role: str, action: str) -> bool:
        allowed = self.permissions.get(role, [])
        return "*" in allowed or action in allowed

# 敏感数据脱敏
import re

def mask_sensitive(text: str) -> str:
    """脱敏处理"""
    text = re.sub(r'\b\d{11}\b', '***-****-****', text)  # 手机号
    text = re.sub(r'\b\d{18}\b', '******************', text)  # 身份证
    text = re.sub(r'\b\d{16,19}\b', '****-****-****-****', text)  # 银行卡
    return text

# 应用安全包装
permission_mgr = PermissionManager()

@audit_log("create_ticket")
def secure_create_ticket(device_id: str, description: str, priority: int = 3, 
                          operator_role: str = "operator") -> str:
    if not permission_mgr.check_permission(operator_role, "create_ticket"):
        return "权限不足：当前角色无法创建工单"
    description = mask_sensitive(description)
    return create_ticket(device_id, description, priority)

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import Tool
from langchain.prompts import ChatPromptTemplate
from datetime import datetime, timedelta
import json

# ===== 数据层（模拟） =====
MOCK_DATA = {
    "alarms": [
        {"time": "2026-07-22 08:30", "device": "SW-001", "area": "北京", 
         "level": "critical", "type": "光功率异常", "status": "已处理"},
        {"time": "2026-07-22 09:15", "device": "SW-002", "area": "北京", 
         "level": "warning", "type": "CPU高", "status": "处理中"},
        {"time": "2026-07-22 10:00", "device": "RT-001", "area": "上海", 
         "level": "critical", "type": "链路中断", "status": "已处理"},
        {"time": "2026-07-22 14:30", "device": "SW-003", "area": "北京", 
         "level": "warning", "type": "内存高", "status": "已处理"},
        {"time": "2026-07-22 16:00", "device": "RT-002", "area": "广州", 
         "level": "info", "type": "配置变更", "status": "已记录"},
    ]
}

# ===== 工具层 =====
def get_alarm_stats(area: str = "all", days: int = 1) -> str:
    """获取告警统计数据"""
    alarms = [a for a in MOCK_DATA["alarms"] if area == "all" or a["area"] == area]
    stats = {
        "total": len(alarms),
        "critical": sum(1 for a in alarms if a["level"] == "critical"),
        "warning": sum(1 for a in alarms if a["level"] == "warning"),
        "info": sum(1 for a in alarms if a["level"] == "info"),
        "resolved": sum(1 for a in alarms if a["status"] == "已处理"),
        "pending": sum(1 for a in alarms if a["status"] != "已处理"),
    }
    return json.dumps(stats, ensure_ascii=False)

def get_top_issues(area: str = "all", top_n: int = 5) -> str:
    """获取TOP问题列表"""
    alarms = [a for a in MOCK_DATA["alarms"] if area == "all" or a["area"] == area]
    from collections import Counter
    type_counter = Counter(a["type"] for a in alarms)
    top = type_counter.most_common(top_n)
    return json.dumps([{"type": t, "count": c} for t, c in top], ensure_ascii=False)

def get_device_list(area: str = "all") -> str:
    """获取设备列表"""
    devices = list(set(a["device"] for a in MOCK_DATA["alarms"]
                       if area == "all" or a["area"] == area))
    return json.dumps(devices, ensure_ascii=False)

# ===== Agent层 =====
llm = ChatOpenAI(model="gpt-4o", temperature=0)

tools = [
    Tool(name="get_alarm_stats", func=lambda **kw: get_alarm_stats(**kw),
         description="获取告警统计数据，参数：area(区域), days(天数)"),
    Tool(name="get_top_issues", func=lambda **kw: get_top_issues(**kw),
         description="获取TOP问题列表，参数：area(区域), top_n(数量)"),
    Tool(name="get_device_list", func=lambda **kw: get_device_list(**kw),
         description="获取设备列表，参数：area(区域)"),
]

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是运维报告生成助手。请根据查询到的数据，生成规范的运维报告。

报告格式：
# 运维分析报告
## 一、告警概览
（统计表格）
## 二、TOP问题分析
（问题列表和分析）
## 三、处理情况
（已处理/未处理统计）
## 四、改进建议
（针对性建议）

要求：数据准确，分析有深度，建议可操作。使用Markdown格式。"),
    ("human", "{input}"),
    ("assistant", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=10)

# ===== 执行 =====
result = executor.invoke({
    "input": "生成北京区域今天的运维分析报告"
})
print(result["output"])

# 通过FastAPI部署为Web服务
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="运维数字员工API")

class ReportRequest(BaseModel):
    area: str = "all"
    days: int = 1
    template: str = "standard"

@app.post("/api/report/generate")
async def generate_report(req: ReportRequest):
    result = executor.invoke({
        "input": f"生成{req.area}区域最近{req.days}天的运维报告"
    })
    return {"status": "success", "report": result["output"]}

@app.post("/api/alarm/query")
async def query_alarm_api(area: str = "all", level: str = "all"):
    stats = get_alarm_stats(area=area)
    return {"status": "success", "data": json.loads(stats)}

# 企微/钉钉webhook对接
@app.post("/api/webhook/wechat")
async def wechat_webhook(message: str):
    """企业微信消息回调"""
    result = executor.invoke({"input": message})
    return {"msgtype": "text", "text": {"content": result["output"]}}

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
import operator

class OpsState(TypedDict):
    """运维多Agent系统共享状态"""
    # 输入
    alarm: dict                     # 原始告警信息
    alarm_type: str                 # 告警分类
    
    # 诊断结果
    diag_link: str                  # 传输诊断结果
    diag_device: str                # 设备诊断结果
    
    # 知识库匹配
    similar_cases: str              # 历史相似案例
    recommended_fix: str            # 推荐修复方案
    
    # 决策
    action_plan: str                # 执行方案
    risk_level: str                 # 风险等级
    need_human: bool                # 是否需要人工
    
    # 执行
    execution_result: str           # 执行结果
    
    # 报告
    final_report: str               # 最终报告

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# ===== 1. 监控Agent =====
def monitor_agent(state: OpsState) -> dict:
    """接收告警，进行分类和去重"""
    alarm = state["alarm"]
    
    prompt = ChatPromptTemplate.from_template("""
    你是网络监控Agent。请对以下告警进行分类：
    
    告警信息：{alarm}
    
    分类标准：
    - "link": 传输链路类（光功率、链路中断、带宽异常等）
    - "device": 设备硬件类（CPU、内存、温度、硬件故障等）
    - "security": 安全类（异常访问、攻击等）
    - "other": 其他
    
    只返回分类结果（link/device/security/other）。
    """)
    
    result = (prompt | llm).invoke({"alarm": str(alarm)})
    return {"alarm_type": result.content.strip()}

# ===== 2. 传输诊断Agent =====
def link_diagnostic_agent(state: OpsState) -> dict:
    """传输层故障诊断"""
    alarm = state["alarm"]
    
    prompt = ChatPromptTemplate.from_template("""
    你是传输网络诊断专家。请分析以下告警：
    
    告警：{alarm}
    
    诊断步骤：
    1. 分析告警特征
    2. 判断可能的故障点（光路、光模块、链路质量）
    3. 给出初步诊断结论
    
    请输出诊断报告（200字内）。
    """)
    
    result = (prompt | llm).invoke({"alarm": str(alarm)})
    return {"diag_link": result.content}

# ===== 3. 设备诊断Agent =====
def device_diagnostic_agent(state: OpsState) -> dict:
    """设备层故障诊断"""
    alarm = state["alarm"]
    
    prompt = ChatPromptTemplate.from_template("""
    你是设备硬件诊断专家。请分析以下告警：
    
    告警：{alarm}
    
    诊断步骤：
    1. 分析设备类型和告警指标
    2. 判断硬件健康状态
    3. 检查性能瓶颈
    4. 给出诊断结论
    
    请输出诊断报告（200字内）。
    """)
    
    result = (prompt | llm).invoke({"alarm": str(alarm)})
    return {"diag_device": result.content}

# ===== 4. 知识库Agent =====
def knowledge_agent(state: OpsState) -> dict:
    """历史案例检索和方案推荐"""
    alarm = state["alarm"]
    diag = state.get("diag_link", "") + "\n" + state.get("diag_device", "")
    
    prompt = ChatPromptTemplate.from_template("""
    你是运维知识库Agent。基于以下信息检索匹配的处理方案：
    
    告警：{alarm}
    诊断结果：{diag}
    
    请模拟检索知识库，提供：
    1. 最可能的2个历史案例
    2. 推荐的处理方案
    3. 预计恢复时间
    """)
    
    result = (prompt | llm).invoke({"alarm": str(alarm), "diag": diag})
    return {
        "similar_cases": result.content,
        "recommended_fix": result.content
    }

# ===== 5. 决策Agent =====
def decision_agent(state: OpsState) -> dict:
    """方案选择与风险评估"""
    prompt = ChatPromptTemplate.from_template("""
    你是运维决策Agent。基于以下信息制定执行方案：
    
    诊断结果-传输：{diag_link}
    诊断结果-设备：{diag_device}
    推荐方案：{recommended_fix}
    
    请制定执行方案：
    1. 选择最优处理方案
    2. 评估风险（low/medium/high）
    3. 是否需要人工介入（true/false）
    4. 具体执行步骤
    
    格式：
    风险等级：[low/medium/high]
    需要人工：[true/false]
    执行方案：...
    """)
    
    result = (prompt | llm).invoke({
        "diag_link": state.get("diag_link", ""),
        "diag_device": state.get("diag_device", ""),
        "recommended_fix": state.get("recommended_fix", "")
    })
    
    content = result.content
    risk = "low"
    need_human = False
    if "high" in content.lower():
        risk = "high"
        need_human = True
    elif "medium" in content.lower():
        risk = "medium"
    
    return {
        "action_plan": content,
        "risk_level": risk,
        "need_human": need_human
    }

# ===== 6. 执行Agent =====
def execution_agent(state: OpsState) -> dict:
    """执行处理方案"""
    plan = state["action_plan"]
    need_human = state.get("need_human", False)
    
    if need_human:
        return {"execution_result": "已通知值班人员，等待人工处理。方案已发送至企微群。"}
    
    prompt = ChatPromptTemplate.from_template("""
    你是执行Agent。请执行以下方案（模拟执行）：
    
    方案：{plan}
    
    请模拟执行并返回执行结果，包括：
    1. 创建的工单号
    2. 发送的通知
    3. 执行的操作
    4. 执行结果（成功/失败）
    """)
    
    result = (prompt | llm).invoke({"plan": plan})
    return {"execution_result": result.content}

# ===== 7. 报告Agent =====
def report_agent(state: OpsState) -> dict:
    """生成运维报告"""
    prompt = ChatPromptTemplate.from_template("""
    你是报告Agent。请基于以下信息生成运维处理报告：
    
    原始告警：{alarm}
    诊断结果-传输：{diag_link}
    诊断结果-设备：{diag_device}
    历史案例：{similar_cases}
    执行方案：{action_plan}
    执行结果：{execution_result}
    
    生成Markdown格式报告，包含：
    # 故障处理报告
    ## 一、故障概述
    ## 二、诊断过程
    ## 三、处理方案
    ## 四、执行结果
    ## 五、经验总结
    """)
    
    result = (prompt | llm).invoke({
        "alarm": str(state["alarm"]),
        "diag_link": state.get("diag_link", ""),
        "diag_device": state.get("diag_device", ""),
        "similar_cases": state.get("similar_cases", ""),
        "action_plan": state.get("action_plan", ""),
        "execution_result": state.get("execution_result", "")
    })
    return {"final_report": result.content}