在业务系统里接入大模型能力时很多团队会遇到一个很现实的纠结直接引入一套完整的 Agent 框架学习成本高、依赖重、和现有业务代码耦合严重但如果不引入框架自己从零写一个又容易把逻辑写死后续扩展工具、切换模型都很痛苦。本文要讲的 embabel / embabel-agent就是围绕“把智能代理能力内嵌到业务系统”这一场景展开的一套轻量实现思路。先说清楚本文中的 embabel 并不是特指某个已经存在的开源项目而是用作“embeddable可嵌入”的变体代号我把这套可嵌入智能代理实现方案称为 embabel-agent。如果你在 GitHub 或其他平台看到同名项目请以对应项目的 README 和源码为准。本文的核心是给你一套可以跑起来、可以继续扩展的最小工程架构包含完整代码、接口联调示例、常见报错排查和工程化建议适合后端开发、架构师以及想要把大模型能力集成到内部系统里的同学参考。1. 背景与核心概念1.1 embabel / embabel-agent 到底是什么从名字看embabel 可以理解为 embeddable 的缩写感含义是“可以被嵌入的”。embabel-agent 对应的是一个轻量智能代理运行单元它以标准服务或组件的形式嵌在你的业务系统里对外提供“接收任务 - 理解意图 - 调用工具 - 返回结果”的能力。那它和普通的 API 封装有什么区别呢普通的 GPT API 封装通常只是把用户的输入透传给大模型再把模型的输出返回给调用方。它的问题在于你没办法让模型去查数据库、查订单状态、调用内部服务也没办法让模型记住多轮对话里的关键信息。而 embabel-agent 在模型之外多了一层“能力路由”模型不再直接回答问题而是根据任务内容决定“该调用哪个工具”或“该直接回答”。用一个简单的例子来理解用户说“帮我查一下订单 10086 的物流状态。”普通 API 封装会直接把这句话发给模型模型只能根据训练数据猜测或者告诉你“我无法查询实时物流”。加了 Agent 能力后系统会把这句话解析为一个 Query 动作调用订单系统的接口把返回结果组织成自然语言再回复给用户。embabel-agent 解决的就是这类“模型 工具 业务数据”的联动问题而且希望这种联动是以一种低耦合、可插拔、可观测的方式实现。1.2 它解决什么问题在实际项目里我们会遇到几类痛点第一模型不可信。直接裸调大模型输出格式不稳定可能答非所问也可能给你一段编造的 JSON。Agent 通过工具调用和结果校验把模型从“内容生产者”变成“流程调度者”降低幻觉带来的影响。第二业务接入成本高。很多 AI 框架为了让开发者更容易上手会提供一整套复杂抽象Chain、Memory、Callback、Loader、VectorStore 等等。对于只想要一个“能查库存、能算价格、能转人工”的轻量能力来说这些概念过重排障链路也很长。第三系统和 AI 逻辑混在一起。如果直接把大模型 SDK 散落在业务代码里后续切换厂商、升级 Prompt、增加工具都会很痛苦。embabel-agent 倾向于把 AI 调度逻辑收敛到一个独立服务或独立模块中业务系统只需要按约定格式发起请求。1.3 和常见 Agent 框架的区别对比 LangChain、AutoGPT 等方案embabel-agent 的定位更偏向“轻量嵌入式运行时”而不是一个完整的开发平台。对比维度LangChain 等重型框架embabel-agent 思路集成方式通常作为独立应用或复杂编排层存在可以作为一个子服务 / SDK 嵌入业务系统学习曲线概念多抽象层次多只保留 Agent 核心循环任务、工具、结果工具扩展需要通过框架规范注册用统一函数签名注册简单直接运行时依赖可能引入很多第三方库核心依赖少便于嵌入适合场景AI 产品、复杂工作流企业内部系统、业务系统能力增强当然这并不意味着轻量方案一定更好。如果你的目标是做一个复杂的多智能体系统或者需要大量文档检索、向量记忆、图形化编排主流框架会更合适。但如果只是想“让业务系统快速具备工具调用能力”从 embabel-agent 这种轻量思路开始往往见效更快。2. 环境准备与项目结构2.1 运行环境说明本文代码使用 Python 3.10 编写核心用 FastAPI 提供 HTTP 服务。大模型部分我会以 OpenAI SDK 为例但底层调用方式你可以替换成任意兼容接口的模型服务比如国内厂商的 OpenAI 兼容 API、本地部署的 vLLM 服务等。版本方面不建议完全照搬下面的数字请根据你的实际环境调整Python 3.10 fastapi 0.100 uvicorn 0.20 openai 1.0 pydantic 2.0如果网络环境有限制不需要依赖外部向量库也不需要安装 LangChain。这整套代码的依赖非常少。2.2 初始化项目先创建一个项目目录并初始化虚拟环境。以 Linux/macOS 为例mkdir embabel-agent-demo cd embabel-agent-demo python3 -m venv venv source venv/bin/activate pip install fastapi uvicorn openai pydanticWindows 下激活虚拟环境的命令是venv\Scripts\activate2.3 项目结构规划为了让代码保持清晰我按下面的结构组织项目embabel-agent-demo/ ├── agent_core.py # Agent 核心循环 ├── tools.py # 工具定义与注册 ├── api.py # FastAPI 接口 ├── config.py # 配置管理 └── requirements.txt # 依赖清单这是一个比较经典的“工具层 - Agent 层 - API 层”三层结构工具层负责定义能力Agent 层负责调度API 层负责对外暴露入口。后续增加新能力时多数只需要在 tools.py 里新增一个函数并完成注册。3. 核心原理拆解3.1 Agent 的最小运行循环embabel-agent 的核心是一个循环可以简化成四步接收用户任务把用户输入、历史会话、可用工具描述一起组装成模型的输入。模型决策模型判断是直接回答还是需要调用某个工具。如果是调用工具会输出一个结构化的调用请求包括工具名和参数。执行工具系统根据模型输出的工具名找到对应函数并执行拿到结果。生成回复把工具返回结果交给模型让模型组织成最终的自然语言回复。这个循环可以只执行一次也可以反复执行多次。比如第一个工具返回的是“订单 ID”第二个工具需要根据“订单 ID”查物流那么就需要让模型多做一轮决策。最小循环示意如下用户输入 - 组装 Prompt - 模型输出直接回答 or 调用工具 - 如果是调用工具则执行并回填结果 - 模型生成最终回答 - 返回给用户在本文的代码里我会用“单轮工具调用”为主后续如果你想支持多轮调用可以在run方法里加一个循环深度控制即可。3.2 工具注册机制工具注册是 Agent 系统里最基础、也最关键的机制。它的目标只有一个让模型知道“你能用什么工具”然后让系统知道“当模型选择某个工具时应该执行哪段代码”。我建议用统一函数签名来规范工具def tool_name(param1: str, param2: int 0) - str: 函数的 docstring 用来描述工具的用途和参数含义。 ...为什么强调统一签名因为模型输出的 JSON 参数是不可控的如果每个工具的参数风格完全不一样Agent 层就得写大量解析和容错代码。统一签名之后Agent 层可以用**kwargs的方式把参数直接传给工具函数再做异常处理。3.3 Prompt 设计Agent 系统的效果很大程度上取决于 Prompt 是否能清晰描述工具。模型需要知道当前有哪些工具可用。每个工具是做什么的。什么时候该用工具什么时候该直接回答。参数格式是什么。一套可复用的 Prompt 模板通常包含这几段System 角色设定可用工具列表名称、描述、参数 schema历史对话用户当前输入在下面的实战代码中我会把工具描述动态渲染到 Prompt 里这样新增工具时不需要改模型调用逻辑。3.4 上下文与记忆管理很多人一开始会忽略上下文长度问题。多轮对话中如果每次都把所有历史消息传给模型很快会超出上下文窗口限制。embabel-agent 初步可以只保留最近 N 轮对话核心原则是保留最近的用户意图和系统回答。截断过长的历史。对关键业务数据比如订单 ID、用户名可以考虑直接从历史中提取后放到“当前会话变量”里而不是完全依赖模型记忆。在工程上可以先从“保留最近 6 轮”开始后续再根据业务场景引入向量记忆或摘要记忆。4. 完整实战实现一个 embabel-agent 服务接下来我们正式写代码。我会按照“工具层 - Agent 层 - API 层”的顺序展开。4.1 实现工具层文件路径tools.py# tools.py from typing import Dict, Callable import datetime import json import random def get_current_time() - str: 返回当前系统时间格式为 YYYY-MM-DD HH:MM:SS return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_order_status(order_id: str) - str: 根据订单ID查询订单状态。 参数: order_id: 订单ID字符串类型例如 10086 返回: 订单状态描述例如 已发货、已签收、待支付 # 示例实现实际项目中这里通常要调用订单服务 status_map { 10086: 已发货正在运输途中, 10010: 已完成签收, 10000: 待支付, } status status_map.get(order_id, 未查询到该订单) return json.dumps({order_id: order_id, status: status}, ensure_asciiFalse) def calculate_express_fee(weight: float, city: str) - str: 根据包裹重量和目的城市计算运费。 参数: weight: 包裹重量单位千克浮点数 city: 目的城市名称字符串 返回: 运费结果单位元 base 8.0 if city in (北京, 上海, 广州, 深圳): base 10.0 total base max(0, weight - 1) * 2 return json.dumps({weight: weight, city: city, fee: round(total, 2)}, ensure_asciiFalse) # 工具注册表 TOOLS: Dict[str, Callable] { get_current_time: get_current_time, get_order_status: get_order_status, calculate_express_fee: calculate_express_fee, } def get_tool_descriptions() - str: 生成工具描述文本用于注入到 Prompt 中。 descriptions [] for name, func in TOOLS.items(): descriptions.append(f- {name}: {func.__doc__}) return \n.join(descriptions)这里我用get_tool_descriptions动态生成工具描述模型可以从描述里知道每个工具的作用。需要注意calculate_express_fee和get_order_status返回的是 JSON 字符串这比直接返回一个 Python 对象更稳定因为模型看到的是文本理解成本更低。4.2 实现 Agent 核心文件路径agent_core.py# agent_core.py import json from typing import List, Dict, Any from openai import OpenAI from tools import TOOLS, get_tool_descriptions SYSTEM_PROMPT_TEMPLATE 你是一个嵌入在业务系统中的智能助手你的代号是 embabel-agent。 你可以使用以下工具来完成用户任务 {tool_descriptions} 请根据用户问题判断是否需要调用工具 - 如果用户问题需要实时数据请返回一个 JSON格式如下 {{action: call_tool, tool_name: 工具名, arguments: {{参数名: 参数值}}}} - 如果用户问题可以直接回答请返回一个 JSON格式如下 {{action: reply, content: 你的回答}} 注意 1. 你只能返回 JSON不要输出其他解释性文字。 2. 如果无法确定参数值请使用合理默认值或向用户询问。 3. 如果工具返回结果请结合结果给用户一个友好的自然语言回复。 class EmbabelAgent: def __init__(self, api_key: str, base_url: str None, model: str gpt-4o-mini): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.history: List[Dict[str, str]] [] def _build_messages(self, user_input: str) - List[Dict[str, str]]: system_prompt SYSTEM_PROMPT_TEMPLATE.format( tool_descriptionsget_tool_descriptions() ) messages [{role: system, content: system_prompt}] # 只保留最近 6 轮历史 messages.extend(self.history[-6:]) messages.append({role: user, content: user_input}) return messages def _call_llm(self, messages: List[Dict[str, str]]) - str: response self.client.chat.completions.create( modelself.model, messagesmessages, temperature0.2, ) return response.choices[0].message.content def _parse_tool_call(self, model_output: str) - Dict[str, Any]: 解析模型输出期望拿到 JSON。 output model_output.strip() # 去掉可能的 json 包裹 if output.startswith(): lines output.split(\n) lines lines[1:-1] if lines[-1].strip() else lines[1:] output \n.join(lines) try: parsed json.loads(output) except json.JSONDecodeError: return {action: reply, content: 系统暂时无法理解你的请求请稍后再试。} return parsed def run(self, user_input: str) - str: messages self._build_messages(user_input) model_output self._call_llm(messages) parsed self._parse_tool_call(model_output) if parsed.get(action) reply: reply parsed.get(content, 好的已收到你的消息。) elif parsed.get(action) call_tool: tool_name parsed.get(tool_name) arguments parsed.get(arguments, {}) if tool_name not in TOOLS: reply 当前没有找到可用工具请换个问题。 else: try: tool_result TOOLS[tool_name](**arguments) reply f工具 {tool_name} 返回结果{tool_result} except TypeError as e: reply f工具参数错误{e} except Exception as e: reply f工具执行失败{e} else: reply 模型返回格式异常请重新提问。 # 记录历史 self.history.append({role: user, content: user_input}) self.history.append({role: assistant, content: reply}) return reply这段代码是整个方案的核心有几个地方需要重点解释_build_messages负责把系统提示词、历史记录和用户输入组装成模型需要的消息格式。_parse_tool_call处理了模型输出可能带 Markdown 代码块的情况。大模型在返回 JSON 时经常会把内容包在 json 里面所以这里做了一层兼容处理。run方法里做了参数异常处理工具参数不合法时不会让服务直接崩溃。历史记录这里保存的是“原始工具结果”并没有让模型把结果重写成自然语言。如果希望最终回复更自然可以在拿到工具结果后再让模型进行一次“总结”。这里先保留原始结果便于排查。4.3 创建 HTTP 接口文件路径api.py# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent_core import EmbabelAgent from config import OPENAI_API_KEY, OPENAI_BASE_URL, MODEL_NAME app FastAPI(titleembabel-agent API, version1.0.0) agent EmbabelAgent( api_keyOPENAI_API_KEY, base_urlOPENAI_BASE_URL, modelMODEL_NAME, ) class ChatRequest(BaseModel): message: str session_id: str default class ChatResponse(BaseModel): reply: str session_id: str app.post(/chat, response_modelChatResponse) def chat(request: ChatRequest): if not request.message.strip(): raise HTTPException(status_code400, detail消息不能为空) # 这里 session_id 先只做透传后续可基于 session 维护独立记忆 reply agent.run(request.message) return ChatResponse(replyreply, session_idrequest.session_id) app.get(/health) def health(): return {status: ok}这里agent是全局单例。实际生产环境中如果服务会同时处理大量用户你需要考虑每个会话独立记忆这里先用简单方式演示。4.4 配置文件文件路径config.py# config.py import os OPENAI_API_KEY os.getenv(OPENAI_API_KEY, sk-xxx) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, None) # 兼容 OpenAI 或国内兼容接口 MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini)将模型密钥放到环境变量中避免写死在代码里。如果你使用的是国内模型平台的 OpenAI 兼容接口只需要设置OPENAI_BASE_URL和MODEL_NAME。4.5 运行与验证先创建一个简单的启动脚本也可以直接用 uvicorn 命令export OPENAI_API_KEY你的key uvicorn api:app --host 0.0.0.0 --port 8000 --reload看到类似下面的输出说明服务已经启动INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Application startup complete.然后使用 curl 发起一个请求curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 现在几点了, session_id: test1}如果模型判断需要调用get_current_time工具返回结果类似{ reply: 工具 get_current_time 返回结果2025-01-15 14:30:22, session_id: test1 }再试一个订单查询curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 帮我查一下订单10086状态, session_id: test1}这个例子中模型会调用get_order_status工具并返回订单状态 JSON 字符串。4.6 让回复更自然上面代码返回的是“工具调用结果”原文从用户体感上看不够友好。要让 Agent 把工具结果转成自然语言可以在拿到tool_result后再用模型做一次总结。修改run方法中对应的分支if parsed.get(action) call_tool: tool_name parsed.get(tool_name) arguments parsed.get(arguments, {}) if tool_name not in TOOLS: reply 当前没有找到可用工具请换个问题。 else: try: tool_result TOOLS[tool_name](**arguments) # 调用模型做一次总结 summary_messages [ {role: system, content: 你是一个友好客服助手请把工具返回内容转成简洁自然的中文回复。}, {role: user, content: f工具执行结果{tool_result}}, ] summary_response self.client.chat.completions.create( modelself.model, messagessummary_messages, temperature0.3, ) reply summary_response.choices[0].message.content except TypeError as e: reply f工具参数错误{e} except Exception as e: reply f工具执行失败{e}这样用户看到的不再是 JSON而是类似“您的订单 10086 已发货正在运输途中”这样的自然语言。5. 嵌入业务系统的三种常见方式embabel-agent 不止能作为一个独立 HTTP 服务存在它还可以以其他方式嵌入不同形态的业务系统。5.1 HTTP 直连调用这是最简单的方式也是上面示例所演示的。业务系统将用户消息 POST 到/chat接口拿到返回结果。适合Web 客服系统移动端 App 的后端内部管理后台这种方式需要重点考虑请求超时。大模型响应通常需要 1 到 5 秒工具调用多个轮次时可能更长。建议业务侧设置 10 到 30 秒的超时时间并增加重试机制。5.2 消息队列异步接入如果业务系统是事件驱动架构可以引入消息队列如 RabbitMQ、Kafka、Redis Stream来解耦。流程是业务系统把任务消息发送到队列。embabel-agent 消费消息调用大模型和相关工具。将结果写入结果队列或回调接口。业务系统异步获取结果。这种模式适合耗时长、或者不希望阻塞主流程的任务比如批量内容总结、工单自动分类、报表生成等。使用异步方案时要注意任务状态管理和幂等性。5.3 SDK 内嵌如果业务系统本身就是 Python 服务并且不想额外引入网络请求可以直接把agent_core.py和tools.py作为一个 SDK 模块引入。from agent_core import EmbabelAgent agent EmbabelAgent(api_keyxxx, modelgpt-4o-mini) result agent.run(帮我查一下订单10086状态) print(result)这样做的好处是调用延迟最低坏处是会让业务系统直接依赖大模型网络调用一旦模型服务出问题业务主流程也会受影响。所以 SDK 内嵌通常建议配合降级开关使用模型不可用时直接返回预设兜底话术。6. 常见问题与排查思路6.1 模型调用超时现象接口长时间无响应最终报超时错误。常见原因模型服务本身响应慢。Prompt 过长尤其是塞入大量工具描述和历史记录。网络链路不稳定。处理思路缩短历史记录轮数。精简工具描述必要时只把当前场景相关的工具描述传给模型。设置合理的超时时间建议客户端超时设为 10 到 30 秒。6.2 模型输出 JSON 解析失败现象服务端日志提示 JSONDecodeError或者返回“模型返回格式异常”。常见原因大模型在 JSON 前后加了 Markdown 代码块。Prompt 没有足够强地约束输出格式。模型版本或参数导致输出不稳定。处理思路在解析前做字符串清理去掉多余空白和代码块。强化 System Prompt明确要求“只返回 JSON不要解释”。适当降低 temperature建议设置在 0 到 0.3 之间。如果还不行可以考虑使用函数调用功能让模型按 schema 输出结构化结果而不是纯文本 JSON。6.3 工具参数错误现象工具执行时报 TypeError 或参数缺失。常见原因模型生成的参数名与函数参数名不一致。模型把数字参数当成字符串传递。Prompt 里的工具描述不够明确。处理思路在工具注册时补齐参数类型说明。在 Agent 层捕获参数异常并返回“工具参数错误”。对关键参数做类型兼容处理例如float(weight)强制转换。6.4 并发请求导致全局变量污染现象不同用户之间的对话历史相互串线。常见原因agent是全局单例self.history是实例变量。多个用户并发访问同一个实例会导致记忆混乱。处理思路按session_id维护会话上下文而不是把所有历史放在同一个 Agent 实例里。可以使用一个全局字典key 为session_idvalue 为EmbabelAgent实例。生产环境中建议使用 Redis 保存会话状态保证多实例部署时状态一致。6.5 问题排查清单1. 模型有没有返回 - 没有返回检查 key、base_url、网络、模型名。 2. 模型返回了但解析失败 - 查看原始输出清理代码块加强 Prompt。 3. 解析成功但工具没执行 - 检查工具名是否在注册表里。 4. 工具执行报错 - 查看参数类型和函数签名是否匹配。 5. 结果返回了但用户不满意 - 调整总结模型 Prompt或者把工具结果做得更结构化。7. 最佳实践与工程建议7.1 工具层独立于模型层设计工具时不要让工具函数直接依赖 OpenAI SDK 或 Agent 类。工具函数应该是纯 Python 函数输入参数、输出字符串。这样做的最大好处是你可以单独写单元测试来验证工具逻辑也可以让其他模型框架复用这套工具。工具函数建议统一返回 JSON 字符串而不是裸文本。后续解析更稳定日志也更易读。7.2 Prompt 与模型参数工具描述用中文还是英文取决于模型能力。中文业务场景下工具描述用中文往往效果更好。每个工具的描述要写清楚“什么时候用”和“参数含义”只写函数名是不够的。temperature建议控制在 0.3 以内减少随机性。如果模型支持 function calling优先使用原生 function calling 能力比让模型裸输出 JSON 更稳定。7.3 安全边界Agent 能调用工具就相当于拿到了执行能力。需要特别注意以下几点工具白名单只注册业务真正需要的能力不要盲目暴露所有内部接口。参数校验工具函数内部要对入参做合法性校验避免模型传入非法值。权限隔离在 API 层引入认证和授权至少要做到 token 校验避免未授权调用。敏感信息不要让 Agent 把内部数据库连接串、密钥、Token 等敏感信息写入工具描述或 Prompt。我这里再强调一次涉及生产环境的任何工具接入都要遵循最小权限原则并经过测试环境充分验证。7.4 日志与可观测性Agent 系统排障比普通接口难因为中间多了一次“模型决策”过程。建议在关键节点打日志收到用户请求 模型原始输出 解析后的结构化指令 工具执行结果 最终返回内容日志里最好带上session_id和请求耗时。这样一旦出现问题可以快速定位是模型决策错、解析错、还是工具执行错。7.5 生产落地顺序如果要在真实业务中落地 embabel-agent我建议按以下顺序推进先定义 3 到 5 个核心工具完成最小闭环。在测试环境跑通端到端流程重点看模型的工具选择准确率。加认证、限流、日志和监控。小流量灰度观察线上请求的成功率和耗时。再逐步扩展工具数量和 Prompt 场景。不要一上来就把几十个工具全部注册进去工具太多会让模型选择准确率下降Prompt 也会越来越长最后性能劣化。8. 总结与下一步本文围绕 embabel / embabel-agent 这个主题讲解了可嵌入智能代理的核心概念并提供了一个可直接运行的 FastAPI 实现。我们完成了工具注册、Agent 调度、HTTP 接口、结果解析以及自然语言总结等关键环节同时也整理了超时、JSON 解析、参数错误、会话隔离等高频问题的排查思路。这套方案在落地时最有价值的地方往往不是 Agent 本身而是工具层的沉淀。每接入一个工具就等于让业务系统多了一种可以被模型调用的能力日积月累Agent 能解决的问题范围会越来越大。如果你正准备让业务系统接入大模型能力我建议先不要急着上重型框架用 embabel-agent 这种轻量方式跑通一个真实场景再根据问题逐步完善。下一步可以尝试的方向包括接入 function calling 原生能力、使用 Redis 管理多用户会话、把工具执行改成异步任务、引入简单的模型输出校验机制。希望这篇文章能帮你少踩一些坑也欢迎你在自己的项目里动手试一下这套完整代码。 轻量级可嵌入Agent：从零实现业务系统大模型工具调用
在业务系统里接入大模型能力时很多团队会遇到一个很现实的纠结直接引入一套完整的 Agent 框架学习成本高、依赖重、和现有业务代码耦合严重但如果不引入框架自己从零写一个又容易把逻辑写死后续扩展工具、切换模型都很痛苦。本文要讲的 embabel / embabel-agent就是围绕“把智能代理能力内嵌到业务系统”这一场景展开的一套轻量实现思路。先说清楚本文中的 embabel 并不是特指某个已经存在的开源项目而是用作“embeddable可嵌入”的变体代号我把这套可嵌入智能代理实现方案称为 embabel-agent。如果你在 GitHub 或其他平台看到同名项目请以对应项目的 README 和源码为准。本文的核心是给你一套可以跑起来、可以继续扩展的最小工程架构包含完整代码、接口联调示例、常见报错排查和工程化建议适合后端开发、架构师以及想要把大模型能力集成到内部系统里的同学参考。1. 背景与核心概念1.1 embabel / embabel-agent 到底是什么从名字看embabel 可以理解为 embeddable 的缩写感含义是“可以被嵌入的”。embabel-agent 对应的是一个轻量智能代理运行单元它以标准服务或组件的形式嵌在你的业务系统里对外提供“接收任务 - 理解意图 - 调用工具 - 返回结果”的能力。那它和普通的 API 封装有什么区别呢普通的 GPT API 封装通常只是把用户的输入透传给大模型再把模型的输出返回给调用方。它的问题在于你没办法让模型去查数据库、查订单状态、调用内部服务也没办法让模型记住多轮对话里的关键信息。而 embabel-agent 在模型之外多了一层“能力路由”模型不再直接回答问题而是根据任务内容决定“该调用哪个工具”或“该直接回答”。用一个简单的例子来理解用户说“帮我查一下订单 10086 的物流状态。”普通 API 封装会直接把这句话发给模型模型只能根据训练数据猜测或者告诉你“我无法查询实时物流”。加了 Agent 能力后系统会把这句话解析为一个 Query 动作调用订单系统的接口把返回结果组织成自然语言再回复给用户。embabel-agent 解决的就是这类“模型 工具 业务数据”的联动问题而且希望这种联动是以一种低耦合、可插拔、可观测的方式实现。1.2 它解决什么问题在实际项目里我们会遇到几类痛点第一模型不可信。直接裸调大模型输出格式不稳定可能答非所问也可能给你一段编造的 JSON。Agent 通过工具调用和结果校验把模型从“内容生产者”变成“流程调度者”降低幻觉带来的影响。第二业务接入成本高。很多 AI 框架为了让开发者更容易上手会提供一整套复杂抽象Chain、Memory、Callback、Loader、VectorStore 等等。对于只想要一个“能查库存、能算价格、能转人工”的轻量能力来说这些概念过重排障链路也很长。第三系统和 AI 逻辑混在一起。如果直接把大模型 SDK 散落在业务代码里后续切换厂商、升级 Prompt、增加工具都会很痛苦。embabel-agent 倾向于把 AI 调度逻辑收敛到一个独立服务或独立模块中业务系统只需要按约定格式发起请求。1.3 和常见 Agent 框架的区别对比 LangChain、AutoGPT 等方案embabel-agent 的定位更偏向“轻量嵌入式运行时”而不是一个完整的开发平台。对比维度LangChain 等重型框架embabel-agent 思路集成方式通常作为独立应用或复杂编排层存在可以作为一个子服务 / SDK 嵌入业务系统学习曲线概念多抽象层次多只保留 Agent 核心循环任务、工具、结果工具扩展需要通过框架规范注册用统一函数签名注册简单直接运行时依赖可能引入很多第三方库核心依赖少便于嵌入适合场景AI 产品、复杂工作流企业内部系统、业务系统能力增强当然这并不意味着轻量方案一定更好。如果你的目标是做一个复杂的多智能体系统或者需要大量文档检索、向量记忆、图形化编排主流框架会更合适。但如果只是想“让业务系统快速具备工具调用能力”从 embabel-agent 这种轻量思路开始往往见效更快。2. 环境准备与项目结构2.1 运行环境说明本文代码使用 Python 3.10 编写核心用 FastAPI 提供 HTTP 服务。大模型部分我会以 OpenAI SDK 为例但底层调用方式你可以替换成任意兼容接口的模型服务比如国内厂商的 OpenAI 兼容 API、本地部署的 vLLM 服务等。版本方面不建议完全照搬下面的数字请根据你的实际环境调整Python 3.10 fastapi 0.100 uvicorn 0.20 openai 1.0 pydantic 2.0如果网络环境有限制不需要依赖外部向量库也不需要安装 LangChain。这整套代码的依赖非常少。2.2 初始化项目先创建一个项目目录并初始化虚拟环境。以 Linux/macOS 为例mkdir embabel-agent-demo cd embabel-agent-demo python3 -m venv venv source venv/bin/activate pip install fastapi uvicorn openai pydanticWindows 下激活虚拟环境的命令是venv\Scripts\activate2.3 项目结构规划为了让代码保持清晰我按下面的结构组织项目embabel-agent-demo/ ├── agent_core.py # Agent 核心循环 ├── tools.py # 工具定义与注册 ├── api.py # FastAPI 接口 ├── config.py # 配置管理 └── requirements.txt # 依赖清单这是一个比较经典的“工具层 - Agent 层 - API 层”三层结构工具层负责定义能力Agent 层负责调度API 层负责对外暴露入口。后续增加新能力时多数只需要在 tools.py 里新增一个函数并完成注册。3. 核心原理拆解3.1 Agent 的最小运行循环embabel-agent 的核心是一个循环可以简化成四步接收用户任务把用户输入、历史会话、可用工具描述一起组装成模型的输入。模型决策模型判断是直接回答还是需要调用某个工具。如果是调用工具会输出一个结构化的调用请求包括工具名和参数。执行工具系统根据模型输出的工具名找到对应函数并执行拿到结果。生成回复把工具返回结果交给模型让模型组织成最终的自然语言回复。这个循环可以只执行一次也可以反复执行多次。比如第一个工具返回的是“订单 ID”第二个工具需要根据“订单 ID”查物流那么就需要让模型多做一轮决策。最小循环示意如下用户输入 - 组装 Prompt - 模型输出直接回答 or 调用工具 - 如果是调用工具则执行并回填结果 - 模型生成最终回答 - 返回给用户在本文的代码里我会用“单轮工具调用”为主后续如果你想支持多轮调用可以在run方法里加一个循环深度控制即可。3.2 工具注册机制工具注册是 Agent 系统里最基础、也最关键的机制。它的目标只有一个让模型知道“你能用什么工具”然后让系统知道“当模型选择某个工具时应该执行哪段代码”。我建议用统一函数签名来规范工具def tool_name(param1: str, param2: int 0) - str: 函数的 docstring 用来描述工具的用途和参数含义。 ...为什么强调统一签名因为模型输出的 JSON 参数是不可控的如果每个工具的参数风格完全不一样Agent 层就得写大量解析和容错代码。统一签名之后Agent 层可以用**kwargs的方式把参数直接传给工具函数再做异常处理。3.3 Prompt 设计Agent 系统的效果很大程度上取决于 Prompt 是否能清晰描述工具。模型需要知道当前有哪些工具可用。每个工具是做什么的。什么时候该用工具什么时候该直接回答。参数格式是什么。一套可复用的 Prompt 模板通常包含这几段System 角色设定可用工具列表名称、描述、参数 schema历史对话用户当前输入在下面的实战代码中我会把工具描述动态渲染到 Prompt 里这样新增工具时不需要改模型调用逻辑。3.4 上下文与记忆管理很多人一开始会忽略上下文长度问题。多轮对话中如果每次都把所有历史消息传给模型很快会超出上下文窗口限制。embabel-agent 初步可以只保留最近 N 轮对话核心原则是保留最近的用户意图和系统回答。截断过长的历史。对关键业务数据比如订单 ID、用户名可以考虑直接从历史中提取后放到“当前会话变量”里而不是完全依赖模型记忆。在工程上可以先从“保留最近 6 轮”开始后续再根据业务场景引入向量记忆或摘要记忆。4. 完整实战实现一个 embabel-agent 服务接下来我们正式写代码。我会按照“工具层 - Agent 层 - API 层”的顺序展开。4.1 实现工具层文件路径tools.py# tools.py from typing import Dict, Callable import datetime import json import random def get_current_time() - str: 返回当前系统时间格式为 YYYY-MM-DD HH:MM:SS return datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S) def get_order_status(order_id: str) - str: 根据订单ID查询订单状态。 参数: order_id: 订单ID字符串类型例如 10086 返回: 订单状态描述例如 已发货、已签收、待支付 # 示例实现实际项目中这里通常要调用订单服务 status_map { 10086: 已发货正在运输途中, 10010: 已完成签收, 10000: 待支付, } status status_map.get(order_id, 未查询到该订单) return json.dumps({order_id: order_id, status: status}, ensure_asciiFalse) def calculate_express_fee(weight: float, city: str) - str: 根据包裹重量和目的城市计算运费。 参数: weight: 包裹重量单位千克浮点数 city: 目的城市名称字符串 返回: 运费结果单位元 base 8.0 if city in (北京, 上海, 广州, 深圳): base 10.0 total base max(0, weight - 1) * 2 return json.dumps({weight: weight, city: city, fee: round(total, 2)}, ensure_asciiFalse) # 工具注册表 TOOLS: Dict[str, Callable] { get_current_time: get_current_time, get_order_status: get_order_status, calculate_express_fee: calculate_express_fee, } def get_tool_descriptions() - str: 生成工具描述文本用于注入到 Prompt 中。 descriptions [] for name, func in TOOLS.items(): descriptions.append(f- {name}: {func.__doc__}) return \n.join(descriptions)这里我用get_tool_descriptions动态生成工具描述模型可以从描述里知道每个工具的作用。需要注意calculate_express_fee和get_order_status返回的是 JSON 字符串这比直接返回一个 Python 对象更稳定因为模型看到的是文本理解成本更低。4.2 实现 Agent 核心文件路径agent_core.py# agent_core.py import json from typing import List, Dict, Any from openai import OpenAI from tools import TOOLS, get_tool_descriptions SYSTEM_PROMPT_TEMPLATE 你是一个嵌入在业务系统中的智能助手你的代号是 embabel-agent。 你可以使用以下工具来完成用户任务 {tool_descriptions} 请根据用户问题判断是否需要调用工具 - 如果用户问题需要实时数据请返回一个 JSON格式如下 {{action: call_tool, tool_name: 工具名, arguments: {{参数名: 参数值}}}} - 如果用户问题可以直接回答请返回一个 JSON格式如下 {{action: reply, content: 你的回答}} 注意 1. 你只能返回 JSON不要输出其他解释性文字。 2. 如果无法确定参数值请使用合理默认值或向用户询问。 3. 如果工具返回结果请结合结果给用户一个友好的自然语言回复。 class EmbabelAgent: def __init__(self, api_key: str, base_url: str None, model: str gpt-4o-mini): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model self.history: List[Dict[str, str]] [] def _build_messages(self, user_input: str) - List[Dict[str, str]]: system_prompt SYSTEM_PROMPT_TEMPLATE.format( tool_descriptionsget_tool_descriptions() ) messages [{role: system, content: system_prompt}] # 只保留最近 6 轮历史 messages.extend(self.history[-6:]) messages.append({role: user, content: user_input}) return messages def _call_llm(self, messages: List[Dict[str, str]]) - str: response self.client.chat.completions.create( modelself.model, messagesmessages, temperature0.2, ) return response.choices[0].message.content def _parse_tool_call(self, model_output: str) - Dict[str, Any]: 解析模型输出期望拿到 JSON。 output model_output.strip() # 去掉可能的 json 包裹 if output.startswith(): lines output.split(\n) lines lines[1:-1] if lines[-1].strip() else lines[1:] output \n.join(lines) try: parsed json.loads(output) except json.JSONDecodeError: return {action: reply, content: 系统暂时无法理解你的请求请稍后再试。} return parsed def run(self, user_input: str) - str: messages self._build_messages(user_input) model_output self._call_llm(messages) parsed self._parse_tool_call(model_output) if parsed.get(action) reply: reply parsed.get(content, 好的已收到你的消息。) elif parsed.get(action) call_tool: tool_name parsed.get(tool_name) arguments parsed.get(arguments, {}) if tool_name not in TOOLS: reply 当前没有找到可用工具请换个问题。 else: try: tool_result TOOLS[tool_name](**arguments) reply f工具 {tool_name} 返回结果{tool_result} except TypeError as e: reply f工具参数错误{e} except Exception as e: reply f工具执行失败{e} else: reply 模型返回格式异常请重新提问。 # 记录历史 self.history.append({role: user, content: user_input}) self.history.append({role: assistant, content: reply}) return reply这段代码是整个方案的核心有几个地方需要重点解释_build_messages负责把系统提示词、历史记录和用户输入组装成模型需要的消息格式。_parse_tool_call处理了模型输出可能带 Markdown 代码块的情况。大模型在返回 JSON 时经常会把内容包在 json 里面所以这里做了一层兼容处理。run方法里做了参数异常处理工具参数不合法时不会让服务直接崩溃。历史记录这里保存的是“原始工具结果”并没有让模型把结果重写成自然语言。如果希望最终回复更自然可以在拿到工具结果后再让模型进行一次“总结”。这里先保留原始结果便于排查。4.3 创建 HTTP 接口文件路径api.py# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from agent_core import EmbabelAgent from config import OPENAI_API_KEY, OPENAI_BASE_URL, MODEL_NAME app FastAPI(titleembabel-agent API, version1.0.0) agent EmbabelAgent( api_keyOPENAI_API_KEY, base_urlOPENAI_BASE_URL, modelMODEL_NAME, ) class ChatRequest(BaseModel): message: str session_id: str default class ChatResponse(BaseModel): reply: str session_id: str app.post(/chat, response_modelChatResponse) def chat(request: ChatRequest): if not request.message.strip(): raise HTTPException(status_code400, detail消息不能为空) # 这里 session_id 先只做透传后续可基于 session 维护独立记忆 reply agent.run(request.message) return ChatResponse(replyreply, session_idrequest.session_id) app.get(/health) def health(): return {status: ok}这里agent是全局单例。实际生产环境中如果服务会同时处理大量用户你需要考虑每个会话独立记忆这里先用简单方式演示。4.4 配置文件文件路径config.py# config.py import os OPENAI_API_KEY os.getenv(OPENAI_API_KEY, sk-xxx) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL, None) # 兼容 OpenAI 或国内兼容接口 MODEL_NAME os.getenv(MODEL_NAME, gpt-4o-mini)将模型密钥放到环境变量中避免写死在代码里。如果你使用的是国内模型平台的 OpenAI 兼容接口只需要设置OPENAI_BASE_URL和MODEL_NAME。4.5 运行与验证先创建一个简单的启动脚本也可以直接用 uvicorn 命令export OPENAI_API_KEY你的key uvicorn api:app --host 0.0.0.0 --port 8000 --reload看到类似下面的输出说明服务已经启动INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Application startup complete.然后使用 curl 发起一个请求curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 现在几点了, session_id: test1}如果模型判断需要调用get_current_time工具返回结果类似{ reply: 工具 get_current_time 返回结果2025-01-15 14:30:22, session_id: test1 }再试一个订单查询curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 帮我查一下订单10086状态, session_id: test1}这个例子中模型会调用get_order_status工具并返回订单状态 JSON 字符串。4.6 让回复更自然上面代码返回的是“工具调用结果”原文从用户体感上看不够友好。要让 Agent 把工具结果转成自然语言可以在拿到tool_result后再用模型做一次总结。修改run方法中对应的分支if parsed.get(action) call_tool: tool_name parsed.get(tool_name) arguments parsed.get(arguments, {}) if tool_name not in TOOLS: reply 当前没有找到可用工具请换个问题。 else: try: tool_result TOOLS[tool_name](**arguments) # 调用模型做一次总结 summary_messages [ {role: system, content: 你是一个友好客服助手请把工具返回内容转成简洁自然的中文回复。}, {role: user, content: f工具执行结果{tool_result}}, ] summary_response self.client.chat.completions.create( modelself.model, messagessummary_messages, temperature0.3, ) reply summary_response.choices[0].message.content except TypeError as e: reply f工具参数错误{e} except Exception as e: reply f工具执行失败{e}这样用户看到的不再是 JSON而是类似“您的订单 10086 已发货正在运输途中”这样的自然语言。5. 嵌入业务系统的三种常见方式embabel-agent 不止能作为一个独立 HTTP 服务存在它还可以以其他方式嵌入不同形态的业务系统。5.1 HTTP 直连调用这是最简单的方式也是上面示例所演示的。业务系统将用户消息 POST 到/chat接口拿到返回结果。适合Web 客服系统移动端 App 的后端内部管理后台这种方式需要重点考虑请求超时。大模型响应通常需要 1 到 5 秒工具调用多个轮次时可能更长。建议业务侧设置 10 到 30 秒的超时时间并增加重试机制。5.2 消息队列异步接入如果业务系统是事件驱动架构可以引入消息队列如 RabbitMQ、Kafka、Redis Stream来解耦。流程是业务系统把任务消息发送到队列。embabel-agent 消费消息调用大模型和相关工具。将结果写入结果队列或回调接口。业务系统异步获取结果。这种模式适合耗时长、或者不希望阻塞主流程的任务比如批量内容总结、工单自动分类、报表生成等。使用异步方案时要注意任务状态管理和幂等性。5.3 SDK 内嵌如果业务系统本身就是 Python 服务并且不想额外引入网络请求可以直接把agent_core.py和tools.py作为一个 SDK 模块引入。from agent_core import EmbabelAgent agent EmbabelAgent(api_keyxxx, modelgpt-4o-mini) result agent.run(帮我查一下订单10086状态) print(result)这样做的好处是调用延迟最低坏处是会让业务系统直接依赖大模型网络调用一旦模型服务出问题业务主流程也会受影响。所以 SDK 内嵌通常建议配合降级开关使用模型不可用时直接返回预设兜底话术。6. 常见问题与排查思路6.1 模型调用超时现象接口长时间无响应最终报超时错误。常见原因模型服务本身响应慢。Prompt 过长尤其是塞入大量工具描述和历史记录。网络链路不稳定。处理思路缩短历史记录轮数。精简工具描述必要时只把当前场景相关的工具描述传给模型。设置合理的超时时间建议客户端超时设为 10 到 30 秒。6.2 模型输出 JSON 解析失败现象服务端日志提示 JSONDecodeError或者返回“模型返回格式异常”。常见原因大模型在 JSON 前后加了 Markdown 代码块。Prompt 没有足够强地约束输出格式。模型版本或参数导致输出不稳定。处理思路在解析前做字符串清理去掉多余空白和代码块。强化 System Prompt明确要求“只返回 JSON不要解释”。适当降低 temperature建议设置在 0 到 0.3 之间。如果还不行可以考虑使用函数调用功能让模型按 schema 输出结构化结果而不是纯文本 JSON。6.3 工具参数错误现象工具执行时报 TypeError 或参数缺失。常见原因模型生成的参数名与函数参数名不一致。模型把数字参数当成字符串传递。Prompt 里的工具描述不够明确。处理思路在工具注册时补齐参数类型说明。在 Agent 层捕获参数异常并返回“工具参数错误”。对关键参数做类型兼容处理例如float(weight)强制转换。6.4 并发请求导致全局变量污染现象不同用户之间的对话历史相互串线。常见原因agent是全局单例self.history是实例变量。多个用户并发访问同一个实例会导致记忆混乱。处理思路按session_id维护会话上下文而不是把所有历史放在同一个 Agent 实例里。可以使用一个全局字典key 为session_idvalue 为EmbabelAgent实例。生产环境中建议使用 Redis 保存会话状态保证多实例部署时状态一致。6.5 问题排查清单1. 模型有没有返回 - 没有返回检查 key、base_url、网络、模型名。 2. 模型返回了但解析失败 - 查看原始输出清理代码块加强 Prompt。 3. 解析成功但工具没执行 - 检查工具名是否在注册表里。 4. 工具执行报错 - 查看参数类型和函数签名是否匹配。 5. 结果返回了但用户不满意 - 调整总结模型 Prompt或者把工具结果做得更结构化。7. 最佳实践与工程建议7.1 工具层独立于模型层设计工具时不要让工具函数直接依赖 OpenAI SDK 或 Agent 类。工具函数应该是纯 Python 函数输入参数、输出字符串。这样做的最大好处是你可以单独写单元测试来验证工具逻辑也可以让其他模型框架复用这套工具。工具函数建议统一返回 JSON 字符串而不是裸文本。后续解析更稳定日志也更易读。7.2 Prompt 与模型参数工具描述用中文还是英文取决于模型能力。中文业务场景下工具描述用中文往往效果更好。每个工具的描述要写清楚“什么时候用”和“参数含义”只写函数名是不够的。temperature建议控制在 0.3 以内减少随机性。如果模型支持 function calling优先使用原生 function calling 能力比让模型裸输出 JSON 更稳定。7.3 安全边界Agent 能调用工具就相当于拿到了执行能力。需要特别注意以下几点工具白名单只注册业务真正需要的能力不要盲目暴露所有内部接口。参数校验工具函数内部要对入参做合法性校验避免模型传入非法值。权限隔离在 API 层引入认证和授权至少要做到 token 校验避免未授权调用。敏感信息不要让 Agent 把内部数据库连接串、密钥、Token 等敏感信息写入工具描述或 Prompt。我这里再强调一次涉及生产环境的任何工具接入都要遵循最小权限原则并经过测试环境充分验证。7.4 日志与可观测性Agent 系统排障比普通接口难因为中间多了一次“模型决策”过程。建议在关键节点打日志收到用户请求 模型原始输出 解析后的结构化指令 工具执行结果 最终返回内容日志里最好带上session_id和请求耗时。这样一旦出现问题可以快速定位是模型决策错、解析错、还是工具执行错。7.5 生产落地顺序如果要在真实业务中落地 embabel-agent我建议按以下顺序推进先定义 3 到 5 个核心工具完成最小闭环。在测试环境跑通端到端流程重点看模型的工具选择准确率。加认证、限流、日志和监控。小流量灰度观察线上请求的成功率和耗时。再逐步扩展工具数量和 Prompt 场景。不要一上来就把几十个工具全部注册进去工具太多会让模型选择准确率下降Prompt 也会越来越长最后性能劣化。8. 总结与下一步本文围绕 embabel / embabel-agent 这个主题讲解了可嵌入智能代理的核心概念并提供了一个可直接运行的 FastAPI 实现。我们完成了工具注册、Agent 调度、HTTP 接口、结果解析以及自然语言总结等关键环节同时也整理了超时、JSON 解析、参数错误、会话隔离等高频问题的排查思路。这套方案在落地时最有价值的地方往往不是 Agent 本身而是工具层的沉淀。每接入一个工具就等于让业务系统多了一种可以被模型调用的能力日积月累Agent 能解决的问题范围会越来越大。如果你正准备让业务系统接入大模型能力我建议先不要急着上重型框架用 embabel-agent 这种轻量方式跑通一个真实场景再根据问题逐步完善。下一步可以尝试的方向包括接入 function calling 原生能力、使用 Redis 管理多用户会话、把工具执行改成异步任务、引入简单的模型输出校验机制。希望这篇文章能帮你少踩一些坑也欢迎你在自己的项目里动手试一下这套完整代码。