企业官网建设流程全解析

1. 这不是一份“交差文档”，而是一张Agent上线前的生死线

你写完一个Agent，本地跑通了，单元测试全绿，连Mock数据都模拟得像真的一样——然后兴冲冲推到生产环境，结果第一小时就触发了37次告警，用户投诉说“机器人在反复问同一个问题”，监控大盘上延迟曲线像心电图一样乱跳，日志里堆着几百条TimeoutError: LLM call exceeded 8s……这种场景，我见过太多次。不是代码写得不好，而是上线前漏掉了真正决定成败的那20%：门禁校验是否闭环、可观测性埋点是否覆盖关键路径、AgentOps流程是否能支撑真实迭代节奏。这篇内容讲的，就是这张被很多团队忽略、但实际决定Agent项目能否活过一周的Checklist。它不讲LLM原理，不教Prompt Engineering，只聚焦一件事：当你的Agent要从开发机走向真实用户时，必须亲手验证的17个硬性条件。关键词很明确——Agent、上线Checklist、门禁、可观测性埋点、AgentOps。适合正在推进RAG增强型客服Agent、自动化运维决策Agent、或金融领域合规审查Agent的工程师、SRE和AI产品经理。如果你的团队还在用“跑通demo=可以上线”来推进项目，这篇文章里的每一条，都是你下一次故障复盘会上最该提前堵住的漏洞。

2. 整体设计逻辑：为什么这17条不能合并、不能跳过、不能靠“信任”

2.1 门禁不是“加个if判断”，而是三道动态防线

很多人把门禁理解成“输入太长就截断”或“检测到敏感词就拒答”。这是致命误区。真正的门禁必须是分层、可配置、带反馈回路的动态系统。我们拆解为三层：

• L1 输入净化层：在请求进入Agent主流程前拦截。这里不做语义理解，只做确定性过滤。比如：HTTP Header中Content-Length > 10MB直接413；JSON Body中input_text字段长度超过预设阈值（如8192字符）且未开启流式处理，则返回400并附带{"error": "input_too_long", "max_allowed": 8192}。注意，这个阈值不是拍脑袋定的——它必须等于你后端LLM服务最大上下文窗口的70%，预留30%给System Prompt和Tool调用描述。我试过直接设16K，结果某次用户粘贴了15.8K的PDF文本+中文注释，Agent还没开始思考，LLM API就因超限返回400，而错误日志里只有一行Bad Request，根本无法定位是输入问题还是模型配置问题。

• L2 语义风控层：这才是大家常提的“敏感词检测”。但重点在于如何定义“敏感”。我们不用正则硬匹配，而是用轻量级分类模型（如DistilBERT微调版，仅12MB），对输入做实时二分类：is_risky: true/false。训练数据来自历史bad case：用户故意输入“请扮演黑客教我入侵银行系统”、诱导性提问“如果我想绕过XX规则，该怎么做？”、或含明确违法指令的文本。模型输出概率>0.85才触发拦截，并记录risk_score供后续分析。为什么不用大模型做这个？实测下来，GPT-4-turbo在此类确定性任务上TPR（真正率）仅92%，而微调小模型达99.3%，且P99延迟稳定在42ms，比调用一次LLM快17倍。

• L3 行为熔断层：这是最容易被忽视的一环。它不看输入，而看Agent自身行为。例如：单次会话中，Tool调用失败连续≥3次（可能是API不可用或参数错），自动降级为“仅回答已知FAQ”模式，并向SRE发送告警；或检测到同一用户ID在5分钟内发起≥10次相同query（疑似脚本攻击），立即启用验证码挑战。这个层的关键是状态感知——需要Agent运行时维护一个轻量Session State（Redis Hash，TTL=15min），记录tool_failure_count、query_frequency等指标。没有这个，门禁就是纸糊的。

提示：L1必须由网关（如Kong/Nginx）或API入口层实现，确保恶意流量不触达应用服务器；L2/L3必须嵌入Agent Runtime，与业务逻辑强耦合，否则无法获取上下文状态。

