# KV Cache显存占用计算（以7B模型为例）
# 参数: hidden_size=4096, num_layers=32, num_heads=32

# 每个token的KV Cache大小
kv_cache_per_token = 2 * num_layers * hidden_size * dtype_size
# FP16: 2 * 32 * 4096 * 2 bytes = 524,288 bytes ≈ 0.5MB

# 生成4096个token的KV Cache总量
total_kv_cache = kv_cache_per_token * 4096
# ≈ 2GB （仅KV Cache就占2GB显存！）

# 优化：MQA/GQA减少KV头数
# GQA: num_kv_heads=8 (原32头分8组)
kv_cache_gqa = 2 * 32 * 4096 * (8/32) * 2 * 4096  # 节省75%

# Flash Attention在Transformers中的使用
from transformers import AutoModelForCausalLM
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3-8B",
    torch_dtype="auto",
    attn_implementation="flash_attention_2"  # 启用Flash Attention 2
)

# GGUF量化等级
| 量化等级 | 位宽 | 模型大小(7B) | 精度损失 | 适用硬件 |
|---------|------|-------------|---------|---------|
| F16 | 16bit | 13GB | 无 | GPU |
| Q8_0 | 8bit | 7GB | 极小 | GPU/CPU |
| Q4_K_M | 4bit | 4GB | 小 | CPU/GPU混合 |
| Q3_K_S | 3bit | 3GB | 中等 | CPU |

# vLLM部署示例
pip install vllm
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3-8B-Instruct \
    --port 8000 \
    --gpu-memory-utilization 0.9 \
    --max-model-len 8192

# RESTful API封装（FastAPI + vLLM）
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    model: str
    messages: list[dict]
    temperature: float = 0.7
    max_tokens: int = 1024
    stream: bool = False

@app.post("/v1/chat/completions")
async def chat(req: ChatRequest):
    # 调用vLLM推理引擎
    result = llm_engine.generate(req.messages, req.max_tokens)
    return {"choices": [{"message": {"content": result}}]}

# Hermes 2 Pro Function Calling示例
import json

# 定义可用工具（函数）
tools = [
    {
        "type": "function",
        "function": {
            "name": "query_user_info",
            "description": "查询用户套餐信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "phone": {"type": "string", "description": "手机号"}
                },
                "required": ["phone"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "recommend_package",
            "description": "根据用户使用情况推荐套餐",
            "parameters": {
                "type": "object",
                "properties": {
                    "monthly_data": {"type": "number", "description": "月均流量使用(GB)"},
                    "monthly_calls": {"type": "integer", "description": "月均通话(分钟)"},
                    "budget": {"type": "number", "description": "月预算(元)"}
                },
                "required": ["monthly_data", "budget"]
            }
        }
    }
]

# 用户消息
messages = [
    {"role": "system", "content": "你是电信客服助手，可以查询用户信息和推荐套餐。"},
    {"role": "user", "content": "我的手机号是13800138000，帮我看看用什么套餐合适，我每月用大概30G流量，预算100元"}
]

# 调用Hermes模型（兼容OpenAI API格式）
response = client.chat.completions.create(
    model="NousResearch/Hermes-2-Pro-Llama-3-8B",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

# 模型返回工具调用决策
tool_call = response.choices[0].message.tool_calls[0]
print(f"调用函数: {tool_call.function.name}")
print(f"参数: {json.loads(tool_call.function.arguments)}"
)
# 输出: 调用函数: recommend_package
#       参数: {"monthly_data": 30.0, "budget": 100.0}

"""
套餐推荐Demo
用户输入需求 → Hermes模型决策 → 调用推荐函数 → 返回结果
"""

def recommend_package(monthly_data: float, monthly_calls: int = 0, budget: float = 100):
    """根据用户需求推荐套餐"""
    packages = [
        {"name": "5G尊享129", "price": 129, "data": 50, "calls": 1000, "features": "适合大流量用户"},
        {"name": "5G畅享99", "price": 99, "data": 30, "calls": 500, "features": "性价比之选"},
        {"name": "5G活力59", "price": 59, "data": 15, "calls": 200, "features": "经济实惠"},
    ]
    
    # 筛选符合预算且满足流量需求的套餐
    suitable = [
        p for p in packages 
        if p["price"] <= budget * 1.2 and p["data"] >= monthly_data
    ]
    
    if not suitable:
        return {"error": "未找到匹配套餐", "suggestion": "建议适当增加预算"}
    
    # 按价格排序取最优
    best = min(suitable, key=lambda x: x["price"])
    return {
        "recommended": best["name"],
        "price": f"{best['price']}元/月",
        "data": f"{best['data']}GB",
        "calls": f"{best['calls']}分钟",
        "features": best["features"],
        "reason": f"满足{monthly_data}GB流量需求，价格{best['price']}元在预算{budget}元范围内"
    }

# 完整调用链
def run_demo():
    user_input = "我每月用30G流量，通话500分钟，预算100元"
    
    # Step 1: Hermes决策调用哪个函数
    # Step 2: 执行函数
    result = recommend_package(monthly_data=30, monthly_calls=500, budget=100)
    
    # Step 3: 生成交互回复
    print(f"为您推荐【{result['recommended']}】套餐")
    print(f"月费{result['price']}，含{result['data']}流量{result['calls']}")
    print(f"推荐理由：{result['reason']}")

run_demo()

from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline

# 1. from_pretrained: 加载模型和分词器
model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen2.5-7B-Instruct",
    torch_dtype="auto",
    device_map="auto"
)
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B-Instruct")

# 2. pipeline: 快速推理
pipe = pipeline("text-generation", model=model, tokenizer=tokenizer)
result = pipe("什么是5G网络切片？", max_new_tokens=200)
print(result[0]["generated_text"])

