1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# llm_gateway.py — 统一 LLM 调用网关
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional, AsyncIterator
import asyncio, tiktoken

@dataclass
class LLMConfig:
    provider: str          # "openai" | "anthropic" | "qwen" | "local"
    model: str
    temperature: float = 0.7
    max_tokens: int = 2048
    timeout: int = 30
    retry_times: int = 3

@dataclass  
class LLMResponse:
    content: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: float
    model: str

class BaseLLMClient(ABC):
    def __init__(self, config: LLMConfig):
        self.config = config

    @abstractmethod
    async def chat(self, messages: list[dict]) -> LLMResponse:
        pass

    @abstractmethod
    async def stream_chat(self, messages: list[dict]) -> AsyncIterator[str]:
        pass

class OpenAIClient(BaseLLMClient):
    """OpenAI / Azure OpenAI 统一客户端"""
    async def chat(self, messages: list[dict]) -> LLMResponse:
        import time
        from openai import AsyncOpenAI
        client = AsyncOpenAI()  # Key 从环境变量读取
        start = time.time()
        for attempt in range(self.config.retry_times):
            try:
                resp = await client.chat.completions.create(
                    model=self.config.model,
                    messages=messages,
                    temperature=self.config.temperature,
                    max_tokens=self.config.max_tokens,
                )
                return LLMResponse(
                    content=resp.choices[0].message.content,
                    prompt_tokens=resp.usage.prompt_tokens,
                    completion_tokens=resp.usage.completion_tokens,
                    latency_ms=(time.time() - start) * 1000,
                    model=self.config.model,
                )
            except Exception as e:
                if attempt == self.config.retry_times - 1:
                    raise
                await asyncio.sleep(2 ** attempt)  # 指数退避

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

class LLMRouter:
    """根据任务类型、成本预算、延迟 SLA 自动路由到最合适的模型"""
    
    ROUTING_RULES = {
        "simple_qa":       {"provider": "openai",    "model": "gpt-4o-mini"},
        "complex_reason":  {"provider": "anthropic", "model": "claude-sonnet-4-20250514"},
        "code_gen":        {"provider": "openai",    "model": "gpt-4o"},
        "chinese_nlp":     {"provider": "qwen",      "model": "qwen-max"},
        "sensitive_data":  {"provider": "local",     "model": "qwen2.5-72b"},  # 数据不出域
    }
    
    def route(self, task_type: str, budget_usd: float) -> LLMConfig:
        rule = self.ROUTING_RULES.get(task_type, self.ROUTING_RULES["simple_qa"])
        return LLMConfig(**rule)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# vllm-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: qwen2-72b-vllm
spec:
  replicas: 2
  template:
    spec:
      containers:
      - name: vllm
        image: vllm/vllm-openai:latest
        args:
          - "--model=/models/Qwen2.5-72B-Instruct"
          - "--tensor-parallel-size=4"
          - "--max-model-len=32768"
          - "--gpu-memory-utilization=0.92"
          - "--enable-prefix-caching"          # KV Cache 复用，降低重复 Prompt 成本
          - "--max-num-seqs=256"
          - "--served-model-name=qwen2.5-72b"
        resources:
          limits:
            nvidia.com/gpu: "4"
            memory: "200Gi"
        volumeMounts:
        - name: model-storage
          mountPath: /models

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

class ReActAgent:
    """
    ReAct = Reasoning + Acting
    核心循环：Thought → Action → Observation → Thought ...
    """
    MAX_ITERATIONS = 10  # 防止无限循环
    
    def __init__(self, llm: BaseLLMClient, tools: list):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
    
    async def run(self, user_query: str) -> str:
        scratchpad = []
        
        for i in range(self.MAX_ITERATIONS):
            # 1. 思考：让 LLM 决定下一步
            thought_prompt = self._build_prompt(user_query, scratchpad)
            response = await self.llm.chat(thought_prompt)
            
            # 2. 解析 Action
            action = self._parse_action(response.content)
            
            # 3. 终止条件
            if action.type == "FINISH":
                return action.answer
            
            # 4. 执行工具
            tool = self.tools.get(action.tool_name)
            if not tool:
                observation = f"错误：工具 {action.tool_name} 不存在"
            else:
                observation = await tool.execute(action.tool_input)
            
            # 5. 记录到 scratchpad
            scratchpad.append({
                "thought": response.content,
                "action": action,
                "observation": observation,
            })
        
        return "已达最大迭代次数，任务未完成"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