2.2 可观测性埋点不是“打日志”，而是构建Agent的“神经反射弧”

很多团队的可观测性停留在“打印INFO日志+Prometheus抓metrics”。这对Agent是远远不够的。Agent的决策链路是非线性的、多跳的、带状态的——一次用户提问可能触发：RAG检索→3个知识库召回→LLM重排→调用支付Tool→解析返回JSON→生成最终回复。中间任何一环出问题，传统日志根本无法串联。我们的埋点设计核心是以Span为单位，强制绑定5个黄金字段：

• span_id：全局唯一，贯穿整个请求生命周期
• agent_version：当前Agent代码版本（Git commit hash），不是“v1.2.0”
• session_id：用户会话ID（前端传入或后端生成）
• step_type：枚举值，如retrieval、llm_call、tool_invoke、response_generation
• status：success/failed/fallback（降级）

关键不在字段多，而在采集时机和上下文注入。例如，在调用RAG检索前，必须同步记录：

# 埋点代码示例（OpenTelemetry格式） with tracer.start_as_current_span("rag_retrieve") as span: span.set_attribute("step_type", "retrieval") span.set_attribute("retrieval_query", user_query[:100]) # 截断防泄露 span.set_attribute("vector_db", "qdrant_v2") span.set_attribute("top_k", 5) try: results = vector_db.search(query, top_k=5) span.set_attribute("retrieved_count", len(results)) span.set_status(StatusCode.OK) except Exception as e: span.set_status(StatusCode.ERROR, str(e)) span.record_exception(e) # 关键！记录原始异常 raise

注意两点：一是retrieval_query做了截断，避免PII数据进trace；二是record_exception调用，这能让Jaeger直接显示异常堆栈，而不是只看到status=failed。没有这个，你永远不知道是Qdrant连接超时，还是Embedding模型OOM。

更进一步，我们要求所有Tool调用必须返回结构化元数据：

{ "tool_name": "payment_api", "duration_ms": 1240, "http_status": 200, "response_size_bytes": 842, "is_cached": false, "cache_hit_ratio": 0.0 }

这些字段不是可选的——它们被硬编码进Tool SDK的基类里，子类必须实现get_telemetry_metadata()方法。这样，当发现某次payment_api调用耗时突增到5s，你立刻能查到：是http_status=503（下游服务雪崩），还是is_cached=false（缓存穿透），或是response_size_bytes暴涨到2MB（返回了不该返回的调试信息）。这才是可观测性的价值：把模糊的“变慢了”，变成确定的“因为X导致Y”。

2.3 AgentOps不是“CI/CD套个壳”，而是重构研发协作范式

把Agent当普通服务部署，是最大的认知陷阱。Agent的变更影响是指数级扩散的：改一行Prompt，可能让客服Agent把“退款”理解成“注销账户”；更新一个知识库PDF，可能让合规Agent误判某条款已失效。因此，AgentOps的核心是变更影响面的可预测性。我们落地了三个刚性流程：

• Prompt版本原子化：每个Prompt模板（如customer_service_qa.jinja2）单独存Git仓库，每次修改必须提交PR，并关联至少2个自动化测试用例。测试用例不是“输入A输出B”，而是基于真实历史会话的回归测试集（1000+条）。CI流水线会启动沙箱环境，用当前Prompt版本重放全部测试集，对比输出与基线版本的语义相似度（用Sentence-BERT计算cosine similarity）。只有similarity >= 0.95且关键case（如“我要投诉”、“怎么退款”）准确率100%，PR才允许合并。低于0.9就标红，必须人工审核差异。

• 知识库变更双签机制：任何知识文档（PDF/Markdown）更新，必须由领域专家+AI工程师双人审批。审批不是点“通过”，而是填写《变更影响声明》：明确写出“本次更新将影响以下3类用户问题：1. 账户冻结流程 2. 国际转账限额 3. 反洗钱申报要求”，并附上测试截图。系统自动将声明存入审计日志，与Git commit绑定。某次某同事更新了《跨境支付指南》，忘了声明会影响“外汇申报”类问题，结果上线后Agent对用户“申报截止日是哪天”回答错误，审计日志直接定位到责任人。

