LLM智能体与传统对话系统混合架构方案 #

class TaskPlanner:
    """基于ReAct模式的任务规划器"""
    
    async def plan(self, user_request: str, context: DialogueContext) -> ExecutionPlan:
        # 1. LLM生成计划（Thought → Action → Observation循环）
        prompt = self._build_react_prompt(user_request, context)
        llm_output = await self.llm.generate(prompt)
        
        # 2. 解析LLM输出为结构化计划
        raw_plan = self._parse_plan(llm_output)
        
        # 3. 规则校验（防止LLM生成危险操作）
        validated_plan = await self._validate(raw_plan)
        if not validated_plan.safe:
            return self.fallback_to_traditional(user_request)
        
        # 4. 转换为DAG执行图
        dag = self._to_dag(validated_plan)
        
        return ExecutionPlan(
            steps=dag,
            estimated_time=dag.estimated_duration(),
            required_tools=dag.required_tools(),
            fallback_plan=self.fallback_to_traditional(user_request)
        )
    
    async def _validate(self, plan: RawPlan) -> ValidatedPlan:
        """规则校验层 - 防止LLM幻觉导致的危险操作"""
        rules = [
            self._check_sensitive_operations,      # 敏感操作拦截
            self._check_tool_permissions,           # 工具权限校验
            self._check_data_access_scope,          # 数据访问范围
            self._check_rate_limits,               # 频率限制
        ]
        
        for rule in rules:
            result = await rule(plan)
            if not result.passed:
                return ValidatedPlan(safe=False, reason=result.reason)
        
        return ValidatedPlan(safe=True, steps=plan.steps)

class MemoryManager:
    """分层Memory管理器"""
    
    async def retrieve(self, query: str, session_id: str, top_k: int = 5) -> MemoryContext:
        # 1. 短期记忆（直接读取，无需检索）
        short_term = await self.short_term_store.get(session_id)
        
        # 2. 情景记忆（向量检索）
        query_embedding = await self.embedder.encode(query)
        episodic = await self.vector_db.search(
            collection="episodic_memory",
            vector=query_embedding,
            top_k=top_k,
            filter={"user_id": self.current_user_id}
        )
        
        # 3. 长期记忆（全文检索 + 向量混合检索）
        long_term = await self.hybrid_search(
            query=query,
            collections=["faq", "product_manual", "user_profile"]
        )
        
        return MemoryContext(
            short_term=short_term,
            episodic=episodic,
            long_term=long_term
        )
    
    async def compress(self, session_id: str) -> CompressedMemory:
        """对话压缩 - 将长对话摘要为结构化记忆"""
        conversation = await self.short_term_store.get(session_id)
        
        if len(conversation.turns) < self.compression_threshold:
            return  # 对话太短，不需要压缩
        
        # LLM生成摘要
        summary = await self.llm.summarize(
            conversation=conversation,
            format="structured"  # 结构化摘要而非自由文本
        )
        
        # 存储到情景记忆
        await self.episodic_store.store(
            user_id=self.current_user_id,
            memory=summary,
            embedding=await self.embedder.encode(summary.text)
        )
        
        # 清理短期记忆（保留最近N轮）
        await self.short_term_store.trim(session_id, keep_turns=5)

class ToolOrchestrator:
    """工具编排器 - 选择、执行、结果处理"""
    
    async def execute_tool_call(self, tool_call: ToolCall) -> ToolResult:
        # 1. 查找工具定义
        tool_def = self.registry.get(tool_call.name)
        if not tool_def:
            return ToolResult(error=f"Unknown tool: {tool_call.name}")
        
        # 2. 参数校验（JSON Schema验证）
        validation = self._validate_params(tool_call.args, tool_def.schema)
        if not validation.valid:
            return ToolResult(error=f"Invalid params: {validation.errors}")
        
        # 3. 权限检查
        if not self._check_permission(tool_call.name, self.current_user):
            return ToolResult(error="Permission denied")
        
        # 4. 沙箱执行
        try:
            async with self.sandbox.timeout(tool_def.timeout):
                result = await tool_def.execute(tool_call.args)
            
            # 5. 结果后处理（截断过长输出）
            return self._post_process(result, tool_def)
            
        except TimeoutError:
            return ToolResult(error="Tool execution timeout")
        except Exception as e:
            return ToolResult(error=str(e))
    
    def select_tools(self, query: str, available_tools: List[ToolDef]) -> List[ToolDef]:
        """工具选择：语义匹配 + 规则过滤"""
        # 语义相似度
        query_embedding = self.embedder.encode(query)
        tool_embeddings = {t.name: t.embedding for t in available_tools}
        
        scores = {
            name: cosine_similarity(query_embedding, embedding)
            for name, embedding in tool_embeddings.items()
        }
        
        # 规则增强（高频工具权重提升）
        for tool in available_tools:
            if tool.name in self.popular_tools:
                scores[tool.name] *= 1.2
        
        # 返回Top-K
        return sorted(available_tools, key=lambda t: scores[t.name], reverse=True)[:3]

