📖 对话系统与Agent化NLP¶

import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer

class JointIntentSlotModel(nn.Module):  # 继承nn.Module定义网络层
    """
    联合意图识别 + 槽填充模型

    意图识别: [CLS] → 多分类
    槽填充: 每个token → BIO标注
    """
    def __init__(self, bert_name='bert-base-chinese',
                 num_intents=10, num_slot_labels=30):
        super().__init__()  # super()调用父类方法
        self.bert = BertModel.from_pretrained(bert_name)
        hidden_size = self.bert.config.hidden_size

        # 意图分类器（用[CLS]表示）
        self.intent_classifier = nn.Sequential(
            nn.Dropout(0.1),
            nn.Linear(hidden_size, num_intents)
        )

        # 槽填充分类器（每个token）
        self.slot_classifier = nn.Sequential(
            nn.Dropout(0.1),
            nn.Linear(hidden_size, num_slot_labels)
        )

    def forward(self, input_ids, attention_mask):
        outputs = self.bert(input_ids, attention_mask=attention_mask)

        # 意图: [CLS] token
        cls_output = outputs.last_hidden_state[:, 0, :]
        intent_logits = self.intent_classifier(cls_output)

        # 槽位: 所有token
        sequence_output = outputs.last_hidden_state
        slot_logits = self.slot_classifier(sequence_output)

        return intent_logits, slot_logits

# 使用示例
model = JointIntentSlotModel(num_intents=5, num_slot_labels=15)
tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')

text = "帮我订明天从北京飞上海的机票"
inputs = tokenizer(text, return_tensors='pt', padding=True)
intent_logits, slot_logits = model(inputs['input_ids'], inputs['attention_mask'])

intent_labels = ['BookFlight', 'BookHotel', 'Weather', 'Music', 'Other']
pred_intent = intent_labels[intent_logits.argmax(-1).item()]  # 将单元素张量转为Python数值
print(f"意图: {pred_intent}")
print(f"槽位标签数: {slot_logits.shape}")  # (1, seq_len, num_slot_labels)

class DialogueStateTracker:
    """
    规则+模型混合的DST

    维护对话状态 = {domain: {slot: value}}
    """
    def __init__(self, domains):
        self.state = {}
        for domain in domains:
            self.state[domain] = {}
        self.history = []

    def update(self, turn_nlu_result):
        """
        根据NLU结果更新状态

        Args:
            turn_nlu_result: {
                'intent': str,
                'slots': {slot_name: slot_value},
                'domain': str
            }
        """
        domain = turn_nlu_result['domain']
        for slot, value in turn_nlu_result['slots'].items():
            if value is not None:
                self.state[domain][slot] = value

        self.history.append({
            'nlu': turn_nlu_result,
            'state': self._copy_state()
        })

    def get_unfilled_slots(self, domain, required_slots):
        """检查必填槽位是否都已填充"""
        filled = set(self.state.get(domain, {}).keys())
        unfilled = [s for s in required_slots if s not in filled]
        return unfilled

    def _copy_state(self):
        import copy
        return copy.deepcopy(self.state)

    def __repr__(self):
        return f"DialogueState: {self.state}"

# 示例对话
tracker = DialogueStateTracker(domains=['flight', 'hotel'])
required_flight_slots = ['from_city', 'to_city', 'date', 'time']

# 第1轮
tracker.update({
    'intent': 'BookFlight', 'domain': 'flight',
    'slots': {'from_city': '北京', 'to_city': '上海', 'date': '明天'}
})
unfilled = tracker.get_unfilled_slots('flight', required_flight_slots)
print(f"轮次1状态: {tracker}")
print(f"未填槽位: {unfilled}")  # ['time']

# 第2轮
tracker.update({
    'intent': 'BookFlight', 'domain': 'flight',
    'slots': {'time': '上午'}
})
unfilled = tracker.get_unfilled_slots('flight', required_flight_slots)
print(f"轮次2状态: {tracker}")
print(f"未填槽位: {unfilled}")  # []