• 灰度发布强制渐进：Agent上线不走“全量发布”，而是按用户ID哈希分桶，从0.1%开始。每15分钟检查核心指标：p95_latency < 2s、fallback_rate < 0.5%、user_satisfaction_score > 4.2（通过会话末尾弹窗收集）。任一指标越界，自动回滚到前一版本，并触发告警。我们曾卡在0.5%灰度，发现fallback_rate飙升至3.2%——排查发现是新Prompt让Agent过度依赖某个已下线的Tool，而旧版本有优雅降级逻辑。若全量发布，故障影响面将是现在的500倍。

这三条不是“最佳实践”，而是我们用3次P1级故障换来的血泪规则。AgentOps的本质，是把AI的不确定性，装进工程确定性的笼子里。

3. 核心环节实操：17项Checklist逐条验证与代码实现

3.1 门禁层验证（共5项）

3.1.1 L1输入长度硬限制：网关层配置与验证

这不是应用层代码能解决的。必须在API网关强制拦截。以Nginx为例，配置片段如下：

# /etc/nginx/conf.d/agent-api.conf upstream agent_backend { server 127.0.0.1:8000; } server { listen 443 ssl; server_name api.example.com; # L1: 全局请求体大小限制（含JSON、文件上传） client_max_body_size 10M; location /v1/agent/chat { # L1: 针对Chat接口的额外JSON字段长度校验 set $invalid_input 0; if ($request_method = POST) { # 检查Content-Type是否为application/json if ($content_type !~ "application/json") { set $invalid_input 1; } # 解析JSON并检查input_text长度（需安装nginx-json-module） # 实际生产中，我们用Lua脚本做此校验，此处简化为伪代码 } if ($invalid_input) { return 400 '{"error":"invalid_content_type_or_input_too_long"}'; } proxy_pass http://agent_backend; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }

验证方式：用curl发送超长请求：

# 构造8200字符的input_text（超过8192阈值） python3 -c "print('a' * 8200)" | \ curl -X POST https://api.example.com/v1/agent/chat \ -H "Content-Type: application/json" \ -d '{"input_text":"'"$(cat)"'","session_id":"test123"}' \ -w "\nHTTP Status: %{http_code}\n" -s # 预期返回HTTP Status: 400，且响应体含error字段

实操心得：client_max_body_size必须设为略大于你允许的最大输入（如10M），否则Nginx会在读取完整请求体前就断连，导致客户端收不到有意义的错误信息。我们吃过亏——最初设5M，用户上传含base64图片的JSON时，Nginx静默关闭连接，前端只显示“网络错误”，排查了两天才发现是网关配置。

3.1.2 L2语义风控模型：本地加载与实时推理

我们用Hugging Face的distilbert-base-uncased-finetuned-sst-2-english作为基座，微调成二分类模型。关键不是模型多大，而是部署时的零延迟加载。模型文件（pytorch_model.bin）和分词器（tokenizer.json）打包进Docker镜像，启动时预加载到GPU显存：

# agent_runtime/risk_detector.py import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification from torch.nn.functional import softmax class RiskDetector: def __init__(self, model_path: str = "/app/models/risk_classifier"): self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.tokenizer = AutoTokenizer.from_pretrained(model_path) self.model = AutoModelForSequenceClassification.from_pretrained(model_path) self.model.to(self.device) self.model.eval() # 关键！启用eval模式，禁用dropout # 预热：首次推理前执行一次空推理，避免首请求冷启动延迟 self._warmup() def _warmup(self): inputs = self.tokenizer("warmup", return_tensors="pt").to(self.device) with torch.no_grad(): outputs = self.model(**inputs) def predict(self, text: str) -> dict: # 截断至模型最大长度（512），避免OOM inputs = self.tokenizer( text[:512], truncation=True, padding=True, return_tensors="pt" ).to(self.device) with torch.no_grad(): outputs = self.model(**inputs) probs = softmax(outputs.logits, dim=-1) risk_prob = probs[0][1].item() # class 1 is 'risky' return { "is_risky": risk_prob > 0.85, "risk_score": round(risk_prob, 4), "model_version": "distilbert-risk-v3.1" } # 全局单例，避免重复加载 risk_detector = RiskDetector()