# 3. tokenizer: 文本编码/解码
inputs = tokenizer("大模型推理优化", return_tensors="pt")
tokens = tokenizer.convert_ids_to_tokens(inputs["input_ids"][0])
print(f"Token化: {tokens}")
print(f"Token数: {len(tokens)}")

# 默认缓存路径: ~/.cache/huggingface/hub
# 可自定义cache_dir
model = AutoModel.from_pretrained("Qwen/Qwen2.5-7B", cache_dir="./models")

# ModelScope下载
from modelscope import snapshot_download
model_dir = snapshot_download("Qwen/Qwen2.5-7B-Instruct", revision="master")
print(f"模型下载到: {model_dir}")

# 方式1: 环境变量配置hf-mirror镜像
export HF_ENDPOINT=https://hf-mirror.com
# Windows PowerShell:
$env:HF_ENDPOINT="https://hf-mirror.com"

# 方式2: 使用ModelScope（国内速度快）
pip install modelscope
# ModelScope是阿里达摩院开源的模型社区，国内访问速度快

# 方式3: 配置.gitconfig加速
# 对于git clone方式下载模型
git config --global url."https://hf-mirror.com".insteadOf "https://huggingface.co"

# 安装Ollama（Linux）
curl -fsSL https://ollama.com/install.sh | sh

# Windows: 从ollama.com下载安装包

# 基本命令
ollama pull qwen2.5:7b         # 下载模型
ollama run qwen2.5:7b          # 运行模型
ollama list                    # 查看已安装模型
ollama stop qwen2.5:7b         # 停止模型
ollama rm qwen2.5:7b           # 删除模型

# 1. 下载GGUF格式模型
# 从HuggingFace或ModelScope下载 .gguf 文件

# 2. 创建Modelfile指定GGUF路径
echo 'FROM ./qwen2.5-7b-instruct-q4_k_m.gguf' > Modelfile

# 3. 导入
ollama create my-qwen -f Modelfile

# 4. 运行
ollama run my-qwen

import requests

# Ollama默认提供OpenAI兼容API（端口11434）
url = "http://localhost:11434/v1/chat/completions"
headers = {"Content-Type": "application/json"}
data = {
    "model": "qwen2.5:7b",
    "messages": [
        {"role": "system", "content": "你是电信助手"},
        {"role": "user", "content": "什么是5G网络切片？"}
    ],
    "stream": False  # 非流式
}

response = requests.post(url, json=data, headers=headers)
print(response.json()["choices"][0]["message"]["content"])

# 流式调用
data["stream"] = True
response = requests.post(url, json=data, headers=headers, stream=True)
for line in response.iter_lines():
    if line:
        import json
        chunk = json.loads(line)
        if chunk["choices"][0].get("delta", {}).get("content"):
            print(chunk["choices"][0]["delta"]["content"], end="")

import requests
import json

# 通用OpenAI兼容API调用
def call_llm(messages, model="gpt-4o", stream=False):
    """大模型API通用调用函数"""
    url = "https://api.openai.com/v1/chat/completions"
    headers = {
        "Authorization": "Bearer sk-your-api-key",
        "Content-Type": "application/json"
    }
    payload = {
        "model": model,
        "messages": messages,
        "temperature": 0.7,
        "max_tokens": 2048,
        "stream": stream
    }
    
    if stream:
        response = requests.post(url, json=payload, headers=headers, stream=True)
        for line in response.iter_lines():
            if line and line.startswith(b"data: "):
                data = json.loads(line[6:])
                if data != "[DONE]":
                    yield data["choices"][0]["delta"].get("content", "")
    else:
        response = requests.post(url, json=payload, headers=headers)
        return response.json()["choices"][0]["message"]["content"]

import time
from requests.exceptions import RequestException

def call_with_retry(messages, max_retries=3, timeout=30):
    """带重试和超时的API调用"""
    for attempt in range(max_retries):
        try:
            response = requests.post(
                url, json=payload, headers=headers,
                timeout=timeout
            )
            
            # Rate Limit处理
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"触发限流，等待{retry_after}秒后重试")
                time.sleep(retry_after)
                continue
            
            response.raise_for_status()
            return response.json()
            
        except RequestException as e:
            print(f"第{attempt+1}次请求失败: {e}")
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 指数退避
                time.sleep(wait)
    
    raise Exception(f"请求失败，已重试{max_retries}次")

import asyncio
import aiohttp

async def async_call(session, message):
    """异步单次调用"""
    async with session.post(url, json={"model": "gpt-4o", "messages": message}) as resp:
        return await resp.json()

async def batch_call(messages_list):
    """批量异步调用"""
    async with aiohttp.ClientSession() as session:
        tasks = [async_call(session, msg) for msg in messages_list]
        results = await asyncio.gather(*tasks)
        return results

# 批量调用示例
messages_batch = [
    [{"role": "user", "content": f"分析第{i}个告警: ..."}]
    for i in range(10)
]
results = asyncio.run(batch_call(messages_batch))

from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatOllama

# OpenAI
llm_openai = ChatOpenAI(model="gpt-4o", api_key="sk-xxx")

# 本地Ollama
llm_ollama = ChatOllama(model="qwen2.5:7b", base_url="http://localhost:11434")

# 统一调用接口
llm = llm_openai  # 或 llm_ollama，切换无需改业务代码
response = llm.invoke("什么是5G网络切片？")
print(response.content)

# 零样本（Zero-shot）
prompt_zeroshot = """
请将以下工单分类（宽带/话费/增值业务/其他）：
工单：用户反馈手机无法上网，显示无信号。
"""

# 少样本（Few-shot）
prompt_fewshot = """
请将工单分类。
示例：
工单：用户说宽带频繁掉线 → 宽带
工单：用户质疑上月话费偏高 → 话费
工单：用户想取消增值服务 → 增值业务

工单：用户反馈手机无法上网，显示无信号 → 
"""