SYSTEM_PROMPT_TEMPLATE = """
# 角色定义
你是 {role_name}，{role_description}

# 能力边界
你具备以下能力：
{capabilities}

你不应该做以下事情：
{constraints}

# 输出规范
- 语言：{output_language}
- 格式：{output_format}
- 长度限制：{max_length}

# 可用工具
{tools_description}

# 思考框架
在回答前，请按以下步骤思考：
1. 理解用户意图
2. 确认所需信息是否充分
3. 选择最优解决路径
4. 验证输出是否符合约束

# 当前上下文
{context}
"""

class PromptBuilder:
    """结构化 Prompt 构建器，确保一致性"""
    
    def build_cot_prompt(self, task: str) -> str:
        """思维链（Chain of Thought）Prompt"""
        return f"""请解决以下任务，并展示完整的思考过程：

任务：{task}

请按照以下格式回答：
<thinking>
[在这里写出你的分步思考过程]
</thinking>

<answer>
[最终答案]
</answer>"""

    def build_few_shot_prompt(self, task: str, examples: list[dict]) -> str:
        """少样本（Few-Shot）Prompt"""
        examples_text = "\n\n".join([
            f"示例 {i+1}:\n输入：{e['input']}\n输出：{e['output']}"
            for i, e in enumerate(examples)
        ])
        return f"{examples_text}\n\n现在请处理：\n输入：{task}\n输出："

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

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

# 基础 Chain
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 1. 简单问答链
qa_chain = (
    ChatPromptTemplate.from_template("请回答：{question}")
    | llm
    | StrOutputParser()
)

# 2. 并行处理链（同时执行多个子任务）
parallel_chain = RunnableParallel(
    summary=ChatPromptTemplate.from_template("请总结：{text}") | llm | StrOutputParser(),
    keywords=ChatPromptTemplate.from_template("提取关键词（JSON格式）：{text}") | llm | JsonOutputParser(),
    sentiment=ChatPromptTemplate.from_template("分析情感（正面/负面/中立）：{text}") | llm | StrOutputParser(),
)

# 3. 带路由的条件链
from langchain_core.runnables import RunnableBranch

routing_chain = RunnableBranch(
    (lambda x: "代码" in x["question"], code_chain),
    (lambda x: "数据" in x["question"], analysis_chain),
    default_chain,  # 默认链
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 自定义工具规范
from langchain_core.tools import tool
from pydantic import BaseModel, Field

class DatabaseQueryInput(BaseModel):
    """数据库查询工具的输入 Schema，Pydantic 严格校验"""
    sql: str = Field(description="要执行的 SQL 查询语句，只允许 SELECT")
    database: str = Field(description="目标数据库名称", default="production")
    limit: int = Field(description="返回行数上限", default=100, le=1000)

@tool("database_query", args_schema=DatabaseQueryInput)
async def database_query(sql: str, database: str, limit: int) -> str:
    """安全地执行数据库只读查询，返回结果的 JSON 字符串"""
    # 安全检查：禁止非 SELECT 语句
    if not sql.strip().upper().startswith("SELECT"):
        return "错误：只允许 SELECT 查询"
    
    # 注入防护：使用参数化查询
    # ... 实际数据库连接逻辑
    return json.dumps(results, ensure_ascii=False)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 1. 定义全局状态 Schema（强类型）
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]   # 消息历史（自动追加）
    user_query: str
    plan: list[str]                            # 任务计划
    current_step: int
    tool_results: dict
    final_answer: str
    error_count: int                           # 错误计数，用于熔断