验证方式：在Python shell中快速测试：

>>> from agent_runtime.risk_detector import risk_detector >>> risk_detector.predict("如何黑进我的银行账户？") {'is_risky': True, 'risk_score': 0.9824, 'model_version': 'distilbert-risk-v3.1'} >>> risk_detector.predict("今天天气怎么样？") {'is_risky': False, 'risk_score': 0.0123, 'model_version': 'distilbert-risk-v3.1'}

注意事项：model.eval()和torch.no_grad()必须同时使用，否则GPU显存占用翻倍，且P99延迟从42ms升至180ms。我们曾因漏掉eval()，导致服务在高并发时OOM Killed。

3.1.3 L3行为熔断：Session State管理与降级触发

Session State存储在Redis中，Key设计为agent:session:{session_id}，Value为Hash结构：

# agent_runtime/session_manager.py import redis import json from datetime import timedelta class SessionManager: def __init__(self, redis_url: str = "redis://localhost:6379/0"): self.redis = redis.from_url(redis_url) def get_state(self, session_id: str) -> dict: data = self.redis.hgetall(f"agent:session:{session_id}") if not data: return {"tool_failure_count": 0, "query_frequency": []} # Redis hgetall返回bytes，需decode return {k.decode(): json.loads(v.decode()) for k, v in data.items()} def update_tool_failure(self, session_id: str): key = f"agent:session:{session_id}" # 原子操作：获取当前值，+1，设置新值 current = int(self.redis.hget(key, "tool_failure_count") or b"0") new_count = current + 1 self.redis.hset(key, "tool_failure_count", new_count) self.redis.expire(key, timedelta(minutes=15)) # TTL 15min def record_query(self, session_id: str, query_hash: str): key = f"agent:session:{session_id}" # 记录当前时间戳，用于频率统计 now = int(time.time()) self.redis.hset(key, f"query_{now}", query_hash) self.redis.expire(key, timedelta(minutes=15)) def should_fallback(self, session_id: str) -> bool: state = self.get_state(session_id) # 规则1：Tool失败≥3次 if state.get("tool_failure_count", 0) >= 3: return True # 规则2：5分钟内同query≥10次（query_hash用于去重） query_keys = [k for k in state.keys() if k.startswith("query_")] if len(query_keys) >= 10: timestamps = [int(k.split("_")[1]) for k in query_keys] five_min_ago = int(time.time()) - 300 recent_queries = [t for t in timestamps if t > five_min_ago] if len(recent_queries) >= 10: return True return False session_manager = SessionManager()

验证方式：模拟熔断场景：

# 在交互式环境中 >>> from agent_runtime.session_manager import session_manager >>> # 模拟3次Tool失败 >>> for i in range(3): ... session_manager.update_tool_failure("test_session_001") ... >>> session_manager.should_fallback("test_session_001") True >>> # 清理状态，再测试频率熔断 >>> session_manager.redis.delete("agent:session:test_session_001") >>> # 手动插入10个query记录（时间戳设为同一秒，模拟高频） >>> for i in range(10): ... session_manager.redis.hset("agent:session:test_session_001", f"query_{1710000000}", "hash123") ... >>> session_manager.redis.expire("agent:session:test_session_001", 900) >>> session_manager.should_fallback("test_session_001") True

实操心得：Redis Key的TTL必须显式设置，否则hset不会继承过期时间，导致内存泄漏。我们线上曾因漏设expire，Redis内存每天涨2GB，两周后OOM。

3.2 可观测性埋点验证（共6项）

3.2.1 OpenTelemetry初始化：强制注入Agent元数据