class RAGPipeline:
    """检索增强生成管线"""
    
    def __init__(self, config: RAGConfig):
        self.chunker = RecursiveChunker(
            chunk_size=config.chunk_size,        # 默认512 tokens
            chunk_overlap=config.chunk_overlap,   # 默认50 tokens
            separators=["\n\n", "\n", "。", ""]
        )
        self.retriever = HybridRetriever(
            vector_weight=0.7,    # 向量检索权重
            text_weight=0.3       # BM25文本检索权重
        )
        self.reranker = CrossEncoderReranker(
            model="cross-encoder/multilingual-MiniLM",  # 多语言支持
            top_k=config.rerank_top_k
        )
    
    async def retrieve_and_generate(self, query: str) -> RAGResponse:
        # 1. 检索
        candidates = await self.retriever.search(
            query=query,
            top_k=self.config.retrieval_top_k  # 默认50
        )
        
        # 2. 重排序（Cross-Encoder精确打分）
        reranked = await self.reranker.rank(query, candidates[:20])
        
        # 3. 上下文组装（适配LLM窗口大小）
        context = self._build_context(
            query=query,
            passages=reranked[:self.config.context_top_k],  # 默认5
            max_tokens=self.config.max_context_tokens       # 默认4096
        )
        
        # 4. LLM生成
        response = await self.llm.generate(
            prompt=self._build_rag_prompt(query, context),
            temperature=0.1  # RAG场景低温度保证确定性
        )
        
        # 5. 引用标注
        return RAGResponse(
            text=response.text,
            citations=self._extract_citations(response, reranked),
            retrieval_metadata={"query": query, "passages": len(reranked)}
        )
    
    def _build_context(self, query: str, passages: List[Passage], max_tokens: int) -> str:
        """上下文组装 - 贪心填充，确保不超过LLM窗口"""
        context_parts = []
        current_tokens = 0
        
        for passage in passages:
            passage_tokens = self._count_tokens(passage.text)
            if current_tokens + passage_tokens > max_tokens:
                break
            context_parts.append(f"[{passage.id}] {passage.text}")
            current_tokens += passage_tokens
        
        return "\n\n".join(context_parts)

class MultiLingualRAG(RAGPipeline):
    """面向非洲多语言的RAG优化"""
    
    async def retrieve(self, query: str, language: str) -> List[Passage]:
        # 策略1：语言感知检索
        # 优先检索同语言文档
        same_lang_docs = await self._search_by_language(query, language)
        
        # 策略2：跨语言检索（当同语言文档不足时）
        if len(same_lang_docs) < self.min_results:
            # 将查询翻译为英语，检索英语文档
            english_query = await self.translator.translate(query, target="en")
            en_docs = await self._search_by_language(english_query, "en")
            
            # 将检索结果翻译回原语言（或保持原文+翻译对照）
            return same_lang_docs + [
                Passage(
                    text=doc.text,
                    translation=await self.translator.translate(doc.text, target=language),
                    score=doc.score * 0.8  # 跨语言结果降权
                )
                for doc in en_docs
            ]
        
        return same_lang_docs