# 思维链（CoT）
prompt_cot = """
请分析工单并分类，请先推理再给出答案。
工单：用户反馈手机无法上网，显示无信号。

思考过程：
1. 症状是"无法上网"+"无信号"
2. 手机无信号通常与网络覆盖或SIM卡有关
3. 无法上网是无信号的直接结果
4. 这属于网络接入问题

分类：宽带/网络接入
"""

SYSTEM_PROMPT = """你是电信智能客服助手。

【可做事项】
- 查询用户套餐、余额、流量
- 回答电信业务咨询
- 协助处理简单故障

【禁止事项】
- 不提供非电信相关信息
- 不进行任何涉及付款的操作
- 不泄露其他用户信息
- 不确定时回答"抱歉，我无法确认，请咨询人工客服"

【输出要求】
- 简洁准确，使用中文
- 复杂问题分步骤回答
- 涉及操作时给出明确指引"""

prompt_meeting = """
请根据以下会议记录生成结构化纪要：

{transcript}

输出格式：
## 会议信息
- 时间/地点/参会人
## 会议要点
1. ...
## 决议事项
- ...
## 待办任务
| 任务 | 负责人 | 截止日期 |
"""

prompt_sql = """
你是SQL专家。根据用户问题生成PostgreSQL查询语句。

表结构：
- call_records(id, phone, call_time, duration, type, cost)
- user_packages(id, phone, package_name, monthly_fee, data_gb)

用户问题：查询2026年7月话费超过100元的用户

请只输出SQL语句，不要解释。
"""

prompt_appeasement = """
用户投诉内容：{complaint}
用户情绪：{emotion_level}（1-5，5最激动）

要求：
1. 先共情回应用户情绪
2. 表示理解并道歉
3. 给出初步处理方案
4. 承诺跟进时间

语气：温和、专业、有同理心
"""

# TokenHub统一API调用示例
import requests

# 无论调用哪个模型，API格式统一
url = "https://tokenhub.ctyun.cn/v1/chat/completions"
headers = {
    "Authorization": "Bearer th-your-token",
    "Content-Type": "application/json"
}

# 调用DeepSeek
payload_deepseek = {
    "model": "deepseek-v3",
    "messages": [{"role": "user", "content": "分析网络告警"}]
}

# 调用豆包
payload_doubao = {
    "model": "doubao-pro",
    "messages": [{"role": "user", "content": "分析网络告警"}]
}

# 调用通义千问
payload_qwen = {
    "model": "qwen-max",
    "messages": [{"role": "user", "content": "分析网络告警"}]
}

# 统一调用方式
response = requests.post(url, json=payload_deepseek, headers=headers)

# HNSW关键参数
| 参数 | 说明 | 取值建议 |
|------|------|---------|
| ef_construction | 建图时搜索宽度 | 200-400（越大越精确） |
| M | 每个节点的最大连接数 | 16-48 |
| ef_search | 查询时搜索宽度 | 50-200（越大越精确） |

# IVF关键参数
| 参数 | 说明 | 取值建议 |
|------|------|---------|
| nlist | 聚类中心数 | sqrt(N)~4*sqrt(N) |
| nprobe | 查询时搜索的簇数 | 8-32（越大越精确） |

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection(
    name="telecom_kb",
    metadata={"hnsw:space": "cosine"}
)

# Create
collection.add(
    documents=["宽带无法上网的排查步骤：1.检查光猫指示灯",
               "5G套餐变更流程：发送短信至10001"],
    metadatas=[{"category": "宽带故障"}, {"category": "套餐业务"}],
    ids=["doc1", "doc2"]
)

# Read（带元数据过滤）
results = collection.query(
    query_texts=["宽带连不上"],
    n_results=3,
    where={"category": "宽带故障"}  # 元数据过滤
)

# Update
collection.update(
    ids=["doc1"],
    documents=["更新后的排查步骤..."]
)

# Delete
collection.delete(ids=["doc2"])

# PDF解析（含表格）
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("资费手册.pdf")
pages = loader.load()

# Word文档
from langchain_community.document_loaders import Docx2txtLoader
loader = Docx2txtLoader("运维手册.docx")

# Markdown
from langchain_community.document_loaders import UnstructuredMarkdownLoader

# HTML
from langchain_community.document_loaders import WebBaseLoader

import re

def clean_text(text):
    # 去除多余空白
    text = re.sub(r'\s+', ' ', text)
    # 去除特殊字符
    text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', '', text)
    # 标准化标点
    text = text.replace('，', ',').replace('。', '.')
    return text.strip()

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    MarkdownHeaderTextSplitter,
    CharacterTextSplitter
)

# 1. 固定长度切分
splitter = CharacterTextSplitter(
    chunk_size=500,      # 每块500字符
    chunk_overlap=50,    # 重叠50字符（避免信息断裂）
    separator="\n\n"
)

# 2. 递归字符切分（推荐）
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "，", " ", ""]  # 优先按段落→句号→逗号
)

# 3. 按Markdown标题结构切分
headers_to_split = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]
splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split)

# 表格切分需要特殊处理，不能按行分割
# 方案1：整表作为一个chunk（如果不太大）
# 方案2：每行转为"套餐名|月费|流量|通话"的自然语言描述

def table_to_text(table):
    """将表格转为文本，保持列关系"""
    rows = []
    for row in table.rows[1:]:  # 跳过表头
        cells = [cell.text for cell in row.cells]
        text = f"套餐:{cells[0]}, 月费:{cells[1]}元, 流量:{cells[2]}GB, 通话:{cells[3]}分钟"
        rows.append(text)
    return "\n".join(rows)

from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma

# 向量检索（语义匹配）
vector_results = vectorstore.similarity_search(query, k=10)

# BM25检索（关键词匹配）
bm25_retriever = BM25Retriever.from_documents(docs)
bm25_results = bm25_retriever.invoke(query)