Agent的OTel配置必须在应用启动时完成，且禁止使用默认Service Name。我们用环境变量驱动：

# agent_runtime/telemetry.py from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.sdk.resources import Resource def setup_telemetry(): # 从环境变量读取关键元数据 service_name = os.getenv("AGENT_SERVICE_NAME", "unknown-agent") agent_version = os.getenv("AGENT_VERSION", "dev") environment = os.getenv("ENVIRONMENT", "prod") # 构建Resource，包含Agent特有属性 resource = Resource.create({ "service.name": service_name, "service.version": agent_version, "deployment.environment": environment, "agent.runtime": "langchain", # 或llama-index等 "llm.provider": os.getenv("LLM_PROVIDER", "openai"), }) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor( OTLPSpanExporter( endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://otel-collector:4318/v1/traces"), timeout=10, ) ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 在main.py顶部调用 setup_telemetry()

验证方式：启动应用后，curl Collector的健康检查：

curl http://localhost:4318/v1/status # 应返回{"status":"available"}，且查看Collector日志，确认收到span含service.version字段

注意事项：OTEL_EXPORTER_OTLP_ENDPOINT必须指向你的OTel Collector，不能直连Jaeger。Collector负责批处理、采样、协议转换，直连会导致Agent应用线程阻塞。

3.2.2 RAG检索埋点：捕获向量DB调用全貌

在LangChain的Retriever中注入埋点：

# agent_runtime/retriever.py from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import DocumentCompressorPipeline from opentelemetry import trace class TracedQdrantRetriever(QdrantRetriever): def _get_relevant_documents(self, query: str, *, run_manager=None): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("qdrant_retrieve") as span: # 注入黄金字段 span.set_attribute("step_type", "retrieval") span.set_attribute("retrieval_query", query[:200]) span.set_attribute("qdrant_collection", self.collection_name) span.set_attribute("search_type", "similarity") span.set_attribute("k", self.k) try: # 调用父类方法 docs = super()._get_relevant_documents(query, run_manager=run_manager) span.set_attribute("retrieved_count", len(docs)) span.set_attribute("retrieved_docs_ids", [doc.metadata.get("id", "unknown") for doc in docs[:3]]) span.set_status(trace.StatusCode.OK) return docs except Exception as e: span.set_status(trace.StatusCode.ERROR, str(e)) span.record_exception(e) raise # 使用时替换原Retriever retriever = TracedQdrantRetriever( vectorstore=qdrant_store, search_type="similarity", k=5, )

验证方式：发起一次检索请求，然后在Jaeger UI搜索qdrant_retrieveSpan，检查：

• Tags中是否有retrieval_query、retrieved_count
• Logs中是否有exception事件（如有错误）
• Duration是否合理（我们期望<300ms）

实操心得：retrieved_docs_ids只记录前3个ID，避免Span过大。Jaeger对单Span大小有限制（默认1MB），记录全部ID可能触发截断，丢失关键信息。

3.2.3 LLM调用埋点：区分模型调用与流式响应

对LLM调用的埋点，必须区分同步和流式两种模式：

# agent_runtime/llm_wrapper.py from langchain_openai import ChatOpenAI from opentelemetry import trace class TracedChatOpenAI(ChatOpenAI): def invoke(self, input, config=None, **kwargs): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("llm_invoke") as span: span.set_attribute("step_type", "llm_call") span.set_attribute("llm_model", self.model_name) span.set_attribute("llm_temperature", self.temperature) span.set_attribute("input_tokens", self.get_num_tokens(input)) try: response = super().invoke(input, config, **kwargs) span.set_attribute("output_tokens", self.get_num_tokens(response.content)) span.set_attribute("total_tokens", self.get_num_tokens(input) + self.get_num_tokens(response.content)) span.set_status(trace.StatusCode.OK) return response except Exception as e: span.set_status(trace.StatusCode.ERROR, str(e)) span.record_exception(e) raise def stream(self, input, config=None, **kwargs): # 流式调用需特殊处理：Span在第一次yield时结束 tracer = trace.get_tracer(__name__) span = tracer.start_span("llm_stream") span.set_attribute("step_type", "llm_call") span.set_attribute("llm_model", self.model_name) span.set_attribute("llm_streaming", True) span.set_attribute("input_tokens", self.get_num_tokens(input)) try: first_chunk = True for chunk in super().stream(input, config, **kwargs): if first_chunk: # 第一块数据到达，标记LLM开始响应 span.set_attribute("first_token_latency_ms", int((time.time() - start_time) * 1000)) first_chunk = False yield chunk span.set_status(trace.StatusCode.OK) except Exception as e: span.set_status(trace.StatusCode.ERROR, str(e)) span.record_exception(e) raise finally: span.end() # 必须手动end