def build_persona_prompt(persona_facts, dialogue_history, user_input):
    """
    构建带人设的对话prompt

    persona_facts: 角色设定
    dialogue_history: 历史对话
    user_input: 用户最新输入
    """
    prompt = "你是一个AI助手，以下是你的人设：\n"
    for fact in persona_facts:
        prompt += f"- {fact}\n"

    prompt += "\n请始终保持人设一致。如果用户的问题超出你的人设，委婉地引导回相关话题。\n\n"
    prompt += "对话历史：\n"
    for role, text in dialogue_history:
        prompt += f"{role}: {text}\n"

    prompt += f"用户: {user_input}\n助手: "
    return prompt

# 示例
persona = [
    "我叫小明，25岁",
    "我是一名AI算法工程师",
    "我喜欢打篮球和看科幻电影",
    "我养了一只叫Cookie的猫"
]

history = [
    ("用户", "你好啊，你是做什么工作的？"),
    ("助手", "你好！我是一名AI算法工程师，主要做NLP相关的工作。")
]

prompt = build_persona_prompt(persona, history, "你平时有什么爱好？")
print(prompt)

class ConversationManager:
    """
    多轮对话管理器

    功能：
    1. 上下文窗口管理（滑动窗口/摘要压缩）
    2. 共指消解（"它"→指代什么？）
    3. 话题追踪与切换检测
    """
    def __init__(self, max_turns=20, summary_threshold=10):
        self.turns = []
        self.max_turns = max_turns
        self.summary_threshold = summary_threshold
        self.summary = ""
        self.current_topic = None
        self.entities = {}  # 实体追踪 {mention: entity}

    def add_turn(self, role, content, entities=None):
        """添加一轮对话"""
        self.turns.append({
            'role': role,
            'content': content,
            'entities': entities or {}
        })

        # 更新实体追踪
        if entities:
            self.entities.update(entities)

        # 如果超过阈值，压缩历史
        if len(self.turns) > self.summary_threshold:
            self._compress_history()

    def resolve_coreference(self, text):
        """
        简易共指消解
        将代词替换为最近提到的实体
        """
        pronouns = {'它': None, '他': None, '她': None, '这个': None, '那个': None}

        # 查找最近的实体
        for turn in reversed(self.turns):
            for mention, entity in turn.get('entities', {}).items():
                if pronouns.get('它') is None and entity.get('type') == 'thing':
                    pronouns['它'] = mention
                    pronouns['这个'] = mention
                if pronouns.get('他') is None and entity.get('type') == 'person_male':
                    pronouns['他'] = mention

        # 替换
        resolved = text
        for pronoun, replacement in pronouns.items():
            if replacement and pronoun in resolved:
                resolved = resolved.replace(pronoun, replacement)

        return resolved

    def detect_topic_switch(self, new_utterance):
        """检测话题切换（简化版：基于关键词重叠）"""
        if not self.turns:
            return True

        last_content = self.turns[-1]['content']  # [-1]负索引取最后元素
        last_words = set(last_content)
        new_words = set(new_utterance)
        overlap = len(last_words & new_words) / max(len(last_words | new_words), 1)

        return overlap < 0.1  # 重叠极少 → 话题切换

    def _compress_history(self):
        """将较早的对话压缩为摘要"""
        # 保留最近的几轮，早期的压缩为摘要
        keep_recent = 5
        to_compress = self.turns[:-keep_recent]

        # 简化版摘要（实际应用中用LLM生成摘要）
        self.summary += "\n".join(
            f"{t['role']}: {t['content'][:50]}..." for t in to_compress  # 切片操作，取前n个元素
        )
        self.turns = self.turns[-keep_recent:]

    def get_context(self):
        """获取当前上下文（摘要 + 近几轮）"""
        context = ""
        if self.summary:
            context += f"[早期对话摘要]\n{self.summary}\n\n"
        context += "[近期对话]\n"
        for turn in self.turns:
            context += f"{turn['role']}: {turn['content']}\n"
        return context

# 使用示例
mgr = ConversationManager()
mgr.add_turn("用户", "我想了解一下BERT模型",
             {"BERT": {"type": "thing"}})
mgr.add_turn("助手", "BERT是Google提出的预训练语言模型，使用MLM和NSP任务预训练。")
mgr.add_turn("用户", "它和GPT有什么区别？")

