Nexent 智能体（Agent）开发核心指南：架构、执行引擎与工具集成

• 人工智能
• AI Agent
• Agent 编排
• 低代码
• AI 技能
• RAG

导读：本文是 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/目录，包含以下关键模块：

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_agent（nexent_agent.py）中可以看到完整的 Agent 装配链路：按agent_config.model_name创建模型 → 创建并监控包装工具列表 → 递归创建managed_agents子智能体（会话级沙箱容器可共享）→ 包装外部 A2A 智能体代理 → 构建ContextManager与ManagedContextRuntime→ 依据sandbox_config构建代码执行器。

1.2 CoreAgent：代码执行引擎

CoreAgent继承并增强了 SmolAgents 的CodeAgent（core_agent.py），提供：

• Python 代码执行：解析并执行模型生成的 Python 代码，动态处理任务；
• 多语言支持：内置中英文提示词模板，通过observer.lang切换；
• 流式输出：通过MessageObserver实现模型输出的实时流式显示；
• 步骤追踪：记录并展示每个执行步骤（step_metrics），便于调试和监控；
• 中断控制：内置stop_event（threading.Event），支持任务优雅停止；
• 错误处理：完善的重试与降级机制；
• 状态管理：维护和传递执行状态（_history_step_count、plan_repo等），支持复杂任务的连续处理。

ReAct 思考-行动-观察循环

CoreAgent 实现了 ReAct 框架的思考-行动-观察循环，这是问题解决型智能体的标准执行模式：