# 混合：合并去重，取并集
all_results = list(set(vector_results + bm25_results))

from langchain_cohere import CohereRerank

# 召回后重排
retriever = vectorstore.as_retriever(search_kwargs={"k": 20})  # 先召回20条
reranker = CohereRerank(top_n=5)  # 重排后取TOP5

# 重排流程：召回20条 → Reranker打分 → 取分数最高的5条

from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

prompt_template = """
基于以下已知信息回答用户问题。

已知信息：
{context}

问题：{question}

要求：
1. 只基于已知信息回答，不要编造
2. 如果已知信息中没有答案，请回答"根据知识库，暂无相关信息"
3. 回答时标注信息来源

回答：
"""

prompt = PromptTemplate(template=prompt_template, input_variables=["context", "question"])

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.invoke({"query": "宽带无法上网怎么排查？"})
print(result["result"])
for doc in result["source_documents"]:
    print(f"来源: {doc.metadata.get('source', 'N/A')}")

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

# 2. 复制环境变量模板
cp .env.example .env

# 3. 修改.env关键配置
# EXPOSE_PORT=8080         # Dify访问端口
# 修改数据库密码等

# 4. 启动（含PostgreSQL、Redis、Weaviate）
docker-compose up -d

# 5. 访问 http://localhost:8080
# 首次访问需创建管理员账号

# 桌面端：从claude.ai/download下载安装

# VSCode插件：
# 1. 打开VSCode扩展商店
# 2. 搜索 "Claude Code"
# 3. 点击安装

# CLI部署
npm install -g @anthropic-ai/claude-code

# 配置环境变量
export ANTHROPIC_API_KEY="sk-ant-xxx"
# Windows PowerShell:
$env:ANTHROPIC_API_KEY="sk-ant-xxx"

# 使用DeepSeek替代Claude（修改base_url）
export ANTHROPIC_BASE_URL="https://api.deepseek.com/v1"
export ANTHROPIC_API_KEY="sk-deepseek-xxx"

# 使用通义千问
export ANTHROPIC_BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"
export ANTHROPIC_API_KEY="sk-qwen-xxx"

# 通过代理访问
export HTTPS_PROXY="http://127.0.0.1:10808"

# 执行第一个代码生成
claude "编写一个Python函数，查询手机号对应的用户套餐信息"

# Claude Code会：
# 1. 读取项目上下文（CLAUDE.md）
# 2. 理解需求
# 3. 生成代码
# 4. 可直接创建/修改文件

# 模型选择
claude --model claude-sonnet-4-20250514

# 温度、最大令牌
# 在CLAUDE.md或对话中设置

# 工作目录
cd /your/project && claude
# Claude Code会读取工作目录下的CLAUDE.md

# 添加文件到上下文
/context add src/main.py
/context add src/utils.py

# 添加URL到上下文
/context add https://docs.example.com/api

# 记忆系统
/remember "本项目使用Python 3.11，数据库为PostgreSQL"
/remember "电信工单表名为work_orders，主键ticket_id"

# 查看记忆列表
/forget  # 删除指定记忆

# skills.yaml - 自定义技能定义
name: telecom-data-export
description: |
  电信数据导出技能
  当用户需要导出话单、工单、告警数据时使用
input_schema:
  type: object
  properties:
    data_type:
      type: string
      enum: [call_records, work_orders, alarms]
      description: 数据类型
    format:
      type: string
      enum: [csv, xlsx, json]
      description: 导出格式
    date_range:
      type: string
      description: 日期范围，如2026-07-01~2026-07-31
  required: [data_type, format]
output_schema:
  type: object
  properties:
    file_path:
      type: string
    row_count:
      type: integer
examples:
  - input: {"data_type": "call_records", "format": "csv", "date_range": "2026-07-01~2026-07-31"}
    output: {"file_path": "export/call_records_202607.csv", "row_count": 15234}

# 用户画像生成器 PRD

## 功能列表
1. 输入手机号，查询用户近3个月使用数据
2. 分析用户消费水平、流量偏好、通话习惯
3. 生成画像标签（如"大流量用户"、"价格敏感"）
4. 输出画像JSON和可视化报告

## 用户故事
- 作为营销人员，我想查看用户画像，以便推荐合适套餐
- 作为客服，我想了解用户特征，以便个性化服务

## 非功能需求
- 响应时间 < 3秒
- 支持批量查询（100个手机号/次）
- 数据脱敏处理

# 用户画像生成器 技术Spec

## 输入
- phone: string (手机号，11位)
- months: int (分析月数，默认3)

## 输出
{
  "phone": "138****8000",
  "profile": {
    "data_preference": "high",      // high/medium/low
    "call_preference": "medium",
    "price_sensitivity": "high",
    "churn_risk": "low"
  },
  "tags": ["大流量用户", "价格敏感", "忠诚度高"],
  "recommendation": "推荐30GB以上套餐"
}

## 异常处理
- 手机号不存在 → {"error": "USER_NOT_FOUND"}
- 数据不足 → {"error": "INSUFFICIENT_DATA"}
- 限流 → {"error": "RATE_LIMITED", "retry_after": 60}

## 依赖库
- requests（HTTP调用）
- pandas（数据分析）

# 将Spec和Task作为Prompt输入Claude Code
claude "
请基于以下技术规格开发用户画像生成器：

[粘贴Spec内容]

开发任务清单：
[粘贴Task列表]

请逐个Task完成，每完成一个Task后运行测试验证。
"

from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

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

# 管道运算符 | 连接组件
chain = prompt | llm | StrOutputParser()

# RunnableParallel: 并行执行
parallel = RunnableParallel(
    summary=summary_chain,
    keywords=keyword_chain,
    original=RunnablePassthrough()  # 原样传递
)

# RunnablePassthrough: 透传输入
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

from langchain.callbacks import StdOutCallbackHandler