验证方式：调用stream方法，观察Jaeger中llm_streamSpan的first_token_latency_ms是否被正确记录。这是评估LLM响应速度的关键指标。

注意事项：流式Span必须手动span.end()，否则会内存泄漏。LangChain的stream是生成器，异常可能发生在任意yield点，因此finally块必不可少。

3.3 AgentOps流程验证（共6项）

3.3.1 Prompt版本回归测试：自动化脚本实现

测试脚本test_prompt_regression.py：

#!/usr/bin/env python3 import json import subprocess import sys from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity import numpy as np # 加载语义相似度模型 model = SentenceTransformer('all-MiniLM-L6-v2') def load_test_cases(file_path: str) -> list: with open(file_path) as f: return json.load(f) # [{"query": "...", "expected_response": "..."}, ...] def run_agent_with_prompt(prompt_file: str, query: str) -> str: # 调用本地Agent API，传入指定prompt版本 result = subprocess.run([ 'curl', '-s', '-X', 'POST', 'http://localhost:8000/v1/agent/chat', '-H', 'Content-Type: application/json', '-d', json.dumps({"input_text": query, "prompt_version": prompt_file}) ], capture_output=True, text=True) if result.returncode != 0: return "ERROR: " + result.stderr try: return json.loads(result.stdout).get("response", "") except: return "ERROR: invalid JSON" def calculate_similarity(text1: str, text2: str) -> float: embeddings = model.encode([text1[:512], text2[:512]]) # 截断防OOM return float(cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]) def main(): if len(sys.argv) != 3: print("Usage: python test_prompt_regression.py <base_prompt> <new_prompt>") sys.exit(1) base_prompt, new_prompt = sys.argv[1], sys.argv[2] test_cases = load_test_cases("test_data/regression_test.json") total_cases = len(test_cases) passed_cases = 0 similarities = [] for i, case in enumerate(test_cases): print(f"Testing case {i+1}/{total_cases}: '{case['query'][:50]}...'") base_resp = run_agent_with_prompt(base_prompt, case['query']) new_resp = run_agent_with_prompt(new_prompt, case['query']) # 关键case必须100%准确（如退款、投诉） if case.get("is_critical", False): if base_resp.strip() == case["expected_response"].strip(): # 基线正确，新版本也必须相同 if new_resp.strip() == case["expected_response"].strip(): passed_cases += 1 else: print(f" CRITICAL FAIL: base='{base_resp[:30]}...', new='{new_resp[:30]}...'") else: # 基线已错，跳过比较（说明基线需先修复） print(f" SKIP: base already wrong") continue else: # 非关键case，允许语义相似 sim = calculate_similarity(base_resp, new_resp) similarities.append(sim) if sim >= 0.95: passed_cases += 1 else: print(f" SIMILARITY LOW: {sim:.3f} (need >=0.95)") # 输出报告 avg_sim = np.mean(similarities) if similarities else 0 print(f"\n=== REGRESSION TEST REPORT ===") print(f"Total cases: {total_cases}") print(f"Passed: {passed_cases}/{total_cases}") print(f"Average similarity: {avg_sim:.3f}") print(f"Critical pass rate: {passed_cases/total_cases:.1%}") if passed_cases / total_cases < 0.95 or avg_sim < 0.95: print("❌ Regression test FAILED - PR cannot be merged") sys.exit(1) else: print("✅ All checks passed") if __name__ == "__main__": main()