1. 思考（Think）：使用大语言模型生成解决方案代码；
2. 行动（Act）：执行生成的 Python 代码或调用工具；
3. 观察（Observe）：收集执行结果和日志；
4. 重复（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(lang="zh") stop_event = Event() model_config = ModelConfig( cite_name="gpt-4", api_key="<YOUR_API_KEY>", model_name="Qwen/Qwen2.5-32B-Instruct", url="https://api.siliconflow.cn/v1", ) agent_config = AgentConfig( name="example_agent", description="An example agent", tools=[], max_steps=5, model_name="gpt-4", ) agent_run_info = AgentRunInfo( query="strrawberry中出现了多少个字母r", model_config_list=[model_config], observer=observer, agent_config=agent_config, stop_event=stop_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内部通过DeferredAgentRun（run_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(role="user", content="你好"), AgentHistory(role="assistant", content="你好，我能帮你做什么？"), ] agent_run_info = AgentRunInfo( # ... history=history, )

优雅中断：

stop_event.set() # 智能体会在当前步完成后停止

2.4 运行期 Token 指标（TOKEN_COUNT）

agent_run_with_observer（nexent_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 核心配置参数

兼容性说明：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/PlanStep（agent_model.py），状态流转为pending → in_progress → completed / skipped。

计划持久化由PlanRepo承担（plan_repo.py）：Redis 优先、本地内存兜底。Redis key 格式为plan:{conversation_id}:{user_id}，默认 TTL 为 86400 秒（24 小时）；Redis 不可用时自动降级为进程内字典存储。Redis 客户端由后端通过AgentRunInfo.redis_client注入，SDK 不自行创建。

配套的ParallelExecutorTool（parallel_executor.py）支持并发调度多个子智能体/工具：

• 输入格式：任务列表中的每个任务为二元组(工具或智能体, 参数字典)或三元组(工具或智能体, 参数字典, 结果名称)，同一次调用不能混用两种格式；全部二元组返回按输入顺序排列的列表，全部三元组返回以结果名称为 key 的字典；
• 并发控制：max_workers（默认 4）控制最大并发线程数，timeout（默认 120 秒）为单任务超时，超时或失败时返回错误字符串，成功结果保留原始类型；
• 依赖约束：只允许并行执行互不依赖的任务，有依赖关系的调用必须串行。

启用规划能力：AgentConfig(enable_planning=True)，并可在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配置SandboxConfig（sandbox.py）后，可启用 Docker/WASM 级别的沙箱隔离：

scope设计值得注意：session提供严格的多租户隔离（并发运行互不干扰，但每次运行都有容器冷启动开销）；system则避免冷启动，通过每个运行独立内核实现内核态隔离，容器存活至应用关闭或故障。文档中的示例配置为：

# 示例：启用 Docker 会话级沙箱 {"level": "docker", "scope": "session", "docker_image": "nexent/nexent-sandbox:latest"}

5.3 Shell 调用拦截

沙箱模式下，ShellPolicy.DISABLED（推荐默认）会在AST 解析阶段拦截subprocess（run/call/check_call/Popen/getoutput等）与os（system/popen/execv/spawn*等）的 shell 调用，见 sandbox.py。同时拦截 IPython shell 转义（!...）、shell magic（%system等）与 cell magic（%%bash等）。例外：合法的无 shell 参数列表pip install（argv 形式）可放行用于依赖安装。

六、Guardrail 安全筛查

内置输入/输出内容安全检查点（guardrail checkpoints），敏感内容可拦截。配置位于AgentVerificationConfig.guardrail_config（agent_model.py）：

• GuardrailConfig.enabled：总开关，为 False 时不创建 GuardrailEngine；
• GuardrailRule：模式匹配规则（黑名单/白名单），每条规则编译为正则，在每个检查点执行匹配；
• severity：block（终止运行）/mask（脱敏替换）/pass（放行）；
• default_action：默认动作。

从源码看（core_agent.py），Guardrail 在运行时设有多处检查点：输入侧逐条消息筛查 LLM 输入（block 则终止运行并输出拒绝文案guardrail_input）、工具调用参数侧包装forward()进行参数筛查（block 抛出ToolInputBlockedError并以guardrail_clarification/guardrail_tool_input结束运行），确保敏感内容在进入模型或工具前被拦截。

七、MessageObserver 流式消息处理

7.1 设计要点

MessageObserver（observer.py）是消息观察者模式的核心实现，用于处理 Agent 的流式输出：

• 流式输出捕获：实时捕获模型生成的 token（<think>标签状态管理、代码块标记识别）；
• 过程类型区分：根据不同的处理阶段（模型输出、代码解析、执行日志等）格式化输出；
• 多语言支持：内置中英文输出模板（如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_RUN：Agent 基本信息；
• 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。

开发新工具的步骤

1. 在backend/tool_collection/mcp/local_mcp_service.py实现工具逻辑；
2. 用@mcp.tool()装饰器注册；
3. 重启 MCP 服务。

@mcp.tool(name="my_tool", description="我的自定义工具") def my_tool(param1: str, param2: int) -> str: # 实现工具逻辑 return f"处理结果: {param1} {param2}"

仓库中已有示例（local_mcp_service.py）：

@local_mcp_service.tool( name="test_tool_name", description="test_tool_description", ) async def demo_tool(para_1: str, para_2: int) -> str: print("tool is called successfully") print(para_1, para_2) return "success"

MCP 工具集成（运行侧）

在AgentRunInfo中通过mcp_host注入 MCP 服务地址，可以是字符串 URL 或包含url/transport的 dict：

agent_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 的实现一致）：

1. 推理：分析问题并制定方法；
2. 行动：执行工具或生成代码；
3. 观察：检查结果和输出；
4. 迭代：继续直到任务完成。

9.2 多智能体协作

• 分层智能体：管理智能体协调工作智能体——通过AgentConfig.managed_agents递归声明子智能体，create_single_agent中逐层构建并包装为可调用工具（见 nexent_agent.py）；
• 专业智能体：特定任务的领域专用智能体；
• 外部 A2A 智能体：通过external_a2a_agents配置，SDK 为其创建ExternalA2AAgentWrapper（a2a_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_cache（ModelConfig.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 智能体设计

1. 明确目标：定义具体、可测量的智能体目标（写入AgentConfig.description与提示词）；
2. 适当工具：选择与智能体能力匹配的工具，通过tools/managed_agents声明，避免过度暴露；
3. 强大提示词：通过prompt_templates/instructions创建全面的系统提示词；
4. 错误处理：利用AgentVerificationConfig（分层 ReAct 自校验，strictness默认balanced、pass_score默认0.75、fail_policy默认repair_then_controlled_summary）与 Guardrail 实现全面错误恢复与内容安全。

11.2 开发工作流

1. 迭代开发：用agent_run增量构建和测试，逐条消费事件流验证行为；
2. 提示词工程：基于测试结果优化提示词与instructions；
3. 工具测试：集成前先通过@mcp.tool()注册并单独验证单个工具；
4. 性能测试：通过TOKEN_COUNT指标监控和优化执行速度与上下文占用。

11.3 生产部署

1. 资源分配：配置SandboxConfig（memory_limit_mb、cpu_quota、timeout_seconds），确保充足且受控的计算资源；
2. 监控设置：启用 monitoring 上下文（AgentRunMetadata、trace_tool_call/trace_retriever_call）实现全面的日志和告警；
3. 扩展策略：通过ParallelExecutorTool与concurrency_limit规划增加的负载；
4. 安全考虑：开启 Guardrail（severity=block拦截敏感内容）、沙箱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 Agent
• Agent 编排
• 低代码
• AI 技能
• RAG