★ FEATURED ARTICLE 人工智能AI AgentAgent 编排低代码AI 技能RAG【免费下载链接】nexentNexent 是一个开源智能体SDK和平台能够将描述流程的自然语言转化为完整的多模态服务 —— 无需编排无需复杂拖拉拽。基于 MCP 工具生态系统构建Nexent 提供灵活的模型集成、可扩展的数据处理和强大的知识库管理。我们的目标很简单将数据、模型和工具整合到一个智能中心中让任何人都能轻松地将 Nexent 集成到项目中使日常工作流程更智能、更互联。项目地址https://gitcode.com/ModelEngine/nexent点击查看免费下载导读本文是 Nexent 开源智能体 SDK 的 Agent 开发核心指南围绕 doc/docs/zh/sdk/core/agents.md 展开系统讲解 NexentAgent 与 CoreAgent 的架构设计、ReAct 思考-行动-观察循环、上下文压缩、沙箱隔离、Guardrail 安全筛查、MessageObserver 流式消息机制以及基于 MCP 的自定义工具开发流程。读完本文你将能够基于sdk/nexent/core/agents下的源码独立构建、运行并调试一个支持流式输出、上下文管理与多智能体协作的企业级 Agent。一、智能体架构总览Nexent 提供完整的 AI 智能体开发和部署框架核心能力涵盖工具集成、推理Reasoning与多模态交互。SDK 侧的智能体实现位于sdk/nexent/core/agents/目录包含以下关键模块模块文件职责nexent_agent.pyNexentAgent工厂模型创建、工具组装、CoreAgent 构建与流式运行编排core_agent.pyCoreAgent继承 SmolAgentsCodeAgent的代码执行引擎run_agent.pyagent_run异步生成器后台线程执行 事件流输出observer.pyMessageObserver/ProcessType流式消息捕获与分类context/config.pyContextManagerConfig上下文管理与压缩配置sandbox.pySandboxConfigDocker/WASM 沙箱隔离配置plan_repo.py计划持久化Redis 优先本地内存兜底agent_model.pyAgentConfig/AgentRunInfo/ModelConfig等数据模型1.1 NexentAgent企业级智能体工厂NexentAgent是智能体系统的核心入口nexent_agent.py其构造参数覆盖了企业级运行所需的全部上下文多模型支持model_config_list传入多个ModelConfig通过create_model(model_cite_name)按cite_name匹配并实例化 OpenAI、视觉语言模型VLM或长上下文模型MCP 集成mcp_tool_collection挂载 Model Context Protocol 工具生态模型可调用外部 MCP 服务动态工具加载支持本地工具backend/tool_collection/与 MCP 工具的动态创建与管理create_single_agent中通过create_tool(tool_config)逐个实例化并包装监控分布式执行基于线程池sdk/nexent/core/concurrency/与异步架构的高性能执行引擎stop_event与cancellation_scope提供中断控制状态管理任务状态追踪、workspace_path/workspace_run_id运行工作区管理与错误恢复机制cleanup_run_workspace在 nexent_agent.py。在create_single_agentnexent_agent.py中可以看到完整的 Agent 装配链路按agent_config.model_name创建模型 → 创建并监控包装工具列表 → 递归创建managed_agents子智能体会话级沙箱容器可共享→ 包装外部 A2A 智能体代理 → 构建ContextManager与ManagedContextRuntime→ 依据sandbox_config构建代码执行器。1.2 CoreAgent代码执行引擎CoreAgent继承并增强了 SmolAgents 的CodeAgentcore_agent.py提供Python 代码执行解析并执行模型生成的 Python 代码动态处理任务多语言支持内置中英文提示词模板通过observer.lang切换流式输出通过MessageObserver实现模型输出的实时流式显示步骤追踪记录并展示每个执行步骤step_metrics便于调试和监控中断控制内置stop_eventthreading.Event支持任务优雅停止错误处理完善的重试与降级机制状态管理维护和传递执行状态_history_step_count、plan_repo等支持复杂任务的连续处理。ReAct 思考-行动-观察循环CoreAgent 实现了 ReAct 框架的思考-行动-观察循环这是问题解决型智能体的标准执行模式思考Think使用大语言模型生成解决方案代码行动Act执行生成的 Python 代码或调用工具观察Observe收集执行结果和日志重复Repeat根据观察结果继续思考和执行直到任务完成或达到max_steps。# 基本导入来自 doc/docs/zh/sdk/basic-usage.md from nexent.core.utils.observer import MessageObserver, ProcessType from nexent.core.agents.core_agent import CoreAgent from nexent.core.agents.nexent_agent import NexentAgent from nexent.core.models.openai_llm import OpenAIModel from nexent.core.tools import ExaSearchTool, KnowledgeBaseSearchTool二、Agent 运行流程与事件流机制2.1 推荐运行方式agent_run文档明确指出具体代码示例集中在 基本使用其中包含CoreAgent.run与流式的agent_run两种方式。当需要在服务端或前端以事件流方式消费消息时应使用agent_run它在后台线程执行智能体并从MessageObserver持续产出 JSON 字符串便于 UI 展示与日志采集。import json import asyncio from threading import Event from nexent.core.agents.run_agent import agent_run from nexent.core.agents.agent_model import AgentRunInfo, AgentConfig, ModelConfig from nexent.core.utils.observer import MessageObserver async def main(): observer MessageObserver(langzh) stop_event Event() model_config ModelConfig( cite_namegpt-4, api_keyYOUR_API_KEY, model_nameQwen/Qwen2.5-32B-Instruct, urlhttps://api.siliconflow.cn/v1, ) agent_config AgentConfig( nameexample_agent, descriptionAn example agent, tools[], max_steps5, model_namegpt-4, ) agent_run_info AgentRunInfo( querystrrawberry中出现了多少个字母r, model_config_list[model_config], observerobserver, agent_configagent_config, stop_eventstop_event ) async for message in agent_run(agent_run_info): message_data json.loads(message) print(message_data) # 每条都是 JSON 字符串 asyncio.run(main())消息流中的每条消息格式为type对应ProcessType、content文本内容、agent_name可选产出该消息的智能体。2.2 agent_run 真实调用链下图展示agent_run的真实调用链基于 run_agent.py、nexent_agent.py、core_agent.py从源码看agent_run内部通过DeferredAgentRunrun_agent.py在后台线程中等待请求准备完成后绑定AgentRunInfo并通过copy_context()复制上下文保证线程安全的运行隔离。MCP 工具的连接/执行超时分别由mcp_request_timeout_seconds与mcp_tool_timeout_seconds控制超时后会在事件流中输出本地化的中文错误提示见 run_agent.py。2.3 传入历史与优雅中断可选from nexent.core.agents.agent_model import AgentHistory history [ AgentHistory(roleuser, content你好), AgentHistory(roleassistant, content你好我能帮你做什么), ] agent_run_info AgentRunInfo( # ... historyhistory, )优雅中断stop_event.set() # 智能体会在当前步完成后停止2.4 运行期 Token 指标TOKEN_COUNTagent_run_with_observernexent_agent.py在每个 ActionStep 结束后向 observer 发送ProcessType.TOKEN_COUNT事件载荷包含step_number、duration、step_input_tokens、step_output_tokens、total_output_tokens、estimated_context_tokens、压缩相关指标compression_calls、compression_ratio等以及模型侧的 prompt cache 命中情况provider_cache_status、provider_cache_hit。这些指标可被前端直接消费用于展示每步的 token 消耗与上下文压缩效果。三、上下文管理ContextManager 与自适应压缩通过AgentConfig.context_manager_config对应ContextManagerConfig见 context/config.py开启上下文管理与压缩NexentAgent据此构建ContextManager与ManagedContextRuntime见 nexent_agent.py。3.1 核心配置参数参数默认值说明token_threshold10000上下文压缩阈值超过后触发历史压缩context_window_tokens10000模型级上下文窗口容量用于用户侧窗口占用展示effective_input_limit_tokens0有效输入上限0 表示不限制compaction_trigger_threshold_tokens0自适应压缩触发阈值0 时由effective_input_limit_tokens×compaction_trigger_ratio推导compaction_target_tokens0压缩目标 token 数0 时由effective_input_limit_tokens×compaction_target_ratio推导compaction_trigger_ratio0.8触发压缩的输入占比compaction_target_ratio0.6压缩后目标输入占比minimum_history_reduction_ratio0.05最小历史缩减比例低于则不压缩minimum_history_reduction_tokens32最小历史缩减 token 数keep_recent_steps4压缩时保留的最近步骤数enable_long_term_memory_selectionTrue是否启用长期记忆选择chars_per_token1.5token 估算的字符/token 比例兼容性说明soft_input_budget_tokens/hard_input_budget_tokens是仅用于构造函数输入的兼容参数__post_init__会将其归一化为compaction_trigger_threshold_tokens/effective_input_limit_tokens旧阈值语义hard_input_budget_tokens已废弃并被忽略。3.2 压缩策略与产物约束长对话自动压缩上下文运行时基于 ContextItems 支持自适应压缩压缩产物为结构化 Markdown 摘要summary_system_prompt要求输出# Compact Result of History顶层标题与按字段划分的## Section并拒绝可执行输出不允许输出代码围栏中的可执行内容增量压缩incremental_summary_system_prompt支持对已有摘要进行增量更新保留旧摘要信息并合并新轮次内容JSON Schema 摘要字段summary_json_schema定义了七个摘要字段——task_overview核心请求与成功标准、completed_work已完成工作、key_decisions重要决策及原因、unresolved_issues未解决问题、pending_items待办事项、next_steps下一步行动、context_to_preserve需保留的偏好与上下文并带有字数额度约束指标记录压缩发生后通过ProcessType.HISTORY_SUMMARY事件向前端广播压缩检查点运行期的TOKEN_COUNT事件携带compression_*字段记录压缩调用次数、输入/输出 token 与压缩比。四、Planning Agent先规划后执行Nexent 支持先规划后执行通过CreatePlanTool/UpdatePlanStepTool见 plan_tools.py生成与维护计划待办列表计划对象模型为AgentPlan/PlanStepagent_model.py状态流转为pending → in_progress → completed / skipped。计划持久化由PlanRepo承担plan_repo.pyRedis 优先、本地内存兜底。Redis key 格式为plan:{conversation_id}:{user_id}默认 TTL 为 86400 秒24 小时Redis 不可用时自动降级为进程内字典存储。Redis 客户端由后端通过AgentRunInfo.redis_client注入SDK 不自行创建。配套的ParallelExecutorToolparallel_executor.py支持并发调度多个子智能体/工具输入格式任务列表中的每个任务为二元组(工具或智能体, 参数字典)或三元组(工具或智能体, 参数字典, 结果名称)同一次调用不能混用两种格式全部二元组返回按输入顺序排列的列表全部三元组返回以结果名称为 key 的字典并发控制max_workers默认 4控制最大并发线程数timeout默认 120 秒为单任务超时超时或失败时返回错误字符串成功结果保留原始类型依赖约束只允许并行执行互不依赖的任务有依赖关系的调用必须串行。启用规划能力AgentConfig(enable_planningTrue)并可在AgentRunInfo中传入redis_client以启用 Redis 持久化。五、代码执行沙箱5.1 默认行为代码执行默认使用本地LocalPythonExecutor不启用沙箱。该执行器通过get_local_python_authorized_imports()nexent_agent.py限定可导入模块在 SmolAgents 内置基础模块之上额外放行math、re、json、datetime、statistics、decimal、fractions、random、collections、itertools、functools、heapq、bisect、array、copy、string、textwrap、unicodedata、time、calendar、base64、hashlib、hmac、csv、uuid、pprint、operator、typing等安全模块排除文件修改与系统访问类模块注释明确说明 Safe base imports for Python interpreter - excludes file modification and system access modules。5.2 SandboxConfig 沙箱配置通过AgentConfig.sandbox_policy/AgentRunInfo.sandbox_config配置SandboxConfigsandbox.py后可启用 Docker/WASM 级别的沙箱隔离参数默认值说明levellocal隔离级别local/docker/wasm安全性递增见 SandboxLevelscopesession容器生命周期session默认每次运行一个容器运行结束销毁system系统级共享持久容器每个运行使用独立 Jupyter 内核见 SandboxScopedocker_imagenexent/nexent-sandbox:latest沙箱容器镜像memory_limit_mb2048内存上限MBcpu_quota1.0CPU 配额network_disabledTrue是否禁用网络timeout_seconds120单次代码执行超时shell_policydisabledShell 策略disabled/restricted/boxed见 ShellPolicyoutput_dir/home/sandbox/workdir/output沙箱输出目录auto_sync_outputsTrue是否自动同步输出为 True 时需提供minio_clientscope设计值得注意session提供严格的多租户隔离并发运行互不干扰但每次运行都有容器冷启动开销system则避免冷启动通过每个运行独立内核实现内核态隔离容器存活至应用关闭或故障。文档中的示例配置为# 示例启用 Docker 会话级沙箱 {level: docker, scope: session, docker_image: nexent/nexent-sandbox:latest}5.3 Shell 调用拦截沙箱模式下ShellPolicy.DISABLED推荐默认会在AST 解析阶段拦截subprocessrun/call/check_call/Popen/getoutput等与ossystem/popen/execv/spawn*等的 shell 调用见 sandbox.py。同时拦截 IPython shell 转义!...、shell magic%system等与 cell magic%%bash等。例外合法的无 shell 参数列表pip installargv 形式可放行用于依赖安装。六、Guardrail 安全筛查内置输入/输出内容安全检查点guardrail checkpoints敏感内容可拦截。配置位于AgentVerificationConfig.guardrail_configagent_model.pyGuardrailConfig.enabled总开关为 False 时不创建 GuardrailEngineGuardrailRule模式匹配规则黑名单/白名单每条规则编译为正则在每个检查点执行匹配severityblock终止运行/mask脱敏替换/pass放行default_action默认动作。从源码看core_agent.pyGuardrail 在运行时设有多处检查点输入侧逐条消息筛查 LLM 输入block 则终止运行并输出拒绝文案guardrail_input、工具调用参数侧包装forward()进行参数筛查block 抛出ToolInputBlockedError并以guardrail_clarification/guardrail_tool_input结束运行确保敏感内容在进入模型或工具前被拦截。七、MessageObserver 流式消息处理7.1 设计要点MessageObserverobserver.py是消息观察者模式的核心实现用于处理 Agent 的流式输出流式输出捕获实时捕获模型生成的 tokenthink标签状态管理、代码块标记识别过程类型区分根据不同的处理阶段模型输出、代码解析、执行日志等格式化输出多语言支持内置中英文输出模板如StepCountTransformer的**步骤 {0}**/**Step {0}**统一接口通过MessageTransformer基类与DefaultTransformer/ParseTransformer/ExecutionLogsTransformer/FinalAnswerTransformer/TokenCountTransformer等具体实现为不同来源的消息提供统一处理方式线程安全message_query队列配_message_query_lock子智能体嵌套深度与调用栈通过ContextVar隔离避免并发工具执行互相干扰前端嵌套层级见 observer.py。7.2 ProcessType 常用处理阶段ProcessType枚举observer.py的常用处理阶段包括STEP_COUNT当前执行步骤MODEL_OUTPUT_THINKING模型思考过程输出MODEL_OUTPUT_DEEP_THINKING模型深度思考输出MODEL_OUTPUT_CODE模型代码生成输出PARSE代码解析结果EXECUTION_LOGS代码执行结果AGENT_NEW_RUNAgent 基本信息AGENT_FINISH子智能体运行结束标记主要用于前端展示FINAL_ANSWER最终总结结果SEARCH_CONTENT搜索结果内容PICTURE_WEB联网搜索图片处理结果TOKEN_COUNT每步 token 使用量记录HISTORY_SUMMARY新建的上下文压缩检查点ERROR/WARNING错误与可恢复告警SUBAGENT_START/SUBAGENT_END子智能体调用边界前端嵌套分组PLAN/PLAN_STEP_UPDATE规划结构化输出与单步状态更新CLARIFICATION相关HUMAN_INTERACTION澄清问答形式。八、工具集成与自定义工具开发Nexent 基于 Model Context Protocol (MCP) 实现工具系统。后端本地工具 MCP 服务位于 backend/tool_collection/mcp/local_mcp_service.py基于FastMCP构建并挂载了nl2agent_mcp_service。开发新工具的步骤在backend/tool_collection/mcp/local_mcp_service.py实现工具逻辑用mcp.tool()装饰器注册重启 MCP 服务。mcp.tool(namemy_tool, description我的自定义工具) def my_tool(param1: str, param2: int) - str: # 实现工具逻辑 return f处理结果: {param1} {param2}仓库中已有示例local_mcp_service.pylocal_mcp_service.tool( nametest_tool_name, descriptiontest_tool_description, ) async def demo_tool(para_1: str, para_2: int) - str: print(tool is called successfully) print(para_1, para_2) return successMCP 工具集成运行侧在AgentRunInfo中通过mcp_host注入 MCP 服务地址可以是字符串 URL 或包含url/transport的 dictagent_run_info AgentRunInfo( # ... mcp_host[http://localhost:3000], # 或包含 url/transport 的 dict )从 agent_model.py 的字段说明可知transport支持sse与streamable-http若传字符串则按 URL 后缀自动探测——以/sse结尾用sse以/mcp结尾用streamable-http认证可通过authorization如Bearer token或headersdict 提供。九、智能体执行模式9.1 ReAct 模式问题解决智能体的标准执行模式与 CoreAgent 的实现一致推理分析问题并制定方法行动执行工具或生成代码观察检查结果和输出迭代继续直到任务完成。9.2 多智能体协作分层智能体管理智能体协调工作智能体——通过AgentConfig.managed_agents递归声明子智能体create_single_agent中逐层构建并包装为可调用工具见 nexent_agent.py专业智能体特定任务的领域专用智能体外部 A2A 智能体通过external_a2a_agents配置SDK 为其创建ExternalA2AAgentWrappera2a_agent_proxy.py以 HTTP 方式调用模型可以像调用工具一样调用外部智能体通信协议智能体间通过标准化的消息传递SUBAGENT_START/SUBAGENT_END事件标记嵌套调用边界。9.3 错误处理和恢复优雅降级工具失败时的备选策略如 MCP 超时输出本地化错误提示、PlanRepo的 Redis→本地内存降级状态持久化通过PlanRepo保存计划状态AgentRunInfo携带conversation_id/user_id用于运行级持久化重试机制带退避策略的自动重试模型侧retry.py实现output_protocol协议错误支持enable_protocol_repair_retry。十、性能优化10.1 执行效率并行工具执行ParallelExecutorTool并发调度独立工具/子智能体max_workers控制并发度缓存策略模型侧prompt_cacheModelConfig.prompt_cache与 provider 侧 prompt cache 命中指标provider_cache_status/provider_cache_hit见 nexent_agent.py资源管理concurrency_limit/concurrency_key控制模型并发上限SandboxConfig.memory_limit_mb/cpu_quota限制沙箱资源。10.2 监控和调试执行跟踪agent_run_with_observer在monitoring_manager.start_agent_run/trace_agent_step上下文中记录每步指标工具调用自动包装监控 span_wrap_tool_with_monitoring见 nexent_agent.py性能指标TOKEN_COUNT事件输出每步耗时、输入/输出 token、上下文估计与压缩指标调试模式开发时可通过 observer 的STEP_COUNT/PARSE/EXECUTION_LOGS事件观察完整执行过程。十一、最佳实践11.1 智能体设计明确目标定义具体、可测量的智能体目标写入AgentConfig.description与提示词适当工具选择与智能体能力匹配的工具通过tools/managed_agents声明避免过度暴露强大提示词通过prompt_templates/instructions创建全面的系统提示词错误处理利用AgentVerificationConfig分层 ReAct 自校验strictness默认balanced、pass_score默认0.75、fail_policy默认repair_then_controlled_summary与 Guardrail 实现全面错误恢复与内容安全。11.2 开发工作流迭代开发用agent_run增量构建和测试逐条消费事件流验证行为提示词工程基于测试结果优化提示词与instructions工具测试集成前先通过mcp.tool()注册并单独验证单个工具性能测试通过TOKEN_COUNT指标监控和优化执行速度与上下文占用。11.3 生产部署资源分配配置SandboxConfigmemory_limit_mb、cpu_quota、timeout_seconds确保充足且受控的计算资源监控设置启用 monitoring 上下文AgentRunMetadata、trace_tool_call/trace_retriever_call实现全面的日志和告警扩展策略通过ParallelExecutorTool与concurrency_limit规划增加的负载安全考虑开启 Guardrailseverityblock拦截敏感内容、沙箱ShellPolicy.DISABLED阻断 shell 调用验证工具输入并保护 API 访问。十二、延伸阅读基本使用agent_run 流式运行与快速开始 —— CoreAgent.run 与流式 agent_run 的完整代码示例模型架构指南 —— OpenAIModel / VLM / 长上下文模型与ModelConfig详解工具开发指南 —— MCP 工具系统深入讲解多模态交互 —— 视觉语言模型与多模态输入输出开发者指南 —— 详细的实现示例和高级模式。SDK 源码与配置均可在当前仓库中直接查阅智能体核心实现见 sdk/nexent/core/agents/流式消息机制见 sdk/nexent/core/utils/observer.py工具集合见 sdk/nexent/core/tools/ 与 backend/tool_collection/。赞分享人工智能AI AgentAgent 编排低代码AI 技能RAG【免费下载链接】nexentNexent 是一个开源智能体SDK和平台能够将描述流程的自然语言转化为完整的多模态服务 —— 无需编排无需复杂拖拉拽。基于 MCP 工具生态系统构建Nexent 提供灵活的模型集成、可扩展的数据处理和强大的知识库管理。我们的目标很简单将数据、模型和工具整合到一个智能中心中让任何人都能轻松地将 Nexent 集成到项目中使日常工作流程更智能、更互联。项目地址https://gitcode.com/ModelEngine/nexent点击查看免费下载相关推荐Nexent 智能体开发全景指南Agent 架构、ReAct 执行内核与流式编排实战Nexent 智能体开发全景指南Agent 架构、ReAct 执行内核与流式编排实战 Nexent SDK 为开发者提供了一套面向生产环境的 AI Agent人工智能AI AgentAgent 编排低代码AI 技能RAGNexent AI 智能体开发指南从 ReAct 核心架构到工具、沙箱与流式运行的完整实战Nexent AI 智能体开发指南从 ReAct 核心架构到工具、沙箱与流式运行的完整实战 导读 本指南以 Nexent SDK 的智能体模块 doc/doAI AgentAI 应用后端前端大模型RAGNexent Agent 开发指南从 ReAct 核心架构到流式执行与沙箱隔离的完整实践Nexent Agent 开发指南从 ReAct 核心架构到流式执行与沙箱隔离的完整实践 Nexent 是一个基于 Harness Engineering 理AI AgentAI 应用后端前端大模型RAG创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
阅读完成 · 觉得有帮助？
 高效批处理实战指南)