class HybridDialogueManager:
    """混合对话管理器"""
    
    async def step(self, nlu_result: NLUResult, context: DialogueContext) -> DialogueAction:
        # 决策1：传统DM可处理的确定性意图
        if nlu_result.intent in self.traditional_intents:
            return await self.traditional_dm.step(nlu_result, context)
        
        # 决策2：LLM处理的开放域对话
        if nlu_result.intent == "open_domain":
            return await self.llm_dm.step(nlu_result, context)
        
        # 决策3：传统DM置信度高，直接执行
        if nlu_result.confidence > 0.85:
            return await self.traditional_dm.step(nlu_result, context)
        
        # 决策4：置信度中等，LLM辅助理解
        if nlu_result.confidence > 0.6:
            clarified = await self.llm_dm.clarify(nlu_result, context)
            if clarified.resolved:
                return await self.traditional_dm.step(clarified.nlu, context)
            return DialogueAction(
                type="clarify",
                text=clarified.question
            )
        
        # 决策5：低置信度，LLM完全接管
        return await self.llm_dm.step(nlu_result, context)
    
    async def _handle_slot_filling(self, nlu: NLUResult, context: DialogueContext):
        """槽位填充 - 传统DM的核心能力"""
        slots = self.slot_manager.get_required_slots(nlu.intent)
        filled_slots = nlu.slots
        
        missing_slots = [s for s in slots if s.name not in filled_slots]
        
        if not missing_slots:
            # 所有槽位已填，执行
            return await self.execute_intent(nlu)
        
        # 请求用户补充槽位
        next_slot = missing_slots[0]
        return DialogueAction(
            type="ask_slot",
            slot=next_slot.name,
            prompt=f"请问{next_slot.prompt}？"
        )

class HybridNLU:
    """混合NLU引擎 - 传统 + LLM"""
    
    async def understand(self, text: str, context: DialogueContext) -> NLUResult:
        # 1. 传统NLU（快速、低成本）
        traditional_result = await self.traditional_nlu.predict(text)
        
        # 2. 高置信度直接返回
        if traditional_result.confidence >= self.high_confidence_threshold:
            return traditional_result
        
        # 3. 低置信度：LLM辅助
        if traditional_result.confidence < self.low_confidence_threshold:
            llm_result = await self.llm_nlu.zero_shot_classify(
                text=text,
                candidate_intents=self.intent_registry.all_intents()
            )
            
            # 融合决策
            return self._fuse_results(traditional_result, llm_result)
        
        # 4. 中等置信度：仅LLM验证
        llm_verification = await self.llm_nlu.verify(
            text=text,
            predicted_intent=traditional_result.intent
        )
        
        if llm_verification.agrees:
            return traditional_result
        else:
            # LLM不同意，使用LLM的分类
            return await self.llm_nlu.classify(text)
    
    def _fuse_results(self, traditional: NLUResult, llm: NLUResult) -> NLUResult:
        """结果融合 - 加权融合"""
        # 传统系统权重更高（更确定、更可审计）
        if traditional.intent == llm.intent:
            return NLUResult(
                intent=traditional.intent,
                confidence=max(traditional.confidence, llm.confidence) * 1.1,
                slots=self._merge_slots(traditional.slots, llm.slots)
            )
        else:
            # 意图不一致，选择置信度更高的
            if llm.confidence > traditional.confidence + 0.2:
                return llm
            return traditional

class HybridRouter:
    """混合路由引擎 - 核心决策组件"""
    
    def __init__(self, config: RouterConfig):
        self.intent_classifier = IntentClassifier()
        self.confidence_thresholds = {
            "traditional": 0.85,    # 高于此值走传统DM
            "llm_direct": 0.6,      # 高于此值走LLM
            "clarify": 0.4,         # 低于此值需要澄清
        }
        self.intent_mapping = self._load_intent_mapping()
    
    async def route(self, request: UserRequest) -> RouteDecision:
        # 1. 意图分类
        nlu_result = await self.nlu.understand(request.text, request.context)
        
        # 2. 规则优先（安全敏感操作必须走传统DM）
        if nlu_result.intent in SENSITIVE_INTENTS:
            return RouteDecision(
                target=Target.TRADITIONAL,
                reason="sensitive_operation",
                nlu_result=nlu_result
            )
        
        # 3. 意图路由
        route_info = self.intent_mapping.get(nlu_result.intent)
        if route_info:
            return RouteDecision(
                target=route_info.target,
                reason=f"intent_mapping:{nlu_result.intent}",
                nlu_result=nlu_result
            )
        
        # 4. 置信度路由
        if nlu_result.confidence >= self.confidence_thresholds["traditional"]:
            return RouteDecision(
                target=Target.TRADITIONAL,
                reason="high_confidence",
                nlu_result=nlu_result
            )
        elif nlu_result.confidence >= self.confidence_thresholds["llm_direct"]:
            return RouteDecision(
                target=Target.LLM,
                reason="medium_confidence",
                nlu_result=nlu_result
            )
        else:
            return RouteDecision(
                target=Target.CLARIFY,
                reason="low_confidence",
                nlu_result=nlu_result
            )

