聊到AI智能体(AI Agent),很多人第一反应是“能聊天、能写代码”,但在真实项目里,一个Agent能不能真正落地,看的不是它的“大脑”有多强,而是它的“手脚”能伸多远。今年我在一个内部项目中折腾了一套叫“Agent-Reach”的触达框架,核心就干一件事:让智能体从“被动等指令”变成“主动触达业务系统、外部服务和多端用户”。这篇文章就把这套方案的设计思路、核心模块和实操踩坑完整拆出来,给想自己搭一套类似架构的朋友参考。
1. Agent-Reach 整体设计与思路拆解
1.1 为什么需要“触达”这一层
先讲个背景。我之前做过一个客服工单助手,最初架构很简单:用户提问,大模型生成回答,返回给页面。上了线才发现问题很多——比如用户问“我的订单什么时候发货”,模型回答得再完美,订单数据在ERP里,模型根本查不到;再比如用户要求“帮我改一下收货地址”,模型回复“已为您提交申请”,实际上什么都没发生,因为没有任何一个接口被调用。
这个问题的本质,是Agent只做了“语义理解”和“内容生成”,没有做“动作执行”和“系统连接”。Agent-Reach这个名字,拆开看就是两层含义:一是Reach the Systems(触达系统),让Agent能够调用内部API、数据库、第三方服务;二是Reach the Users(触达用户),让Agent能把结果通过合适的渠道推送出去——网页、即时通讯、短信、甚至语音都算。
我给这套框架定的核心目标很朴素:让Agent像一个真正的“员工”,而不是一个“只会说话的顾问”。员工的特点是能够收发信息、操作系统、跟进任务、反馈结果,Agent-Reach就是把这些“工作能力”标准化、组件化,让上层业务可以快速组合。
1.2 三层架构:决策层、触达层、执行层
整个Agent-Reach框架我分成了三层感知模型,每一层只关注一件事:
- 决策层(Orchestrator):负责理解用户意图、拆解任务、规划步骤。这个层次就是大模型的核心推理能力,它决定“要做什么”。
- 触达层(Reach Modules):这是Agent-Reach的核心,负责“怎么够到目标”——管理工具清单、路由请求、封装协议、处理鉴权、重试补偿。它相当于Agent的手和脚。
- 执行层(Executors):真正干活的地方。调用具体API、查数据库、改状态、发消息。执行层可以是一个普通的REST接口,也可以是消息队列的消费者。
为什么要把触达层单独拿出来而不是让大模型直接调API?因为大模型直接调API有三个坑:第一,模型输出的工具调用参数经常不稳定,缺字段、类型错、格式乱,需要在触达层做一次“参数清洗”;第二,多个工具之间的依赖关系、顺序控制需要统一管理,不能让模型自己去胡拼乱凑;第三,鉴权、限流、日志这些横切逻辑,放在触达层统一处理,模型层可以保持干净。
1.3 设计时的一个关键取舍:工具越多越好吗
做Agent-Reach初期最容易犯的冲动,是把所有能接的系统全接进来。我一开始也是这么干的,一口气接入了十几个工具,结果发现模型经常选错工具:用户明明想问发票问题,它却去调了订单接口。后来我总结了一个经验,工具清单必须“按需暴露”——不是所有工具都向模型开放,而是根据当前对话上下文动态决定开放哪几个子集。
这个思路类似给员工分派工作时,不会把公司所有权限都发放给一个人,而是按岗位和任务范围授权。Agent-Reach中有一个“技能路由器”(Skill Router),它会先根据意图分类的结果,把大模型能看到的函数列表从几十个缩小到几个候选,再让模型做最终选择。这一步实测下来,工具选择准确率从70%左右提升到了92%以上。
2. 核心模块解析与关键技术点
2.1 技能注册中心:把接口变成模型看得懂的东西
触达层的第一步,是让大模型知道“你有什么工具、每个工具怎么用”。这一步的标准做法是写函数描述(Function Calling Schema),把每个接口的输入参数、返回结构、使用场景都用模型容易理解的语言写清楚。但实操中光有Schema还不够,模型经常会在参数边界上犯错。
Agent-Reach里我给每个Skill设计了三个层次的描述:
- 浅层描述:一句话说清这个技能是干什么的(给模型快速判断用)。
- 参数Schema:完整列出每个字段的类型、必填与否、取值枚举(给模型生成参数用)。
- 约束规则:包括参数间的关联条件、前置依赖状态、权限要求(给触达层的协议校验器用)。
举个例子,一个“查订单”的技能,浅层描述是“根据订单号或手机号查询订单基础信息和物流状态”,参数Schema里规定order_no和phone二选一或者组合,约束规则里写明“订单号必须为12位数字,手机号必须是已通过实名验证的号码”。这些约束在模型调用时不一定每次都遵守,所以触达层必须再做一次硬校验,不达标就返回结构化错误提示让模型自行修正。
2.2 上下文状态管理:一次触达不是“一次性”的
Agent-Reach运行了一段时间后我发现一个容易忽略的问题:工具调用的结果不是孤立的。用户可能先查了订单,然后问“这个订单能改地址吗”,或者先取消了一个订单,接着问“退款什么时候到”。如果触达层每次都是“无状态调用”,那么多轮对话之间就会脱节,模型不知道上一步执行了什么、结果是什么。
为此我引入了“会话状态对象”(Session State Object),每个用户的会话都维护一个JSON结构,里面记录:
- 用户当前正在操作的业务对象ID(比如订单号、工单号)
- 最近一次工具调用的返回摘要
- 当前流程走到了哪一步(比如改地址流程中,是等用户确认还是等填写新地址)
- 规则引擎需要标记的状态位(比如“已校验身份”“已确认取消”)
这个状态对象在触达层内部维护,每次模型发起工具调用前,触达层会自动往系统提示词里注入相关上下文摘要。这样模型就能“记住”自己上一轮做了什么,避免重复执行或逻辑冲突。实测中这个设计把多轮任务完成率提高了近三成,效果非常明显。
2.3 异步执行与回调机制:不能什么任务都当场返回
并不是所有工具调用都能在几秒内返回结果。有些任务天然是异步的,比如生成一份分析报告、批量发送邮件、等待外部系统审批。如果Agent一味等同步结果,用户的等待体验会很糟糕,而且大模型的接口调用也容易超时。
Agent-Reach内部专门做了“同步/异步双通道”的拆分逻辑:快操作(查询、简单配置)走同步通道,直接返回结果;慢操作(提交审批、批量任务、报表生成)走异步通道,触达层先接受任务、立即返回“任务已受理并附上任务ID”,后台通过消息队列执行,执行完毕后把结果推进消息中心,由Agent收到提示后主动向用户汇报。
这条设计让触达层具备了一个非常关键的能力:从“拉模式”变为“推拉结合”。系统不再只是等着用户提问,Agent也能在任务完成时主动推送结果。这个能力在自动化的场景中很重要——比如定时巡检、订阅提醒这类功能,本质上就是触达层的异步调度器在发挥作用。
3. 实操落地:一个最小可用系统的完整实现
3.1 环境选择与依赖安装
我这里用Python来演示,因为生态成熟、日常写Agent原型最快。主要依赖是fastapi(提供轻量API服务)、openai或anthropic的SDK(用于接入大模型)、redis(做会话状态缓存),以及celery或dramatiq(做异步任务队列,选一个即可,我这里选了dramatiq,因为它轻量、出错日志能看得很直观)。
# 创建虚拟环境并安装基础依赖 python -m venv agentreach_env source agentreach_env/bin/activate pip install fastapi uvicorn redis dramatiq openai pydantic我建议在实际项目中至少配上Redis,因为会话状态对象如果只放在进程内存里,一重启就全部丢失,线上会遇到莫名的多轮对话断片。另外如果用Docker部署,把Redis和Worker拆成独立服务,扩容也方便。
3.2 技能注册与基础数据模型
先把Skill的数据模型定义好。这个模型是整个触达层的中枢结构,后面所有的校验、路由、鉴权逻辑都围绕它展开。
# skills/models.py from pydantic import BaseModel, Field from typing import Optional, Dict, List, Any class SkillParameter(BaseModel): name: str type: str # string / integer / number / boolean / object / array required: bool = False description: str = "" enum: Optional[List[str]] = None # 允许的取值 constraints: List[str] = [] # 约束规则,例如 "order_no must be 12 digits" class SkillDefinition(BaseModel): name: str description: str category: str # 技能分类,用于路由缩小候选集 parameters: List[SkillParameter] handler: str # 对应执行层哪个函数 sync: bool = True # 是否同步返回 auth_required: bool = False timeout: int = 5 # 同步请求超时秒数每个技能注册时就是实例化一个SkillDefinition,放到技能注册表中。注册表内部用一个字典维护,Key是技能名,Value是技能定义。同时为了给大模型生成函数调用用,还需要写一个方法把这些定义转成API调用时的tools数组格式。
3.3 路由、校验、鉴权三件套的实现
触达层的核心逻辑在“请求入口”这个函数里完成。它接收模型输出的工具调用请求,依次做:技能路由、参数清洗、约束校验、鉴权检查,然后分发给执行层。
# reach/router.py from skills.registry import skill_registry from skills.validators import validate_params from security.authorizer import check_scope async def dispatch_tool_call(tool_name: str, raw_params: dict, session_state: dict) -> dict: # 1. 路由:确认技能是否注册 skill = skill_registry.get(skill_name=tool_name) if skill is None: return {"status": "error", "error_type": "UNKNOWN_SKILL", "message": f"{tool_name} not found"} # 2. 参数清洗与类型转换 cleaned, issues = validate_params(skill, raw_params) if issues: # 返回结构化错误,模型可以据此修正 return {"status": "error", "error_type": "INVALID_PARAMS", "issues": issues} # 3. 鉴权:检查会话是否有权调用该技能 has_scope, err = check_scope(session_state, skill.auth_required) if not has_scope: return {"status": "error", "error_type": "FORBIDDEN", "message": err} # 4. 同步/异步分流 if skill.sync: # 同步调用执行器 result = await call_executor(skill.handler, cleaned) return {"status": "success", "result": result} else: # 异步提交任务,立即返回受理号 task_id = await submit_async_task(skill.handler, cleaned, session_state) return {"status": "accepted", "task_id": task_id}这里重点说下参数清洗。一般的Function Calling返回参数经常有些小毛病,例如把整数写成字符串、漏掉必填字段、枚举值不匹配。清洗阶段会尝试自动转换类型,比如把数字字符串转成float、int,如果实在无法转换就生成一条明确的issues记录回传给模型。这样模型就能看到“哪个字段格式不对”然后自行纠正,而不是拿到一个冷冰冰的外部错误码。
3.4 会话状态对象的生命周期管理
会话状态对象在redis里以用户会话ID作为Key缓存,我封装了一个简单的管理器:
# state/store.py import json, aioredis redis_client = aioredis.from_url("redis://localhost:6379/2") PREFIX = "agentreach:session:" async def get_session(session_id: str) -> dict: raw = await redis_client.get(PREFIX + session_id) if raw is None: return {"flow": None, "objects": {}, "flags": {}} return json.loads(raw) async def save_session(session_id: str, state: dict) -> None: await redis_client.set(PREFIX + session_id, json.dumps(state), ex=24 * 3600)这个对象的设计遵循“最小必要”原则:只存流程所需的业务对象ID和关键标记,不存完整的对话历史。对话历史交给大模型侧自己管理,状态对象存的是“业务状态”,二者职责分开。
举个例子,改地址的流程:
# 第一次调用:发起改地址申请 state["flow"] = "CHANGE_ADDRESS" state["objects"]["order_no"] = "202511280001" # 第二次调用:获取新地址,校验后提交 state["flow"] = "CHANGE_ADDRESS_CONFIRM" state["flags"]["addr_confirmed"] = True每次模型发起调用前,触达层会把state里当前流程相关的摘要渲染成一段文字注入到对话上下文中。模型看到的是“该用户当前处于改地址流程,订单号是202511280001”,自然不会再跑偏去查其他订单。
3.5 异步任务与主动推送的落地
异步任务我用dramatiq来实现,每个耗时操作就是一个任务函数。任务完成之后需要主动通知用户,即“触达用户”的能力。这里我封装了一个统一的发送渠道抽象:
# reach/notifier.py async def notify_user(user_id: str, event_type: str, data: dict): # 根据用户偏好渠道分发:站内信 / 企业微信 / 短信等 routes = await get_user_channels(user_id) for channel in routes: if channel == "webhook": await post_webhook(user_id, event_type, data) elif channel == "sms": await send_sms(user_id, build_sms_content(event_type, data))异步流程里最关键的一点,是做“任务状态回执”。比如一个批量导出任务,执行到一半可能失败,Agent应该知道这个状态,并在下次用户询问时说“上次的导出任务失败了,原因是文件服务返回超时,我已经重新提交了一次”。要达成这个效果,任务执行器结束时要向会话状态对象里写入一条“事实记录”,而不是单纯把错误日志打在终端里。
4. 常见问题与排查技巧实录
4.1 工具调用参数反复出错:需要加“参数修复回传”机制
如果你在测试Agent时发现大模型生成的工具参数经常不对——比如日期格式不统一、ID少传一位、布尔值写成了字符串——不要急着换模型。我的做法是,在触达层做参数清洗和校验,把错误结构化之后回传给模型,让它根据错误信息修正后重新发起调用。设计两层重试机制:全自动重试两次;两次都失败后,停止调用,把问题及时反馈给用户并给出人工入口。这比让模型死循环重试靠谱得多。
注意:不要在一个函数上设置超过3次自动重试。模型重试次数越多,Token消耗越大,用户等待越久,成功率的边际提升却很小。第二次重试失败后大概率是业务规则冲突或数据问题,该转人工就要果断转。
4.2 技能暴露太多导致选择混乱:用意图分类做预路由
前面提到技能注册表不能一股脑全开放。实操中有个更细的技巧:在路由之前加一层轻量级的意图分类器,输入是用户最新的一句话和已进入的流程状态,输出是候选技能的集合。这个分类器可以用一个小模型,也可以用大模型,但用大模型时会让整体链路变长、变贵。我个人建议用一个规模较小的分类模型,先粗筛出三个候选技能,再让大模型做精排。这样准确率和响应速度都能兼顾。
4.3 异步任务状态不同步:让执行器写会话状态
最常见的问题是:Agent已经答复用户“正在处理,请稍后”,但后台任务跑完以后,用户再次询问,模型完全不知道任务已经完成,又重新提交了一次。原因就是任务执行器没有把结果写回会话状态。修复方案就是上面提到的,每个异步任务在结束时必须更新会话状态对象,把任务完成标记、结果摘要、错误信息都写到state里,Agent下次就能从state中读取事实,而不是凭想象回复。
4.4 高并发下Session丢失:一定要用Redis持久化
刚开始我把Session存在内存字典里,部署单机测试没问题,一放到多Worker的容器环境下就开始随机丢失会话状态,用户多问两轮就失忆。改成Redis后问题就消失了。还有个容易忽略的点:Redis里的session要设置过期时间,不然脏数据堆积会越占越多,一般按业务要求设为12到24小时即可。
4.5 全链路测试工具集
最后给一套我自用的测试清单。Agent触达层的测试不能只测接口,要按“对话场景”来测:
- 正常路径:意图清晰、参数完整、一次调用成功。
- 参数缺漏:用户不说话,只给了模糊指令,看模型是否懂得反问澄清。
- 多条件组合:两个条件冲突怎么办(地址已修改后又取消订单)。
- 超时回退:外部接口5秒不响应,Agent会怎么处理,是等待还是转异步还是降级回答。
- 鉴权边界:未登录用户调用需鉴权技能,是否正确拦截并引导登录。
这套清单跑下来,基本能把线上80%的翻车场景预先打掉。
5. 扩展场景与后续演进思路
Agent-Reach这套框架本身是业务无关的,我后来把它复用到好几个场景里:
- 智能客服工单助手:Agent能直接查订单、改地址、发起退款流程,全部通过触达层操作业务系统。
- 内容巡检机器人:定时浏览指定页面,发现异常时主动把截图和问题描述推到内部群聊渠道。
- 数据分析周报助手:接上BI系统查询接口,自动汇总上一周的数据指标,生成解读文本后推送给管理者。
- 流程审批助手:用户用自然语言发起请假、采购申请,Agent帮助填写表单并提交到审批流,审批后主动通知结果。
后续如果继续演进,我准备把“技能注册中心”改造得更像一套插件化系统,让业务团队自己注册接口,不用改代码就能增加新的技能。另一个方向是把质量评估指标也嵌入触达层——比如记录每次触达的失败原因、修复次数、平均延迟,这样能持续优化路由策略和函数描述。
我个人在落地这套框架的时候,最大的体会是:别急着追求“全自动”,先把“可控”做扎实。触达层本质上是一个交代给Agent的边界——它能做什么、不能做什么、做错了怎么补救,都在这一层定死。边界清晰了,Agent的“手和脚”才能既伸得远,又收得回。这是一条值得长期打磨的架构思路,希望对正好在做类似事情的朋友有帮助。