# 记录执行日志
handler = StdOutCallbackHandler()
chain.invoke("查询用户信息", config={"callbacks": [handler]})

# LangSmith追踪（生产环境推荐）
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-xxx"
# 所有链调用自动上报到LangSmith可视化平台

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.chat_models import ChatOllama

# 所有模型统一接口
models = {
    "gpt4": ChatOpenAI(model="gpt-4o"),
    "claude": ChatAnthropic(model="claude-sonnet-4-20250514"),
    "local": ChatOllama(model="qwen2.5:7b", base_url="http://localhost:11434"),
}

# 统一调用
for name, llm in models.items():
    result = llm.invoke("什么是5G？")
    print(f"[{name}] {result.content[:50]}...")

class ModelRouter:
    """根据任务复杂度动态选择模型"""
    
    def __init__(self):
        self.fast_model = ChatOpenAI(model="gpt-4o-mini")  # 便宜快速
        self.smart_model = ChatOpenAI(model="gpt-4o")      # 贵但智能
    
    def route(self, task_type: str):
        if task_type in ["分类", "提取", "格式化"]:
            return self.fast_model
        elif task_type in ["分析", "推理", "生成"]:
            return self.smart_model
        else:
            return self.fast_model  # 默认用小模型

router = ModelRouter()
llm = router.route("分类")
result = llm.invoke("将此工单分类：宽带频繁掉线")

# 本地部署时，可根据任务动态切换基础模型和LoRA适配器
# 使用vLLM支持的热加载功能
# 基础模型处理通用任务，LoRA模型处理电信专业任务

from langchain.tools import tool
from pydantic import BaseModel, Field

# 方式1: 简单工具
@tool
def query_data_balance(phone: str) -> str:
    """查询用户流量余额
    
    Args:
        phone: 手机号，11位数字
        
    Returns:
        流量余额信息
    """
    # 模拟查询
    return f"手机号{phone}剩余流量：15.2GB"

# 方式2: 使用Pydantic精细化校验入参
class PackageQueryInput(BaseModel):
    phone: str = Field(
        pattern=r"^1[3-9]\d{9}$",
        description="手机号，11位数字"
    )
    month: str = Field(
        default="2026-07",
        description="查询月份，格式YYYY-MM"
    )

@tool(args_schema=PackageQueryInput)
def query_package_detail(phone: str, month: str) -> dict:
    """查询用户套餐使用详情"""
    return {
        "phone": phone,
        "month": month,
        "package": "5G畅享99",
        "data_used": "28.5GB",
        "data_total": "30GB",
        "calls_used": "456分钟",
        "calls_total": "500分钟"
    }

@tool
def create_trouble_ticket(device_id: str, description: str) -> dict:
    """创建运维工单
    
    Returns:
        成功: {"status": "success", "ticket_id": "TKT-xxx"}
        失败: {"status": "error", "message": "失败原因"}
    """
    try:
        if not device_id:
            return {"status": "error", "message": "设备ID不能为空"}
        ticket_id = f"TKT-{device_id}-001"
        return {"status": "success", "ticket_id": ticket_id}
    except Exception as e:
        return {"status": "error", "message": str(e)}

from langchain.chains import LLMChain

# 顺序链：检索 → 生成 → 格式化
retrieval_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 路由链：根据输入类型选择不同处理链
def route_chain(info):
    if "宽带" in info["topic"]:
        return broadband_chain
    elif "话费" in info["topic"]:
        return billing_chain
    else:
        return general_chain

from langchain_core.runnables import RunnableLambda
router_chain = (
    {"topic": RunnablePassthrough()}
    | RunnableLambda(route_chain)
)

# LCEL实现条件分支
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "紧急" in x["input"], urgent_handler),
    (lambda x: "普通" in x["input"], normal_handler),
    default_handler
)

# 并行执行
parallel = RunnableParallel(
    analysis=analysis_chain,
    summary=summary_chain,
    keywords=keyword_chain
)
result = parallel.invoke({"input": "分析此告警"})

from langchain.memory import BaseChatMemory
import redis
import json

class TelecomUserMemory(BaseChatMemory):
    """基于手机号的用户记忆隔离"""
    
    def __init__(self, phone: str, redis_client=None):
        super().__init__()
        self.phone = phone
        self.redis = redis_client or redis.Redis(host='localhost', port=6379)
        self.key = f"chat_memory:{phone}"
    
    def save_context(self, inputs, outputs):
        """保存对话到Redis（设置7天TTL）"""
        history = self.load_memory_variables({})
        messages = history.get("history", [])
        messages.extend([
            {"role": "user", "content": str(inputs)},
            {"role": "assistant", "content": str(outputs)}
        ])
        # 超过50条时做摘要压缩
        if len(messages) > 50:
            messages = self._compress(messages)
        self.redis.setex(self.key, 7*86400, json.dumps(messages))
    
    def load_memory_variables(self, inputs):
        """加载用户历史记忆"""
        data = self.redis.get(self.key)
        if data:
            return {"history": json.loads(data)}
        return {"history": []}
    
    def _compress(self, messages):
        """记忆压缩：保留最近10条 + 前面的摘要"""
        old = messages[:-10]
        recent = messages[-10:]
        summary = f"之前对话摘要：用户咨询了{len(old)}条消息，主要关于套餐和故障问题。"
        return [{"role": "system", "content": summary}] + recent

from langchain.agents import create_tool_calling_agent, AgentExecutor

# OpenAI Functions Agent（推荐）
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=10)

# Plan-and-Execute（复杂任务）
from langchain.agents import create_plan_and_execute_agent

# 客服话术模板注入用户画像
CUSTOMER_SERVICE_PROMPT = """
你是电信客服助手。当前用户画像：
- 姓名：{user_name}
- 套餐：{package_name}
- 月均消费：{avg_spending}元
- 投诉历史：{complaint_count}次

用户提问：{question}

请基于用户画像提供个性化回答。
语气亲切，称呼用户姓名。高消费用户推荐升级套餐，低消费用户侧重性价比。
"""