# 2. 定义节点函数（每个节点是纯函数）
async def planner_node(state: AgentState) -> dict:
    """规划节点：分解用户任务为子步骤"""
    plan = await llm.plan(state["user_query"])
    return {"plan": plan, "current_step": 0}

async def executor_node(state: AgentState) -> dict:
    """执行节点：执行当前步骤"""
    current_task = state["plan"][state["current_step"]]
    result = await execute_task(current_task)
    return {
        "tool_results": {**state["tool_results"], current_task: result},
        "current_step": state["current_step"] + 1,
    }

async def validator_node(state: AgentState) -> dict:
    """验证节点：检查执行结果质量"""
    is_valid = await validate_result(state["tool_results"])
    if not is_valid:
        return {"error_count": state["error_count"] + 1}
    return {}

async def synthesizer_node(state: AgentState) -> dict:
    """综合节点：整合所有结果，生成最终答案"""
    answer = await synthesize(state["tool_results"], state["user_query"])
    return {"final_answer": answer}

# 3. 路由函数（决定下一个节点）
def should_continue(state: AgentState) -> str:
    if state["error_count"] >= 3:
        return "error_handler"           # 熔断
    if state["current_step"] >= len(state["plan"]):
        return "synthesizer"             # 所有步骤完成
    return "executor"                    # 继续执行

# 4. 构建图
workflow = StateGraph(AgentState)

workflow.add_node("planner",      planner_node)
workflow.add_node("executor",     executor_node)
workflow.add_node("validator",    validator_node)
workflow.add_node("synthesizer",  synthesizer_node)
workflow.add_node("error_handler", error_handler_node)

workflow.set_entry_point("planner")
workflow.add_edge("planner",   "executor")
workflow.add_edge("executor",  "validator")
workflow.add_conditional_edges("validator", should_continue)
workflow.add_edge("synthesizer", END)

agent = workflow.compile(checkpointer=MemorySaver())  # 开启持久化

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 在敏感操作前暂停，等待人工审批
from langgraph.graph import interrupt

async def approval_gate_node(state: AgentState) -> dict:
    """高风险操作前的人工审批节点"""
    if state["risk_level"] == "HIGH":
        # 暂停图执行，返回给前端等待审批
        human_decision = interrupt({
            "message": "检测到高风险操作，请审批",
            "operation": state["pending_operation"],
            "risk_reason": state["risk_reason"],
        })
        if human_decision["approved"]:
            return {"approved": True}
        else:
            return {"approved": False, "cancel_reason": human_decision["reason"]}
    return {"approved": True}  # 低风险直接通过

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

class AgenticRAGSystem:
    """
    Agentic RAG：Agent 主动控制检索过程
    核心能力：
    - 查询分解与改写
    - 自适应检索策略选择
    - 检索结果质量评估
    - 迭代式信息补充
    """
    
    def __init__(self, vector_store, bm25_index, llm):
        self.vector_store = vector_store      # 语义检索
        self.bm25_index = bm25_index          # 关键词检索
        self.llm = llm
        self.reranker = CrossEncoderReranker()
    
    async def retrieve_with_reflection(
        self, 
        query: str, 
        max_iterations: int = 3
    ) -> list[Document]:
        """带自我反思的迭代检索"""
        
        all_docs = []
        current_query = query
        
        for iteration in range(max_iterations):
            # 1. 查询改写（第一轮不改写）
            if iteration > 0:
                current_query = await self._rewrite_query(
                    original=query,
                    retrieved_docs=all_docs,
                    missing_info=missing_info,
                )
            
            # 2. 混合检索
            semantic_docs = await self.vector_store.similarity_search(
                current_query, k=10
            )
            keyword_docs = await self.bm25_index.search(current_query, k=10)
            
            # 3. 融合与重排（RRF + Cross-Encoder）
            merged = self._reciprocal_rank_fusion(semantic_docs, keyword_docs)
            reranked = await self.reranker.rerank(current_query, merged, top_k=5)
            all_docs.extend(reranked)
            
            # 4. 评估是否信息充分
            sufficiency = await self._assess_sufficiency(query, all_docs)
            if sufficiency.is_sufficient:
                break
            missing_info = sufficiency.missing_aspects
        
        # 5. 去重与压缩
        return await self._deduplicate_and_compress(all_docs, query)
    
    async def _assess_sufficiency(self, query: str, docs: list) -> SufficiencyResult:
        """评估检索到的文档是否足以回答问题"""
        prompt = f"""
        用户问题：{query}
        已检索文档摘要：{self._summarize_docs(docs)}
        
        请评估：
        1. 当前文档能否完整回答问题？（是/否）
        2. 如果不能，缺少哪些方面的信息？
        
        以 JSON 格式返回：{{"is_sufficient": bool, "missing_aspects": [str]}}
        """
        result = await self.llm.chat([{"role": "user", "content": prompt}])
        return SufficiencyResult(**json.loads(result.content))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