# 共指消解
resolved = mgr.resolve_coreference("它和GPT有什么区别？")
print(f"消解后: {resolved}")  # "BERT和GPT有什么区别？"
print(f"\n当前上下文:\n{mgr.get_context()}")

def disambiguate(user_input, candidates, context):
    """
    多轮对话中的消歧

    场景：用户说"苹果"，是水果还是公司？
    """
    # 策略1：上下文推断
    tech_keywords = ['手机', '电脑', 'Mac', 'iOS', '发布会', '股价']
    fruit_keywords = ['吃', '水果', '甜', '红色', '超市', '好吃']

    context_words = set(context.lower().split())

    tech_score = len(context_words & set(tech_keywords))
    fruit_score = len(context_words & set(fruit_keywords))

    if tech_score > fruit_score:
        return "Apple Inc. (苹果公司)"
    elif fruit_score > tech_score:
        return "苹果 (水果)"
    else:
        # 策略2：主动询问
        return "ASK: 您说的苹果是指苹果公司还是水果呢？"

# 示例
result = disambiguate("苹果", [], "我想买个新手机")
print(f"消歧结果: {result}")  # Apple Inc.

import json

# 工具定义
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索互联网获取实时信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "执行数学计算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式"}
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_database",
            "description": "查询SQL数据库",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL查询语句"},
                    "database": {"type": "string", "description": "数据库名"}
                },
                "required": ["sql", "database"]
            }
        }
    }
]

# 工具执行器
class ToolExecutor:
    def __init__(self):
        self.tools = {
            "search_web": self._search_web,
            "calculator": self._calculator,
            "query_database": self._query_database,
        }

    def execute(self, tool_name, arguments):
        if tool_name not in self.tools:
            return {"error": f"Unknown tool: {tool_name}"}
        return self.tools[tool_name](**arguments)

    def _search_web(self, query):
        # 实际调用搜索API
        return {"results": [f"搜索 '{query}' 的结果..."]}

    def _calculator(self, expression):
        try:  # try/except捕获异常
            # 安全的数学计算（实际应用使用更安全的eval）
            result = eval(expression, {"__builtins__": {}}, {})
            return {"result": result}
        except Exception as e:
            return {"error": str(e)}

    def _query_database(self, sql, database):
        return {"result": f"执行 {sql} on {database}"}

# 模拟完整的Tool Use循环
def tool_use_loop(user_query, tools_spec, executor, max_iterations=5):
    """
    Tool Use主循环

    1. LLM判断是否需要工具
    2. 如需要，执行工具
    3. 将结果反馈给LLM
    4. LLM给出最终回答
    """
    messages = [{"role": "user", "content": user_query}]

    for i in range(max_iterations):
        # 模拟LLM响应（实际调用API）
        # response = llm.chat(messages, tools=tools_spec)

        # 模拟：LLM决定调用calculator
        if "计算" in user_query or any(c.isdigit() for c in user_query):  # any()任一为True则返回True
            tool_call = {
                "name": "calculator",
                "arguments": {"expression": "23 * 47"}
            }
            result = executor.execute(tool_call["name"], tool_call["arguments"])

            messages.append({
                "role": "tool",
                "name": tool_call["name"],
                "content": json.dumps(result)  # json.dumps将Python对象序列化为JSON字符串
            })

            print(f"工具调用: {tool_call['name']}({tool_call['arguments']})")
            print(f"工具结果: {result}")

            # 最终回答
            final_answer = f"23 × 47 = {result['result']}"
            print(f"最终回答: {final_answer}")
            return final_answer
        else:
            return "直接回答（不需要工具）"

executor = ToolExecutor()
tool_use_loop("23乘以47等于多少？", TOOLS, executor)

REACT_PROMPT = """你是一个能使用工具的AI助手。请按以下格式回答：

Thought: 分析用户问题，思考是否需要工具
Action: tool_name(args) 或 FINISH
Observation: 工具返回结果（由系统填入）
... (可以重复多次)
Thought: 根据所有信息，给出最终答案
Action: FINISH
Answer: 最终回答

可用工具：
- search_web(query): 搜索网页
- calculator(expression): 数学计算
- query_database(sql, database): 查询数据库

用户问题: {question}
"""