# 工单分类Few-shot
TICKET_CLASSIFY_PROMPT = """
请将工单分类为以下类别之一：
- 宽带故障
- 话费争议
- 增值业务
- 网络覆盖
- 其他

示例：
工单：用户说家里WiFi频繁断线 → 宽带故障
工单：上月话费多了30元不知为何 → 话费争议
工单：想取消每月10元的来电显示 → 增值业务

工单：{ticket_content}
分类：
"""

# 诱导LLM输出结构化数据
STRUCTURED_OUTPUT_PROMPT = """
请以JSON格式输出分析结果，不要输出其他内容。

{
  "device_id": "设备ID",
  "alarm_level": "critical/warning/info",
  "root_cause": "根因分析",
  "suggestion": "处理建议",
  "priority": 1-5
}
"""

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

# 1. State: 共享状态Schema
class AgentState(TypedDict):
    messages: list
    current_step: str
    result: str

# 2. Node: 节点（处理函数）
def agent_node(state: AgentState) -> dict:
    # 处理逻辑
    return {"result": "处理完成"}

# 3. Edge: 边（连接节点）
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_edge("agent", "tools")
graph.add_edge("tools", END)

# 4. 编译并执行
app = graph.compile()
result = app.invoke({"messages": [("user", "查询套餐")]})

from pydantic import BaseModel

# 方式1: TypedDict（轻量）
class LightState(TypedDict):
    query: str
    result: str

# 方式2: Pydantic Model（带校验）
class StrictState(BaseModel):
    phone: str
    intent: str
    confidence: float
    response: str = ""

def route_by_intent(state: AgentState) -> str:
    """根据意图路由到不同节点"""
    intent = state.get("intent", "")
    if intent == "宽带":
        return "broadband_node"
    elif intent == "话费":
        return "billing_node"
    else:
        return "general_node"

graph.add_conditional_edges(
    "classifier",          # 源节点
    route_by_intent,       # 路由函数
    {
        "broadband_node": "broadband_diagnostic",
        "billing_node": "billing_query",
        "general_node": "general_handler"
    }
)

def quality_check(state: AgentState) -> str:
    """质量检查，不合格则回到生成节点重试"""
    if state.get("quality_score", 0) < 0.8:
        state["retry_count"] = state.get("retry_count", 0) + 1
        if state["retry_count"] <= 3:
            return "regenerate"  # 回到生成节点
    return "pass"  # 通过，继续向下

graph.add_conditional_edges(
    "quality_checker",
    quality_check,
    {"regenerate": "generator", "pass": "output"}
)

from langgraph.checkpoint.memory import MemorySaver

# 在关键节点前暂停，等待人工输入
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["human_review"]  # 在人工审核节点前暂停
)

# 第一次执行（到human_review前暂停）
config = {"configurable": {"thread_id": "task-001"}}
result = app.invoke({"query": "处理此告警"}, config=config)

# 人工输入后恢复
result = app.invoke(
    {"human_feedback": "确认为光模块故障，执行更换"},
    config=config  # 同一thread_id恢复
)

class FaultState(TypedDict):
    raw_text: str           # 原始报障文本
    fault_type: str         # 故障类型
    account_status: str     # 账号状态
    modem_online: bool      # 光猫在线
    ping_results: list      # 测速结果
    retry_count: int        # 重试次数
    diagnosis: str          # 诊断结论
    solution: str           # 解决方案

def ping_test(state: FaultState) -> dict:
    """ping测速节点，支持循环"""
    retry = state.get("retry_count", 0)
    if retry < 3:
        # 执行ping测速
        result = "延迟: 45ms, 丢包: 2%"
        results = state.get("ping_results", []) + [result]
        return {"ping_results": results, "retry_count": retry + 1}
    return {"ping_results": state.get("ping_results", [])}

def check_ping_quality(state: FaultState) -> str:
    """判断ping结果质量"""
    results = state.get("ping_results", [])
    if not results:
        return "retry"  # 没结果，重试
    last = results[-1]
    if "丢包: 0%" in last:
        return "pass"   # 正常
    elif state.get("retry_count", 0) >= 3:
        return "fail"   # 重试3次仍失败
    return "retry"      # 继续重试

# 构建子图
broadband_subgraph = StateGraph(FaultState)
broadband_subgraph.add_node("check_account", check_account_node)
broadband_subgraph.add_node("check_modem", check_modem_node)
broadband_subgraph.add_node("ping_test", ping_test)
broadband_subgraph.add_node("diagnose", diagnose_node)

broadband_subgraph.set_entry_point("check_account")
broadband_subgraph.add_edge("check_account", "check_modem")
broadband_subgraph.add_edge("check_modem", "ping_test")
broadband_subgraph.add_conditional_edges(
    "ping_test", check_ping_quality,
    {"retry": "ping_test", "pass": "diagnose", "fail": "diagnose"}
)
broadband_subgraph.add_edge("diagnose", END)

class MultiAgentState(TypedDict):
    task: str
    subtasks: list
    results: dict
    final_report: str

def supervisor(state: MultiAgentState) -> dict:
    """主Agent：分解任务并分发"""
    task = state["task"]
    subtasks = [
        {"agent": "alarm_analyst", "task": "分析告警内容"},
        {"agent": "fault_locator", "task": "定位故障根因"},
        {"agent": "dispatcher", "task": "生成派单建议"}
    ]
    return {"subtasks": subtasks}

# 子Agent作为独立节点
def alarm_analyst(state: MultiAgentState) -> dict:
    """告警分析子Agent"""
    result = llm.invoke(f"分析告警: {state['task']}")
    return {"results": {"alarm_analysis": result.content}}