# tools/base.py — 工具基类（所有工具必须继承）
from abc import ABC, abstractmethod
from dataclasses import dataclass
import logging, asyncio

@dataclass
class ToolResult:
    success: bool
    data: any
    error: str = ""
    execution_time_ms: float = 0.0

class EnterpriseBaseTool(ABC):
    """企业工具基类，内置安全校验、限流、日志"""
    
    # 子类必须声明
    name: str
    description: str
    rate_limit_per_minute: int = 60
    
    def __init_subclass__(cls):
        assert hasattr(cls, 'name'), f"{cls.__name__} 必须声明 name"
        assert hasattr(cls, 'description'), f"{cls.__name__} 必须声明 description"
    
    @abstractmethod
    async def _execute(self, **kwargs) -> any:
        """子类实现具体逻辑"""
        pass
    
    async def run(self, **kwargs) -> ToolResult:
        """统一执行入口，包含安全防护"""
        import time
        
        # 限流检查
        if not await self._check_rate_limit():
            return ToolResult(success=False, error="调用频率超限，请稍后重试")
        
        # 参数校验
        validated = self._validate_inputs(kwargs)
        if not validated.ok:
            return ToolResult(success=False, error=validated.error)
        
        start = time.time()
        try:
            result = await asyncio.wait_for(
                self._execute(**kwargs), 
                timeout=30.0
            )
            return ToolResult(
                success=True, 
                data=result,
                execution_time_ms=(time.time() - start) * 1000,
            )
        except asyncio.TimeoutError:
            return ToolResult(success=False, error="工具执行超时（30s）")
        except Exception as e:
            logging.error(f"Tool {self.name} error: {e}", exc_info=True)
            return ToolResult(success=False, error=str(e))

# 工具注册表
TOOL_REGISTRY: dict[str, EnterpriseBaseTool] = {}

def register_tool(tool_class):
    """装饰器：自动注册工具"""
    instance = tool_class()
    TOOL_REGISTRY[instance.name] = instance
    return tool_class

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