def simulate_react(question):
    """模拟ReAct推理过程"""
    print(f"问题: {question}\n")

    steps = [
        ("Thought", "用户想知道计算结果，我需要用计算器工具"),
        ("Action", "calculator(expression='(365 * 24 * 3600)')"),
        ("Observation", "{'result': 31536000}"),
        ("Thought", "计算完成，一年有31536000秒"),
        ("Action", "FINISH"),
        ("Answer", "一年有 31,536,000 秒（365天 × 24小时 × 3600秒）")
    ]

    for step_type, content in steps:
        print(f"{step_type}: {content}")

simulate_react("一年有多少秒？")

JUDGE_PROMPT = """请评估以下对话回复的质量，从1-5分打分。

评分维度：
1. 相关性（1-5）：回复是否切题
2. 信息量（1-5）：回复是否提供有用信息
3. 一致性（1-5）：是否与对话历史一致
4. 安全性（1-5）：是否有不当内容
5. 流畅度（1-5）：语言是否自然流畅

对话上下文:
{context}

待评估回复:
{response}

请以JSON格式输出评分和理由:
{{
    "relevance": {{"score": X, "reason": "..."}},
    "informativeness": {{"score": X, "reason": "..."}},
    "consistency": {{"score": X, "reason": "..."}},
    "safety": {{"score": X, "reason": "..."}},
    "fluency": {{"score": X, "reason": "..."}},
    "overall": X
}}
"""

def evaluate_response(context, response):
    """使用GPT-4作为裁判评估对话质量"""
    prompt = JUDGE_PROMPT.format(context=context, response=response)
    # actual: response = openai.chat(prompt)
    print(f"评估prompt长度: {len(prompt)} 字符")
    print("实际应用中调用GPT-4 API进行评估")
    return {"overall": 4.2}  # 示例返回

evaluate_response(
    "用户: BERT和GPT有什么区别？",
    "BERT是双向编码器，GPT是单向解码器。BERT擅长理解任务，GPT擅长生成任务。"
)

NLP_AGENT_TASKS = {
    "自动摘要Agent": {
        "描述": "自动提取长文档关键信息",
        "工具": ["文本分段", "关键句提取", "摘要生成", "事实验证"],
        "流程": "分段 → 每段提取关键信息 → 合并 → 验证事实 → 输出"
    },
    "信息抽取Agent": {
        "描述": "从非结构化文本提取结构化信息",
        "工具": ["NER识别", "关系抽取", "事件抽取", "知识图谱查询"],
        "流程": "识别实体 → 判断关系 → 验证一致性 → 写入知识库"
    },
    "多语言翻译Agent": {
        "描述": "处理复杂的多语言翻译任务",
        "工具": ["语言检测", "翻译API", "术语库查询", "质量评估"],
        "流程": "检测语言 → 分句 → 检查术语 → 翻译 → 质量评估 → 后编辑"
    },
    "文档QA Agent": {
        "描述": "基于大量文档回答问题",
        "工具": ["检索器", "阅读理解", "计算器", "表格解析"],
        "流程": "理解问题 → 检索相关段落 → 阅读理解 → 验证答案 → 回答"
    }
}

for name, info in NLP_AGENT_TASKS.items():
    print(f"\n📋 {name}")
    print(f"   描述: {info['描述']}")
    print(f"   工具: {', '.join(info['工具'])}")
    print(f"   流程: {info['流程']}")

# MCP工具定义示例
MCP_NLP_TOOLS = {
    "text_analysis": {
        "name": "text-analysis",
        "description": "分析文本的情感、实体、关键词",
        "inputSchema": {
            "type": "object",
            "properties": {
                "text": {"type": "string"},
                "tasks": {
                    "type": "array",
                    "items": {"enum": ["sentiment", "ner", "keywords", "summary"]},
                    "description": "要执行的分析任务"
                }
            },
            "required": ["text", "tasks"]
        }
    },
    "knowledge_base": {
        "name": "knowledge-base-query",
        "description": "查询知识库获取背景信息",
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "top_k": {"type": "integer", "default": 5}
            },
            "required": ["query"]
        }
    }
}

print("MCP NLP工具定义:")
for tool_id, tool_def in MCP_NLP_TOOLS.items():
    print(f"  {tool_def['name']}: {tool_def['description']}")