def fault_locator(state: MultiAgentState) -> dict:
    """故障定位子Agent"""
    alarm_result = state["results"].get("alarm_analysis", "")
    result = llm.invoke(f"基于告警分析结果定位故障: {alarm_result}")
    return {"results": {"fault_location": result.content}}

# 每个子Agent有独立的状态空间
# 主Agent通过共享State聚合结果

def aggregate(state: MultiAgentState) -> dict:
    """聚合各子Agent结果"""
    results = state["results"]
    report = f"""
    告警分析: {results.get('alarm_analysis', 'N/A')}
    故障定位: {results.get('fault_location', 'N/A')}
    派单建议: {results.get('dispatch', 'N/A')}
    """
    return {"final_report": report}

# 将LangChain的Runnable链包装为LangGraph节点
from langchain_core.runnables import Runnable

def langchain_node(runnable: Runnable):
    """将LangChain Runnable包装为LangGraph节点"""
    def node(state):
        # 从state中提取输入
        input_data = state.get("input", "")
        result = runnable.invoke(input_data)
        return {"output": result}
    return node

# 在LangGraph中使用LangChain的Retriever和Memory
graph = StateGraph(AgentState)
graph.add_node("retrieve", langchain_node(retriever_chain))
graph.add_node("generate", langchain_node(generation_chain))

class PlanExecuteState(TypedDict):
    task: str
    plan: list          # 执行计划
    current_step: int   # 当前步骤
    step_results: dict  # 各步骤结果
    final_answer: str

def planner(state: PlanExecuteState) -> dict:
    """规划Agent：将任务分解为步骤"""
    task = state["task"]
    prompt = f"""将以下任务分解为3-5个可执行步骤：
    任务：{task}
    输出JSON数组格式。"""
    
    result = llm.invoke(prompt)
    # 解析计划
    plan = parse_plan(result.content)
    return {"plan": plan, "current_step": 0}

def executor(state: PlanExecuteState) -> dict:
    """执行Agent：执行当前步骤"""
    plan = state["plan"]
    step_idx = state["current_step"]
    current = plan[step_idx]
    
    # 动态选择工具执行
    result = execute_step(current)
    
    step_results = state.get("step_results", {})
    step_results[current["name"]] = result
    
    return {
        "step_results": step_results,
        "current_step": step_idx + 1
    }

def should_continue(state: PlanExecuteState) -> str:
    """判断是否继续执行"""
    if state["current_step"] < len(state["plan"]):
        return "continue"
    return "done"

# 构建Plan-Execute工作流
workflow = StateGraph(PlanExecuteState)
workflow.add_node("planner", planner)
workflow.add_node("executor", executor)
workflow.add_node("synthesizer", synthesize_result)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_conditional_edges(
    "executor", should_continue,
    {"continue": "executor", "done": "synthesizer"}
)
workflow.add_edge("synthesizer", END)

plan_execute_app = workflow.compile()

# 执行
result = plan_execute_app.invoke({
    "task": "解决用户信号差投诉：用户在朝阳区反馈5G信号弱",
    "step_results": {}
})
print(result["final_answer"])

---
name: query-balance
description: 查询用户余额和套餐使用情况
version: 1.0.0
author: Telecom Team
---

# 查询余额技能

## 使用场景
当用户询问话费余额、流量余量、套餐使用情况时使用

## 输入参数
- phone: 手机号（必填）
- item: 查询项（balance/data/calls，默认balance）

## 输出格式
{
  "phone": "138****8000",
  "balance": 45.6,
  "package": "5G畅享99",
  "data_remaining": "15.2GB"
}

## 示例
输入: {"phone": "13800138000", "item": "balance"}
输出: {"phone": "138****8000", "balance": 45.6, "currency": "CNY"}

# Skill放置在.claude/skills/目录下
# Agent通过description匹配自动发现

目录结构:
.claude/skills/
├── query-balance/
│   └── SKILL.md
├── create-ticket/
│   └── SKILL.md
└── diagnose-fault/
    ├── SKILL.md
    └── scripts/
        └── diagnose.py

# Claude Code/Agent自动扫描skills目录
# 根据用户意图匹配description
# 加载对应SKILL.md并按流程执行

from fastmcp import FastMCP

mcp = FastMCP("电信客服服务")

# 实现Tools
@mcp.tool()
def query_balance(phone: str) -> dict:
    """查询用户余额
    
    Args:
        phone: 手机号
    Returns:
        余额信息
    """
    return {"phone": phone, "balance": 45.6, "package": "5G畅享99"}

@mcp.tool()
def order_data_package(phone: str, package_id: str) -> dict:
    """订购流量包
    
    Args:
        phone: 手机号
        package_id: 流量包ID（如: 10GB-30元）
    """
    return {"status": "success", "phone": phone, "package": package_id}

@mcp.tool()
def send_notification(channel: str, message: str) -> dict:
    """发送通知"""
    return {"status": "sent", "channel": channel}

# 实现Resources：暴露知识文档
@mcp.resource("docs://troubleshooting/{topic}")
def get_troubleshooting_doc(topic: str) -> str:
    """获取故障排除文档"""
    docs = {
        "broadband": "宽带故障排查：1.检查光猫 2.检查网线 3.重启设备",
        "5g_signal": "5G信号问题：1.检查覆盖 2.检查SIM卡 3.网络设置",
    }
    return docs.get(topic, "暂无此文档")

# 启动Server
if __name__ == "__main__":
    mcp.run(transport="stdio")

# mcp-cli: 命令行调试MCP Server

# 1. 查看MCP Server提供的能力
mcp-cli inspect python telecom_server.py
# 输出: Tools: [query_balance, order_data_package, send_notification]
#       Resources: [docs://troubleshooting/{topic}]

# 2. 直接调用某个工具
mcp-cli call query_balance --phone 13800138000
# 输出: {"phone": "13800138000", "balance": 45.6, "package": "5G畅享99"}