class AgentMemorySystem:
    """三层记忆系统统一管理"""
    
    def __init__(self, vector_db, graph_db, redis_client):
        self.vector_db = vector_db          # Pinecone / Weaviate / Qdrant
        self.graph_db = graph_db            # Neo4j
        self.redis = redis_client           # 短期缓存
        self.summarizer = ConversationSummarizer()
    
    # ===== 工作记忆（短期）=====
    
    def get_working_memory(self, session_id: str, max_tokens: int = 4000) -> list:
        """获取当前会话上下文，自动截断至 Token 限制"""
        messages = self.redis.lrange(f"session:{session_id}", 0, -1)
        # 从最新消息开始，向前取直到 Token 用尽
        return self._truncate_to_token_limit(messages, max_tokens)
    
    async def add_to_working_memory(self, session_id: str, message: dict):
        """添加消息，超过阈值自动触发摘要压缩"""
        self.redis.rpush(f"session:{session_id}", json.dumps(message))
        self.redis.expire(f"session:{session_id}", 3600)  # 1小时 TTL
        
        # 消息超过 20 条时压缩
        if self.redis.llen(f"session:{session_id}") > 20:
            await self._compress_to_episodic(session_id)
    
    # ===== 情景记忆（中期）=====
    
    async def _compress_to_episodic(self, session_id: str):
        """将工作记忆压缩为摘要存入向量DB"""
        messages = self.redis.lrange(f"session:{session_id}", 0, -1)
        summary = await self.summarizer.summarize(messages)
        
        # 存入向量数据库
        await self.vector_db.upsert(
            id=f"episode:{session_id}:{int(time.time())}",
            values=await self._embed(summary),
            metadata={
                "session_id": session_id,
                "timestamp": time.time(),
                "summary": summary,
                "type": "episode",
            }
        )
        
        # 清空工作记忆的旧消息
        self.redis.ltrim(f"session:{session_id}", -5, -1)  # 只保留最近 5 条
    
    async def recall_episodes(self, query: str, user_id: str, top_k: int = 3) -> list:
        """语义检索相关历史情节"""
        query_embedding = await self._embed(query)
        results = await self.vector_db.query(
            vector=query_embedding,
            top_k=top_k,
            filter={"user_id": user_id, "type": "episode"},
        )
        return [r.metadata["summary"] for r in results]
    
    # ===== 语义记忆（长期）=====
    
    async def update_user_profile(self, user_id: str, interaction: dict):
        """从交互中提取并更新用户画像"""
        extracted = await self._extract_preferences(interaction)
        
        # 写入图数据库
        await self.graph_db.run("""
            MERGE (u:User {id: $user_id})
            SET u.preferences = $preferences,
                u.updated_at = timestamp()
            WITH u
            UNWIND $entities as entity
            MERGE (e:Entity {name: entity.name, type: entity.type})
            MERGE (u)-[:KNOWS]->(e)
        """, user_id=user_id, 
            preferences=extracted.preferences,
            entities=extracted.entities)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

class ContextWindowManager:
    """
    精确管理 LLM 上下文窗口，最大化有效信息密度
    遵循 Harness 方法论：结构化、动态、可追溯
    """
    
    CONTEXT_BUDGET = {
        "system_prompt":     0.15,   # 系统提示 15%
        "memory_retrieval":  0.20,   # 记忆检索 20%
        "rag_documents":     0.35,   # RAG 文档 35%
        "conversation":      0.20,   # 对话历史 20%
        "current_query":     0.05,   # 当前问题 5%
        "output_reserve":    0.05,   # 输出预留 5%
    }
    
    def build_context(
        self,
        system_prompt: str,
        user_query: str,
        retrieved_docs: list,
        conversation_history: list,
        memory_summaries: list,
        total_token_limit: int = 128000,
    ) -> list[dict]:
        
        budgets = {k: int(v * total_token_limit) 
                   for k, v in self.CONTEXT_BUDGET.items()}
        
        # 1. 系统提示（最高优先级）
        system = self._truncate(system_prompt, budgets["system_prompt"])
        
        # 2. 记忆摘要（注入到 system 结尾）
        memory_text = self._format_memories(memory_summaries, budgets["memory_retrieval"])
        
        # 3. RAG 文档（重排后按相关性截断）
        docs_text = self._format_docs(retrieved_docs, budgets["rag_documents"])
        
        # 4. 对话历史（从最近往前截断）
        history = self._truncate_history(conversation_history, budgets["conversation"])
        
        # 5. 组装 messages
        messages = [
            {"role": "system", "content": f"{system}\n\n## 用户记忆\n{memory_text}"},
        ]
        
        if docs_text:
            messages.append({
                "role": "user",
                "content": f"## 参考文档\n{docs_text}"
            })
            messages.append({"role": "assistant", "content": "已读取参考文档。"})
        
        messages.extend(history)
        messages.append({"role": "user", "content": user_query})
        
        return messages

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class HybridRetriever:
    """
    混合检索 = 语义检索 + 关键词检索 + 结构化检索
    使用 RRF (Reciprocal Rank Fusion) 融合
    """
    
    async def retrieve(self, query: str, k: int = 20) -> list[Document]:
        # 并行执行三路检索
        semantic, keyword, structured = await asyncio.gather(
            self._semantic_search(query, k),      # 向量相似度
            self._bm25_search(query, k),          # BM25 关键词
            self._structured_filter(query, k),    # 元数据过滤
        )
        
        # RRF 融合（k=60 是经验最优值）
        scores = {}
        for rank, doc in enumerate(semantic):
            scores[doc.id] = scores.get(doc.id, 0) + 1/(60 + rank)
        for rank, doc in enumerate(keyword):
            scores[doc.id] = scores.get(doc.id, 0) + 1/(60 + rank)
        for rank, doc in enumerate(structured):
            scores[doc.id] = scores.get(doc.id, 0) + 1/(60 + rank)
        
        # 按融合分数排序，取 top-k
        all_docs = {d.id: d for d in semantic + keyword + structured}
        sorted_ids = sorted(scores, key=scores.get, reverse=True)[:k]
        return [all_docs[id] for id in sorted_ids]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