# 统一工具定义格式（兼容OpenAI Function Calling和OpenAPI）
tools:
  - name: "search_weather"
    description: "查询指定城市的天气信息"
    parameters:
      type: "object"
      properties:
        city:
          type: "string"
          description: "城市名称，如 Lagos, Nairobi"
        date:
          type: "string"
          format: "date"
          description: "日期，格式 YYYY-MM-DD，默认为今天"
      required: ["city"]
    
    execution:
      endpoint: "https://api.transsion-ai.com/weather"
      method: "GET"
      timeout: 3000  # ms
      auth: "api_key"
      rate_limit: 100/minute
      cache_ttl: 1800  # 30分钟缓存
    
    response_schema:
      temperature: "number"
      humidity: "number"
      condition: "string"
      forecast: "array"

class LLMOptimizer:
    """LLM推理优化 - 降低延迟和成本"""
    
    def __init__(self):
        # 1. vLLM推理引擎（PagedAttention显存优化）
        self.engine = vLLM(
            model="qwen2.5-7b",
            tensor_parallel_size=2,
            max_num_seqs=256,
            gpu_memory_utilization=0.9
        )
        
        # 2. 语义缓存（相似问题直接返回缓存结果）
        self.semantic_cache = SemanticCache(
            embedding_model="text-embedding-3-small",
            similarity_threshold=0.95,
            ttl=3600
        )
        
        # 3. 动态批处理（合并相似请求）
        self.batch_processor = DynamicBatcher(
            max_batch_size=32,
            max_wait_ms=50
        )
    
    async def generate(self, prompt: str) -> str:
        # 语义缓存命中检查
        cached = await self.semantic_cache.lookup(prompt)
        if cached:
            return cached
        
        # 动态批处理
        result = await self.batch_processor.add(prompt)
        
        # 写入缓存
        await self.semantic_cache.store(prompt, result)
        
        return result

# 场景："帮我导航到最近的加油站，路上播放周杰伦的歌"
plan = TaskPlan(
    goal="导航+播放音乐",
    steps=[
        # 步骤1：意图分解（LLM生成）
        Step(
            id=1,
            action="find_nearest",
            params={"category": "gas_station"},
            tool="location_service"
        ),
        # 步骤2：获取路线
        Step(
            id=2,
            action="get_route",
            params={"destination": "step1.result"},
            tool="navigation_service",
            depends_on=[1]
        ),
        # 步骤3：搜索音乐
        Step(
            id=3,
            action="search_music",
            params={"artist": "周杰伦"},
            tool="music_service"
        ),
        # 步骤4：并行执行：开始导航 + 播放音乐
        Step(
            id=4,
            action="execute_parallel",
            children=[
                Step(action="start_navigation", params={"route": "step2.result"}),
                Step(action="play_music", params={"playlist": "step3.result"})
            ],
            depends_on=[2, 3]
        )
    ]
)

builtin_tools:
  - name: "transsion_asr"
    description: "传音语音识别服务"
    endpoint: "grpc://ai.transsion.com/asr"
    
  - name: "transsion_tts"
    description: "传音语音合成服务"
    endpoint: "grpc://ai.transsion.com/tts"
    
  - name: "transsion_nlu"
    description: "传音自然语言理解"
    endpoint: "grpc://ai.transsion.com/nlu"
    
  - name: "transsion_sentiment"
    description: "传音舆情分析"
    endpoint: "grpc://ai.transsion.com/sentiment"