验证方式：在CI中执行：

python test_prompt_regression.py prompts/base.j2 prompts/new.j2

实操心得：is_critical字段必须由产品经理在测试集中标记，不能由工程师随意添加。我们曾因漏标“我要投诉”为critical，导致新Prompt把投诉引导到满意度调查，上线后用户投诉量激增。

3.3.2 知识库变更审计：Git Hook强制声明

在知识库Git仓库根目录创建.githooks/pre-commit：

#!/bin/bash # 检查是否修改了PDF/MD文件 CHANGED_DOCS=$(git status --porcelain | grep -E '\.(pdf|md)$' | wc -l) if [ "$CHANGED_DOCS" -gt 0 ]; then echo "⚠️ Knowledge base files changed. Please fill CHANGELOG.md:" echo " - List affected user question types" echo " - Paste screenshots of before/after behavior" echo " - Get approval from domain expert and AI engineer" # 检查CHANGELOG.md是否存在且非空 if [ ! -f "CHANGELOG.md" ] || [ ! -s "CHANGELOG.md" ]; then echo "❌ ERROR: CHANGELOG.md is missing or empty" exit 1 fi # 检查CHANGELOG是否包含必要关键词 if ! grep -q "affected user question types" "CHANGELOG.md"; then echo "❌ ERROR: CHANGELOG.md missing 'affected user question types' section" exit 1 fi fi

验证方式：尝试提交PDF修改：

git add new_policy.pdf git commit -m "update policy" # 应被hook拦截，提示填写CHANGELOG.md

注意事项：Hook只检查提交内容，不替代人工审批。它只是确保“声明”动作发生，审批仍需走Jira或飞书流程。

3.3.3 灰度发布控制器：基于Redis的动态流量分配

控制器服务gray_controller.py：

# gray_controller.py import redis import hashlib import time class GrayController: def __init__(self, redis_url: str = "redis://localhost:6379/0"): self.redis = redis.from_url(redis_url) def get_traffic_ratio(self, service_name: str) -> float: # 从Redis读取当前灰度比例（0.0 ~ 1.0） ratio_str = self.redis.get(f"gray:ratio:{service_name}") return float(ratio_str) if ratio_str else 0.0 def is_in_gray(self, user_id: str, service_name: str) -> bool: # 用user_id哈希决定是否灰度，确保同一用户始终路由一致 hash_val = int(hashlib.md5(user_id.encode()).hexdigest()[:8], 16) max_val = 2**32 traffic_ratio = self.get_traffic_ratio(service_name) return (hash_val % max_val) < (max_val * traffic_ratio) def report_metrics(self, service_name: str, metrics: dict): # 上报指标到Redis，供自动扩缩容用 key = f"gray:metrics:{service_name}:{int(time.time())}" self.redis.hmset(key, metrics) self.redis.expire(key, 3600) # 1小时 controller = GrayController() # 在Agent入口处调用 def handle_request(user_id: str, query: str): if controller.is_in_gray(user_id, "customer-service-agent"): # 走新版本Agent return new_agent.invoke(query) else: # 走旧版本Agent return old_agent.invoke(query)

验证方式：设置灰度比为0.5，用不同user_id测试：

>>> from gray_controller import controller >>> controller.redis.set("gray:ratio:customer-service-agent", "0.5") >>> controller.is_in_gray("user_001", "customer-service-agent") True >>> controller.is_in_gray("user_002", "customer-service-agent") False >>> controller.is_in_gray("user_001", "customer-service-agent") # 同一用户始终True True

实操心得：哈希算法必须用MD5（或SHA256），不能用Python内置hash()，因为后者在不同进程间不一致，导致同一用户在不同实例上路由不同。

4. 常见问题与排查技巧实录：那些没写在文档里的坑

4.1 门禁相关问题

4.1.