from ragas import evaluate
from ragas.metrics import (
    faithfulness,           # 忠实度：答案是否基于文档
    answer_relevancy,       # 相关性：答案是否回答了问题
    context_precision,      # 精确率：检索文档是否相关
    context_recall,         # 召回率：关键信息是否被检索到
)

# 自动化评估流水线
async def evaluate_rag_pipeline(test_dataset):
    results = evaluate(
        dataset=test_dataset,
        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    )
    
    # SLA 门限：所有指标 >= 0.8 才允许发布
    thresholds = {
        "faithfulness": 0.85,
        "answer_relevancy": 0.80,
        "context_precision": 0.75,
        "context_recall": 0.80,
    }
    
    failures = [m for m, threshold in thresholds.items() 
                if results[m] < threshold]
    
    if failures:
        raise ValueError(f"RAG 质量未达标: {failures}")
    
    return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

from dataclasses import dataclass
from enum import Enum
import uuid

class MessageType(Enum):
    TASK_ASSIGN   = "task_assign"    # 分配任务
    TASK_RESULT   = "task_result"    # 返回结果
    REQUEST_HELP  = "request_help"   # 请求协助
    STATUS_UPDATE = "status_update"  # 进度更新
    ERROR         = "error"          # 错误报告

@dataclass
class AgentMessage:
    """Agent 间通信的标准消息格式"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    from_agent: str = ""
    to_agent: str = ""
    type: MessageType = MessageType.TASK_ASSIGN
    content: dict = field(default_factory=dict)
    priority: int = 5                # 1-10，10最高
    timeout_seconds: int = 120
    timestamp: float = field(default_factory=time.time)
    parent_task_id: str = ""         # 追溯任务链

class MultiAgentOrchestrator:
    """多 Agent 编排器"""
    
    def __init__(self, agents: dict[str, BaseAgent]):
        self.agents = agents
        self.message_queue = asyncio.PriorityQueue()
        self.task_tracker = {}
    
    async def execute_complex_task(self, task: str) -> str:
        """编排多个 Agent 协作完成复杂任务"""
        
        # 1. 任务分解
        subtasks = await self.agents["planner"].decompose(task)
        
        # 2. 构建依赖图（DAG）
        dag = self._build_dependency_dag(subtasks)
        
        # 3. 并行执行无依赖的任务
        results = {}
        for batch in dag.topological_batches():
            batch_tasks = [
                self._dispatch_to_agent(subtask, results)
                for subtask in batch
            ]
            batch_results = await asyncio.gather(*batch_tasks)
            results.update(dict(zip(batch, batch_results)))
        
        # 4. 综合所有结果
        return await self.agents["synthesizer"].synthesize(task, results)
    
    async def _dispatch_to_agent(
        self, subtask: SubTask, context: dict
    ) -> str:
        """将子任务路由到最合适的 Agent"""
        # 基于任务类型和 Agent 能力匹配
        best_agent = self._match_agent(subtask.required_skills)
        
        message = AgentMessage(
            from_agent="orchestrator",
            to_agent=best_agent.name,
            type=MessageType.TASK_ASSIGN,
            content={"task": subtask.description, "context": context},
            priority=subtask.priority,
        )
        
        return await best_agent.handle_message(message)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# Dockerfile — 生产级多阶段构建
FROM python:3.12-slim AS builder
WORKDIR /app

# 安全：非 root 用户
RUN groupadd -r agent && useradd -r -g agent agent

# 依赖安装（利用 Docker 缓存层）
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

FROM python:3.12-slim AS runtime
WORKDIR /app

# 仅复制运行时依赖
COPY --from=builder /root/.local /root/.local
COPY --from=builder /app /app
COPY src/ ./src/

# 安全配置
RUN chown -R agent:agent /app
USER agent
ENV PATH=/root/.local/bin:$PATH

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s \
    CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

# k8s/agent-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent-service
  namespace: production
  labels:
    app: ai-agent
    version: v1.2.0
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0          # 零停机更新
  selector:
    matchLabels:
      app: ai-agent
  template:
    spec:
      containers:
      - name: agent
        image: registry.company.com/ai-agent:v1.2.0
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:        # 绝不在代码或 ConfigMap 中存储密钥
              name: llm-secrets
              key: openai-api-key
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
        readinessProbe:
          httpGet:
            path: /health/ready
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health/live
            port: 8000
          initialDelaySeconds: 60
          periodSeconds: 30
      affinity:
        podAntiAffinity:
          requiredDuringScheduling:
            - labelSelector:
                matchLabels:
                  app: ai-agent
              topologyKey: kubernetes.io/hostname   # 跨节点分散
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ai-agent-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ai-agent-service
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Pods
    pods:
      metric:
        name: http_requests_per_second   # 基于业务指标扩缩容
      target:
        type: AverageValue
        averageValue: "100"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# tracing/setup.py
from langsmith import Client, traceable
import os

# 全局追踪配置
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-agent-v1"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

langsmith_client = Client()

# 自定义追踪装饰器
def agent_trace(name: str, tags: list[str] = None):
    """业务层追踪装饰器"""
    def decorator(func):
        @traceable(
            name=name,
            tags=tags or [],
            metadata={"env": os.getenv("APP_ENV", "production")},
        )
        async def wrapper(*args, **kwargs):
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# 使用示例
@agent_trace(name="customer_service_agent", tags=["customer_service", "rag"])
async def handle_customer_query(query: str, user_id: str) -> str:
    # Agent 逻辑
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

# evaluation/pipeline.py
class AutoEvaluationPipeline:
    """生产环境自动化质量评估"""
    
    def __init__(self):
        self.langsmith = Client()
        self.evaluators = [
            self._evaluate_correctness,
            self._evaluate_faithfulness,
            self._evaluate_toxicity,
            self._evaluate_latency,
        ]
    
    async def run_daily_evaluation(self):
        """每日自动从生产日志中抽样评估"""
        
        # 1. 从 LangSmith 获取昨日生产数据（随机抽样 5%）
        runs = self.langsmith.list_runs(
            project_name="production-agent-v1",
            start_time=yesterday(),
            filter='and(gt(total_tokens, 0), eq(error, null))',
            limit=500,
        )
        
        # 2. 并行评估
        results = await asyncio.gather(*[
            self._evaluate_run(run) for run in runs
        ])
        
        # 3. 聚合指标
        metrics = self._aggregate_metrics(results)
        
        # 4. 质量告警
        if metrics["correctness"] < 0.85:
            await self._send_alert(
                severity="HIGH",
                message=f"正确率下降至 {metrics['correctness']:.2%}，低于阈值 85%",
                metrics=metrics,
            )
        
        # 5. 生成日报
        return self._generate_daily_report(metrics)
    
    async def _evaluate_correctness(self, run) -> float:
        """使用 LLM 评估答案正确性（LLM-as-Judge）"""
        prompt = f"""
        问题：{run.inputs['query']}
        Agent 答案：{run.outputs['answer']}
        参考答案：{run.reference_output}
        
        请评分（0-1），并给出理由。只返回 JSON：
        {{"score": float, "reason": str}}
        """
        result = await evaluator_llm.chat([{"role": "user", "content": prompt}])
        return json.loads(result.content)["score"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

# monitoring/metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 业务指标
AGENT_REQUESTS_TOTAL = Counter(
    'agent_requests_total',
    'Agent 总请求数',
    ['agent_type', 'status', 'model']
)

AGENT_LATENCY_SECONDS = Histogram(
    'agent_latency_seconds',
    'Agent 端到端延迟',
    ['agent_type'],
    buckets=[0.5, 1, 2, 5, 10, 30, 60]  # 定义 SLA 分位桶
)

LLM_TOKENS_TOTAL = Counter(
    'llm_tokens_total',
    'LLM Token 消耗',
    ['model', 'type']  # type: prompt/completion
)

LLM_COST_USD = Counter(
    'llm_cost_usd_total',
    'LLM 美元成本',
    ['model']
)

ACTIVE_SESSIONS = Gauge(
    'agent_active_sessions',
    '当前活跃会话数'
)

MEMORY_RETRIEVAL_LATENCY = Histogram(
    'memory_retrieval_latency_seconds',
    '记忆检索延迟',
    ['memory_type']  # working/episodic/semantic
)

# 中间件注入
class MetricsMiddleware:
    async def __call__(self, request, call_next):
        start = time.time()
        try:
            response = await call_next(request)
            AGENT_REQUESTS_TOTAL.labels(
                agent_type=request.headers.get("X-Agent-Type", "default"),
                status="success",
                model=request.headers.get("X-Model", "unknown")
            ).inc()
            return response
        except Exception as e:
            AGENT_REQUESTS_TOTAL.labels(status="error").inc()
            raise
        finally:
            AGENT_LATENCY_SECONDS.labels(
                agent_type=request.headers.get("X-Agent-Type", "default")
            ).observe(time.time() - start)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# prometheus/alert_rules.yaml
groups:
- name: ai_agent_sla
  rules:
  
  # P99 延迟告警
  - alert: AgentHighLatency
    expr: histogram_quantile(0.99, agent_latency_seconds) > 10
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Agent P99 延迟超过 10 秒"
      
  # 错误率告警
  - alert: AgentHighErrorRate
    expr: rate(agent_requests_total{status="error"}[5m]) / rate(agent_requests_total[5m]) > 0.05
    for: 2m
    labels:
      severity: critical
    annotations:
      summary: "Agent 错误率超过 5%"
      
  # Token 成本告警（每小时超 $50）
  - alert: HighLLMCost
    expr: increase(llm_cost_usd_total[1h]) > 50
    labels:
      severity: warning
    annotations:
      summary: "过去 1 小时 LLM 成本超过 $50"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 基于 feature flag 的灰度控制
class GradualRollout:
    ROLLOUT_STAGES = [
        {"name": "canary",     "percentage": 5,   "duration_hours": 24},
        {"name": "early",      "percentage": 20,  "duration_hours": 48},
        {"name": "majority",   "percentage": 50,  "duration_hours": 48},
        {"name": "full",       "percentage": 100, "duration_hours": 0},
    ]
    
    def should_use_new_version(self, user_id: str) -> bool:
        current_stage = self._get_current_stage()
        # 基于 user_id hash 确保同一用户体验一致
        user_hash = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
        return user_hash < current_stage["percentage"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class PromptInjectionDefender:
    """防止恶意用户通过 Prompt 注入攻击绕过系统限制"""
    
    INJECTION_PATTERNS = [
        r"ignore (all |previous |above )?instructions",
        r"you are now (a |an )?.*(?:without|ignore|bypass)",
        r"(pretend|act|roleplay|simulate) (you are|as if)",
        r"system prompt|jailbreak|DAN|do anything now",
    ]
    
    def scan(self, user_input: str) -> ScanResult:
        for pattern in self.INJECTION_PATTERNS:
            if re.search(pattern, user_input, re.IGNORECASE):
                return ScanResult(
                    is_safe=False,
                    threat_type="prompt_injection",
                    matched_pattern=pattern,
                )
        return ScanResult(is_safe=True)