# 3. 读取资源
mcp-cli read docs://troubleshooting/broadband

from fastmcp import FastMCP
from functools import wraps

mcp = FastMCP("企业电信服务")

# API Key认证中间件
TENANT_KEYS = {
    "tenant_a": "key-aaa-123",
    "tenant_b": "key-bbb-456",
}

def require_auth(tenant: str, api_key: str) -> bool:
    return TENANT_KEYS.get(tenant) == api_key

@mcp.tool()
def query_user_balance(phone: str, tenant: str, api_key: str) -> dict:
    """查询用户余额（需租户认证）"""
    if not require_auth(tenant, api_key):
        return {"error": "认证失败", "code": "AUTH_FAILED"}
    # 不同租户数据隔离
    return {"phone": phone, "balance": 45.6, "tenant": tenant}

import time
import logging

logger = logging.getLogger("mcp_server")

def monitor_tool(func):
    """工具调用监控装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            duration = time.time() - start
            logger.info(f"工具{func.__name__}执行成功, 耗时{duration:.3f}s")
            return result
        except Exception as e:
            duration = time.time() - start
            logger.error(f"工具{func.__name__}执行失败, 耗时{duration:.3f}s, 错误: {e}")
            raise
    return wrapper

@mcp.tool()
@monitor_tool
def query_balance(phone: str) -> dict:
    return {"balance": 45.6}

# Dockerfile
FROM python:3.11-slim
COPY . /app
WORKDIR /app
RUN pip install fastmcp
EXPOSE 8080
CMD ["python", "server.py", "--transport", "http", "--port", "8080"]

# docker-compose.yml
version: '3'
services:
  mcp-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - DB_HOST=postgres
      - REDIS_HOST=redis
    depends_on:
      - postgres
      - redis
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: telecom
  redis:
    image: redis:7

# MCP Server对接工单数据库
@mcp.tool()
def query_ticket_status(ticket_id: str) -> dict:
    """查询工单状态"""
    return {"ticket_id": ticket_id, "status": "处理中", "handler": "张三"}

@mcp.tool()
def transfer_ticket(ticket_id: str, to_team: str) -> dict:
    """转派工单"""
    return {"ticket_id": ticket_id, "transferred_to": to_team}

@mcp.tool()
def close_ticket(ticket_id: str, resolution: str) -> dict:
    """关闭工单"""
    return {"ticket_id": ticket_id, "status": "closed", "resolution": resolution}

# Skill定义
# SKILL.md:
# 1. 用户说"我的工单处理到哪了" → 调用query_ticket_status
# 2. 用户说"转给宽带组" → 调用transfer_ticket
# 3. 用户说"已解决，可以关闭" → 调用close_ticket

# 画像分析MCP工具
@mcp.tool()
def aggregate_user_profile(phone: str) -> dict:
    """聚合用户画像：上网行为+资费敏感度→360度画像"""
    return {
        "phone": phone,
        "profile": {
            "data_preference": "high",
            "price_sensitivity": "medium",
            "churn_risk": "low",
            "tags": ["大流量用户", "稳定客户"]
        }
    }

# 日志分析MCP Server
@mcp.tool()
def query_error_rate(service: str, hours: int = 1) -> dict:
    """查询错误率（对接ELK/CLS）"""
    return {"service": service, "error_rate": 2.3, "trend": "上升"}

@mcp.tool()
def get_error_stack(trace_id: str) -> str:
    """获取异常堆栈"""
    return f"TraceID: {trace_id}\nNullPointerException at com.telecom.Service..."

@mcp.tool()
def analyze_root_cause(error_log: str) -> str:
    """Agent根据日志自动分析根因"""
    root_cause = llm.invoke(f"分析以下日志的根因: {error_log}")
    return root_cause.content

# 验证所有服务连通性
def verify_environment():
    # 1. Ollama
    import requests
    r = requests.get("http://localhost:11434/api/tags")
    assert r.status_code == 200, "Ollama未启动"
    
    # 2. Chroma
    import chromadb
    client = chromadb.PersistentClient(path="./chroma_db")
    assert client.heartbeat() > 0, "Chroma异常"
    
    # 3. MCP Server
    from mcp import ClientSession
    # 测试MCP连接...
    
    print("所有服务连通性验证通过")

# MCP Server中的测速工具
@mcp.tool()
async def run_speed_test(account_id: str) -> dict:
    """执行宽带测速
    
    Args:
        account_id: 宽带账号ID
    Returns:
        测速结果
    """
    # 调用测速API
    return {
        "account_id": account_id,
        "download_mbps": 95.6,
        "upload_mbps": 28.3,
        "latency_ms": 12,
        "jitter_ms": 2.1,
        "rating": "良好"  # 优秀/良好/一般/差
    }

@mcp.tool()
def query_broadband_account(phone: str) -> dict:
    """查询宽带账号信息
    
    Args:
        phone: 用户手机号
    """
    return {
        "account_id": "KD20260001",
        "phone": phone,
        "package": "千兆宽带200M",
        "address": "北京市朝阳区XX小区",
        "install_date": "2024-03-15",
        "status": "正常"
    }

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings  # 或本地Embedding

# 初始化故障知识库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="fault_kb"
)

def search_fault_kb(symptom: str, k: int = 3) -> list:
    """检索故障知识库
    
    Args:
        symptom: 故障症状描述
        k: 返回结果数
    """
    results = vectorstore.similarity_search(symptom, k=k)
    return [
        {"content": doc.page_content, "source": doc.metadata.get("source")}
        for doc in results
    ]

from typing import TypedDict

class BroadbandAssistantState(TypedDict):
    phone: str               # 用户手机号
    account_info: dict       # 宽带账号信息
    speed_test_result: dict  # 测速结果
    fault_history: list      # 历史故障
    kb_results: list         # 知识库检索结果
    intent: str              # 用户意图
    response: str            # 最终回复