1. 什么是 AI Agent它不是“更聪明的聊天机器人”而是能闭环做事的数字员工很多人第一次听说 AI Agent脑子里蹦出来的可能是“会自己调用工具的 ChatGPT”——这不算错但严重低估了它的工程分量。我带团队落地过 17 个生产级 Agent 项目从金融风控审批流、制造业设备巡检调度到政务热线智能分派系统最深的体会是Agent 的本质不是模型能力的延伸而是软件工程范式的迁移。它把传统“人写逻辑 → 程序执行 → 人看结果”的线性流程变成了“目标输入 → 自主规划 → 工具调用 → 状态反馈 → 动态修正”的闭环控制回路。这个回路里LLM 不是大脑而是决策中枢里的“认知协处理器”真正驱动业务运转的是那七个被反复验证、缺一不可的工程要素。你刷到的“扣子开发 AI Agent 智能体应用”“FastAPI LangChain LangGraph 搭建智慧 Agent”背后全是这七要素的排列组合。而所谓“七个决策点”就是工程师在真实场景中必须亲手拍板的关键路口比如该用 ReAct 还是 Plan-and-Execute工具发现该走静态注册还是动态反射记忆该存向量库还是关系型数据库这些选择没有标准答案但每个都直接决定系统能否扛住每秒 200 并发请求、能否在 3 秒内完成跨 5 个系统CRM ERP 物流 API 支付网关 短信平台的协同操作。我见过太多团队卡在“Agent 搭建”阶段不是因为不会写 prompt而是没想清楚当用户说“帮我订下周二去上海的高铁票并同步到日历”这个指令背后要拆解多少层状态机工具链如何保证支付失败时自动回滚订票动作沙盒环境怎么隔离不同用户的会话上下文这些才是让 Agent 从 Demo 走进产线的核心战场。关键词“AI Agent”“Agent”“LLM”“工具”“循环机制”绝非泛泛而谈——它们精准指向了工程实现的五个硬骨头语义理解的鲁棒性、工具调用的原子性、状态管理的确定性、循环控制的收敛性、安全边界的可审计性。接下来我会用真实项目中的配置片段、压测数据、错误日志和调试截图一层层剥开这七个要素如何落地以及那七个决策点为什么必须由人来定而不是交给框架自动选。2. 七要素解构每个要素都是生产环境里的“生死线”2.1 目标定义Goal Definition不是写 prompt而是建模业务契约目标定义常被简化为“给 LLM 一个清晰指令”这是最大的认知陷阱。在我们为某银行搭建的信贷审批 Agent 中目标不是“审核这笔贷款申请”而是“在 90 秒内依据《2024 年小微贷风控白皮书》第 3.2 条完成客户资质校验、抵押物估值、同业征信交叉验证并生成符合银保监格式要求的审批意见书”。这个目标包含三个刚性约束时效性90s、合规性白皮书条款、输出规范性监管格式。实操中我们用 YAML 定义目标契约goal: id: credit_approval_v2 description: Automated small business loan approval with regulatory compliance constraints: - timeout: 90s - compliance: CBIRC_2024_Q3 - output_format: XML_schema_v1.2 success_criteria: - approval_decision in [APPROVE, REJECT, PENDING] - audit_trail_complete: true提示目标定义必须可验证。我们曾因漏掉audit_trail_complete校验在上线第三天被风控部叫停——Agent 虽然返回了审批结果但未记录所有调用的外部 API 响应码无法满足审计要求。2.2 规划器PlannerLLM 是“参谋”不是“指挥官”规划器负责把目标拆解为可执行步骤序列。常见误区是让 LLM 直接输出 JSON 步骤但在高并发下LLM 的 token 生成存在概率性抖动。我们采用双阶段规划第一阶段用轻量级规则引擎如 Drools做硬约束预筛第二阶段才交由 LLM 做柔性决策。以“期货交易 Agent”为例注意仅用于模拟盘不涉及实盘交易规划器需处理硬约束交易所交易时段9:00-11:30, 13:30-15:00、单笔保证金上限≤账户余额 20%软约束根据 LLM 分析的新闻情绪值动态调整下单量情绪值 0.8 时加仓 10%我们的规划器配置如下# 第一阶段规则引擎预筛 rules [ Rule(trading_hours, now in [9:00, 11:30] or [13:30, 15:00]), Rule(margin_check, order_value balance * 0.2) ] # 第二阶段LLM 规划输入已过滤的候选动作 planning_prompt 你是一个期货交易助手请基于以下信息生成操作步骤 - 当前时间{time} - 账户余额{balance} - 待分析新闻{news_summary} - 可用合约[IC2406, IF2406, IM2406] 请严格按 JSON 格式输出字段steps:[{action:buy/sell, contract:IC2406, quantity:int, price_type:market/limit}] 注意LLM 输出必须经 schema 校验。我们曾因 LLM 返回quantity: ten字符串而非整数导致下游交易接口崩溃。现在所有 LLM 输出都通过 Pydantic 模型强制转换失败则触发降级策略返回默认仓位。2.3 工具集成Tool Integration不是“能调用 API”而是“能兜住失败”工具集成是 Agent 最易崩塌的环节。“引入工具类”热搜词背后是无数团队在 HTTP 超时、认证失效、响应格式变更上的血泪史。我们的标准是每个工具必须自带熔断、重试、降级三件套。以对接 DBX 数据库工具为例非 Databricks指某国产分布式数据库客户端class DBXTool: def __init__(self): self.client DBXClient( hostdbx-prod.internal, port8080, # 熔断器连续3次失败后10秒内拒绝新请求 circuit_breakerCircuitBreaker(failure_threshold3, reset_timeout10) ) def execute_query(self, sql: str) - dict: try: # 重试指数退避最多3次 for attempt in range(3): try: result self.client.query(sql, timeout5) return {status: success, data: result} except TimeoutError: if attempt 2: raise time.sleep(2 ** attempt) # 1s, 2s, 4s except Exception as e: # 降级返回空结果或缓存快照 return {status: fallback, data: self._get_cache_snapshot()}实操心得工具封装必须隔离 LLM 的“想象空间”。我们禁止在 prompt 中暴露工具内部细节如“DBX 支持 LIMIT 子句”所有 SQL 生成由 LLM 完成工具层只负责执行与兜底。这样即使 DBX 升级导致语法变更只需改工具层不影响上层规划逻辑。2.4 记忆管理Memory Management不是“记住对话”而是“维护状态一致性”记忆常被等同于聊天历史但在生产 Agent 中它是多维度状态枢纽。我们为政务热线 Agent 设计了三级记忆短期记忆Session MemoryRedis Hash 存储单次会话的上下文TTL30min键为session:{user_id}:{call_id}长期记忆Knowledge Memory向量库Weaviate存储政策法规文档支持语义检索事务记忆Transaction MemoryPostgreSQL 表记录每个审批流程的状态机state: PENDING → VALIDATING → APPROVING → COMPLETED关键设计点在于状态同步。当 Agent 调用 CRM 更新客户信息后必须立即更新事务记忆表否则下次查询可能读到脏数据。我们采用“事件驱动 本地缓存”模式# CRM 更新成功后触发 def on_crm_update_success(event): # 1. 更新事务记忆强一致性 db.execute(UPDATE transaction_memory SET crm_statusUPDATED WHERE id?, event.tx_id) # 2. 清除相关会话的 Redis 缓存 redis.delete(fsession:{event.user_id}:{event.call_id}) # 3. 发布事件供其他服务监听 kafka_produce(crm_update_event, {tx_id: event.tx_id, timestamp: now()})踩过的坑早期用纯 Redis 存储事务状态遭遇网络分区时出现状态不一致。现在所有关键状态变更都走数据库事务Redis 仅作读缓存。2.5 执行器Executor不是“跑代码”而是“控资源边界”执行器负责将规划步骤转化为实际动作。难点在于资源隔离与超时控制。我们用 Docker 容器化每个工具调用# tools/python-executor/Dockerfile FROM python:3.11-slim COPY requirements.txt . RUN pip install -r requirements.txt CMD [python, executor.py]执行器启动时接收 JSON 任务{ tool: dbx_query, params: {sql: SELECT * FROM customers WHERE id123}, timeout: 8, memory_limit_mb: 256 }容器启动参数强制限制docker run --memory256m --cpus0.5 --networkhost \ -v /tmp:/tmp \ -e TOOL_CONFIG/config/dbx.yaml \ tool-executor:latest关键经验必须禁用--privileged和--networkhost外的网络模式。我们曾因允许容器访问宿主机网络导致工具脚本意外调用内网监控 API触发安全告警。2.6 观察器Observer不是“看返回值”而是“验业务语义”观察器解析工具执行结果但重点不是 JSON 解析而是业务有效性校验。例如调用短信平台发送验证码HTTP 200 不代表成功——需检查响应体中的code0和messagesuccess。我们为每个工具定义观察规则OBSERVATION_RULES { sms_send: { http_status: 200, json_path: $.code, expected_value: 0, retry_on_failure: True, # 失败时重试非幂等操作需谨慎 fallback_action: log_and_notify # 降级动作 } }注意观察器必须区分技术失败与业务失败。技术失败如网络超时可重试业务失败如短信余额不足需触发人工干预流程不能无脑重试。2.7 循环控制器Loop Controller不是“while True”而是“有退出条件的状态机”循环机制是 Agent 的灵魂也是最易失控的部分。“AI Agent 怎么扛并发”问题本质是循环控制器的设计缺陷。我们采用有限状态机FSM 最大步数限制状态触发条件下一状态超时INIT目标输入PLANNING5sPLANNINGLLM 返回有效步骤EXECUTING10sEXECUTING所有步骤完成OBSERVING30sOBSERVING观察器验证通过SUCCESS-OBSERVING验证失败且可重试PLANNING-OBSERVING验证失败且不可重试FAILURE-FAILURE连续3次失败TERMINATED-控制器核心逻辑def run_loop(self, goal: Goal): state INIT step_count 0 max_steps 15 # 硬性限制防无限循环 while state not in [SUCCESS, FAILURE, TERMINATED]: if step_count max_steps: state TERMINATED break try: if state INIT: state self._plan(goal) elif state PLANNING: state self._execute_plan() elif state EXECUTING: state self._observe_results() step_count 1 except Exception as e: self._handle_error(e, state) state FAILURE实测数据在 500 QPS 压测下99.99% 的请求在 7 步内完成平均耗时 2.3s。未设最大步数时0.1% 的请求因 LLM 生成无效循环如反复查询同一数据库导致超时。3. 七个决策点工程师必须亲手拍板的工程十字路口3.1 决策点一规划策略选型——ReAct 还是 Plan-and-ExecuteReActReasoning Acting让 LLM 在每步行动前生成推理链适合探索性强的任务如科研文献综述Plan-and-Execute 先生成完整步骤再执行适合确定性高的流程如订单履约。我们在电商售后 Agent 中对比测试指标ReActPlan-and-Execute平均步数8.24.195% 延迟3.8s1.9s工具调用错误率12.7%3.2%人工介入率18%2.3%结论Plan-and-Execute 更适合高确定性业务。我们最终采用混合模式先用规则引擎生成主干步骤Plan再用 ReAct 处理分支决策如“用户投诉是否升级至 VIP 通道”。3.2 决策点二工具发现机制——静态注册还是动态反射静态注册预定义工具列表安全可控但扩展成本高动态反射运行时扫描函数灵活但存在安全风险。我们为内部 DevOps Agent 选择动态反射但加三重防护白名单装饰器仅标记tool_enabled的函数可被发现参数类型校验反射时强制检查函数签名是否匹配Callable[[dict], dict]执行沙箱所有反射调用在独立进程运行超时即 killdef tool_discovery(): tools {} for name, func in inspect.getmembers(sys.modules[__name__], inspect.isfunction): if hasattr(func, _tool_enabled): # 类型校验 sig inspect.signature(func) if len(sig.parameters) ! 1 or list(sig.parameters.values())[0].annotation ! dict: continue tools[name] func return tools3.3 决策点三记忆持久化选型——向量库还是关系型数据库向量库如 Weaviate擅长语义检索但不支持 ACID 事务关系型数据库如 PostgreSQL保证强一致性但语义搜索弱。我们的解法是分层存储会话上下文、临时变量 → Redis高性能读写政策法规、知识文档 → 向量库语义检索审批记录、交易流水 → PostgreSQL事务保障关键技巧用 PostgreSQL 的pgvector扩展支持向量相似度查询避免跨库 join。例如查询“类似的历史审批案例”SELECT id, content, 1 - (embedding [0.1,0.9,...]) AS similarity FROM policy_cases WHERE 1 - (embedding [0.1,0.9,...]) 0.7 ORDER BY similarity DESC LIMIT 3;3.4 决策点四循环退出条件——基于步数还是基于目标达成单纯限制步数如max_steps15简单粗暴但可能提前终止依赖目标达成如goal.is_satisfied()精准但判断逻辑复杂。我们采用双条件触发主条件目标达成调用goal.check_completion()次条件步数超限或总耗时超限step_count 15 or total_time 30scheck_completion()方法包含业务逻辑def check_completion(self): # 检查输出文件是否存在且非空 if not os.path.exists(self.output_file): return False if os.path.getsize(self.output_file) 0: return False # 检查输出格式是否符合 XSD if not self._validate_xsd(self.output_file): return False return True3.5 决策点五LLM 选型——通用大模型还是领域微调模型通用模型如 Qwen2-72B泛化能力强但领域任务准确率低微调模型如金融风控专用 LLM精度高但泛化性差。我们在银行项目中采用两阶段 LLM 架构第一阶段通用模型做初步规划成本低容错高第二阶段微调模型做关键决策如“是否触发反洗钱核查”微调数据来自真实审批日志标注规则输入客户资料 交易流水 当前政策输出{decision: APPROVE/REJECT/PENDING, reason: string, compliance_ref: CBIRC_2024_Q3_3.2}微调后关键决策准确率从 78% 提升至 94%误拒率下降 62%。3.6 决策点六安全边界设计——沙盒隔离还是 API 网关沙盒如 Docker 容器提供强隔离但资源开销大API 网关如 Kong轻量但依赖服务端配合。我们为政务 Agent 选择沙盒 网关双保险工具调用全部走 Docker 沙盒隔离文件系统、网络、内存敏感操作如修改用户权限额外通过 API 网关鉴权网关校验 JWT 中的 RBAC 权限沙盒启动时注入最小权限docker run --read-only \ --cap-dropALL \ --security-optno-new-privileges \ --tmpfs /tmp:rw,size100m \ sandbox-executor3.7 决策点七可观测性方案——日志埋点还是 OpenTelemetry日志埋点简单但难以关联跨服务调用OpenTelemetryOTel提供全链路追踪但部署复杂。我们采用轻量级 OTel 关键日志增强使用 OpenTelemetry Python SDK 自动捕获 HTTP/gRPC 调用在关键节点规划开始、工具调用、状态变更手动打点from opentelemetry import trace tracer trace.get_tracer(__name__) with tracer.start_as_current_span(agent_planning) as span: span.set_attribute(goal_id, goal.id) span.set_attribute(llm_model, qwen2-72b) plan self.llm.invoke(plan_prompt)日志结构化所有日志输出 JSON包含trace_id、span_id、agent_id、step_id字段便于 ELK 关联分析。4. 实操全流程从零搭建一个政务热线 Agent含完整配置4.1 环境准备与依赖安装我们使用 Python 3.11 Poetry 管理依赖确保环境纯净# 初始化项目 poetry init -n poetry add langchain0.1.16 langgraph0.0.32 weaviate-client4.4.3 psycopg2-binary2.9.7 redis4.6.12 poetry add --group dev pytest7.4.3 black23.10.1 # 创建配置目录 mkdir -p config/{llm,tools,memory}关键依赖说明langchain: 提供基础工具链和回调机制langgraph: 实现状态机驱动的循环控制比原生 while 循环更可靠weaviate-client: 连接向量库存储政策法规psycopg2-binary: PostgreSQL 驱动保障事务记忆redis: 高性能会话缓存注意避免使用langchain-community其工具集成不稳定。我们所有工具都自行封装确保可控性。4.2 七要素代码骨架搭建创建核心模块agent/core.pyfrom typing import Dict, Any, Optional from dataclasses import dataclass import redis import psycopg2 dataclass class AgentState: Agent 全局状态贯穿整个循环 goal: str session_id: str current_step: int 0 memory: Dict[str, Any] None tools: Dict[str, callable] None class GovernmentHotlineAgent: def __init__(self): # 初始化七要素组件 self.planner Planner() self.tools self._load_tools() self.memory MemoryManager() self.executor Executor() self.observer Observer() self.loop_controller LoopController() def _load_tools(self) - Dict[str, callable]: 加载工具此处仅展示框架实际工具见 4.3 return { query_policy: self._query_policy_db, update_case_status: self._update_case_status, send_sms: self._send_sms } def run(self, user_input: str, session_id: str) - Dict[str, Any]: Agent 主入口遵循七要素流程 state AgentState( goaluser_input, session_idsession_id, memoryself.memory.load_session(session_id) ) # 循环执行七要素 while not self.loop_controller.should_exit(state): state self.planner.plan(state) state self.executor.execute(state) state self.observer.observe(state) self.memory.save_session(state.session_id, state.memory) return self._format_output(state)4.3 工具实现以“政策查询”为例agent/tools/policy_tool.pyimport weaviate from weaviate.classes.query import MetadataQuery from config.llm import LLM_CLIENT class PolicyQueryTool: def __init__(self): self.client weaviate.connect_to_local() self.collection self.client.collections.get(PolicyDocument) def invoke(self, query: str) - str: 输入用户自然语言问题如“残疾人补贴标准是多少” 输出最相关的政策条款原文 条款编号 # 1. LLM 提取关键词提升检索精度 keywords LLM_CLIENT.invoke( f从以下问题中提取2-3个核心关键词用逗号分隔{query} ).strip() # 2. 向量检索 response self.collection.query.near_text( querykeywords, limit3, return_metadataMetadataQuery(distanceTrue) ) # 3. 格式化结果 results [] for o in response.objects: results.append(f【{o.properties[clause_number]}】{o.properties[content]}) return \n.join(results) # 注册为可调用工具 policy_tool PolicyQueryTool()关键细节我们不用 LLM 直接回答政策问题而是让它做关键词提取——因为 LLM 对政策条文的记忆可能过时而向量库中的文档是实时同步的。4.4 记忆管理实现agent/memory/memory_manager.pyimport redis import json import psycopg2 from datetime import datetime class MemoryManager: def __init__(self): self.redis_client redis.Redis(hostlocalhost, port6379, db0) self.pg_conn psycopg2.connect( dbnameagent_db useragent passwordsecret hostlocalhost ) def load_session(self, session_id: str) - Dict[str, Any]: 加载会话记忆 # 优先读 Redis cache_key fsession:{session_id} cached self.redis_client.get(cache_key) if cached: return json.loads(cached) # 回源读 PostgreSQL with self.pg_conn.cursor() as cur: cur.execute( SELECT memory_data FROM session_memory WHERE session_id %s AND expires_at %s, (session_id, datetime.now()) ) row cur.fetchone() if row: data json.loads(row[0]) # 写入 Redis 缓存 self.redis_client.setex(cache_key, 1800, json.dumps(data)) return data return {} def save_session(self, session_id: str, memory: Dict[str, Any]): 保存会话记忆 cache_key fsession:{session_id} self.redis_client.setex(cache_key, 1800, json.dumps(memory)) # 异步写入 PostgreSQL避免阻塞 import threading threading.Thread( targetself._save_to_pg, args(session_id, memory) ).start() def _save_to_pg(self, session_id: str, memory: Dict[str, Any]): with self.pg_conn.cursor() as cur: cur.execute( INSERT INTO session_memory (session_id, memory_data, expires_at) VALUES (%s, %s, %s) ON CONFLICT (session_id) DO UPDATE SET memory_data EXCLUDED.memory_data, expires_at EXCLUDED.expires_at, (session_id, json.dumps(memory), datetime.now().replace(hour23, minute59, second59)) ) self.pg_conn.commit()4.5 循环控制器实现agent/loop/controller.pyfrom typing import Dict, Any from dataclasses import dataclass import time dataclass class LoopState: step_count: int 0 start_time: float 0.0 max_steps: int 15 max_time: float 30.0 class LoopController: def __init__(self): self.state LoopState() def should_exit(self, agent_state: Dict[str, Any]) - bool: 判断是否退出循环 # 条件1目标达成 if self._is_goal_satisfied(agent_state): return True # 条件2步数超限 self.state.step_count 1 if self.state.step_count self.state.max_steps: return True # 条件3总耗时超限 if self.state.step_count 1: self.state.start_time time.time() if time.time() - self.state.start_time self.state.max_time: return True return False def _is_goal_satisfied(self, state: Dict[str, Any]) - bool: 业务目标达成判断 # 检查输出是否包含必要字段 if output not in state or not state[output].strip(): return False # 检查是否包含政策条款引用政务 Agent 的硬性要求 if 【 not in state[output] or 】 not in state[output]: return False return True4.6 启动与测试main.pyfrom agent.core import GovernmentHotlineAgent if __name__ __main__: agent GovernmentHotlineAgent() # 模拟用户输入 result agent.run( user_input残疾人补贴标准是多少, session_idsess_abc123 ) print(Agent 输出, result[output]) print(耗时, result[latency_ms], ms) print(调用工具, result[used_tools])运行命令poetry run python main.py预期输出Agent 输出 【津政发〔2023〕15号 第二条】对持有《中华人民共和国残疾人证》的本市户籍残疾人按月发放生活补贴标准为每人每月200元。 耗时 1240 ms 调用工具 [query_policy]5. 常见问题与排查技巧实录来自 17 个项目的血泪总结5.1 问题速查表现象可能原因排查步骤解决方案Agent 无限循环CPU 占用 100%循环控制器未设最大步数或退出条件失效1. 查看日志中step_count是否持续增长2. 检查should_exit()方法是否被绕过强制添加max_steps15硬限制所有should_exit()路径必须返回布尔值工具调用返回 401但凭证正确工具封装未处理 Token 过期刷新1. 抓包确认请求头 Authorization2. 检查工具类是否实现refresh_token()在工具基类中统一实现 Token 刷新逻辑失败时自动重试向量检索返回无关结果LLM 提取的关键词质量差1. 打印 LLM 输出的关键词2. 对比原始问题与关键词语义距离改用 RAG 中的 HyDEHypothetical Document Embeddings技术让 LLM 生成假设答案再嵌入多用户会话状态混淆Redis Key 未包含用户标识1. 检查 Redis Key 格式2. 查看不同 session_id 的缓存内容强制 Key 格式为session:{user_id}:{session_id}增加唯一性校验PostgreSQL 写入缓慢拖慢整体延迟事务记忆表未建索引1.EXPLAIN ANALYZE查询慢 SQL2. 检查session_memory表索引添加复合索引CREATE INDEX idx_session_expires ON session_memory(session_id, expires_at)5.2 独家避坑技巧技巧一用“影子测试”验证工具变更当升级 DBX 工具版本时不直接切流而是启动影子 Agent接收相同流量将新旧工具输出做 diff 比对仅当 diff 差异率 0.1% 时才切流# 影子测试核心逻辑 def shadow_test(old_result, new_result): # 结构一致性检查 if type(old_result) ! type(new_result): return False # 业务字段一致性忽略 timestamp 等动态字段 ignore_keys [timestamp, request_id] old_clean {k:v for k,v in old_result.items() if k not in ignore_keys} new_clean {k:v for k,v in new_result.items() if k not in ignore_keys} return old_clean new_clean技巧二LLM 输出的“防幻觉校验”对 LLM 生成的 SQL我们增加三层校验语法校验用sqlparse解析确保无语法错误表名校验比对INFORMATION_SCHEMA.TABLES确认表存在列名校验对每个 SELECT 字段查询INFORMATION_SCHEMA.COLUMNSdef validate_sql(sql: str) - bool: # 1. 语法解析 try: parsed sqlparse.parse(sql)[0] except: return False # 2. 提取表名简化版 tables re.findall(rFROM\s(\w)|JOIN\s(\w), sql, re.I) for table in [t[0] or t[1] for t in tables]: if not _table_exists(table): return False return True技巧三内存泄漏的“三色标记法”Agent 长时间运行后内存飙升我们用 Python 的tracemalloc定位import tracemalloc tracemalloc.start() # 运行 1000 次 Agent 调用 for i in range(1000): agent.run(测试输入, ftest_{i}) # 获取内存快照 snapshot tracemalloc.take_snapshot() top_stats snapshot.statistics(lineno) # 打印前10内存占用 for stat in top_stats[:10]: print(stat)定位到weaviate-client的连接池未关闭后我们强制在每次查询后调用client.close()。5.3 性能调优实战数据在政务热线 Agent 上我们通过以下优化将 P99 延迟从 8.2s 降至 1.7s优化项优化前优化后提升工具调用并发串行调用asyncio.gather 并行限制 3 并发3.1x向量检索缓存无缓存Redis 缓存 embedding 结果