恒美微站 Logo 恒美微站
  • 首页
  • 关于我们
  • 建站服务
  • 主题模板
  • 案例展示
  • 资讯中心
  • 联系我们

.NET AI智能体集成Python代码执行引擎:安全架构与工程实践

  • 首页
  • 资讯中心
  • /
  • .NET AI智能体集成Python代码执行引擎:安全架构与工程实践

相关资讯

20天阳台小龙虾家庭微养殖:从零搭建稳定水生生态系统 2026/8/16 13:24:34
大地水准面与高程异常的球谐级数展开形式 2026/8/16 13:19:33
老款 Mac 再战三年的完整指南:用 OpenCore Legacy Patcher 免费跑最新 macOS,真的可行吗 2026/8/16 13:19:33

最新资讯

电脑搜索应用找不到如何解决
Havennlon | 杂谈:AI补出来的,是表达完整性,不是认知完整性
syscall_intercept 系统调用拦截:多线程与 clone 特殊逻辑(magic syscalls)全解析
Revit模型转换怎么做?一份Web3D可视化上手的完整指南
聊天记录永久保存实战:零基础用WeChatMsg把微信数据完整搬回家
网盘直链下载助手实测:告别限速焦虑,8大网盘真实链接一步到位

今日推荐

【文章复现】非线性值迭代自适应动态规划(ADP):离散时间非线性系统的策略迭代自适应动态规划算法研究附Matlab代码
【双层规划,节点出清价,绿证交易,CVaR方法】两级电力市场环境下计及风险的省间交易商最优购电模型附Matlab代码
隐式mpc+自适应mpc+时变mpc,线性时变模型预测控制附Simulink仿真

本周热门

【文章复现】非线性值迭代自适应动态规划(ADP):离散时间非线性系统的策略迭代自适应动态规划算法研究附Matlab代码
【双层规划,节点出清价,绿证交易,CVaR方法】两级电力市场环境下计及风险的省间交易商最优购电模型附Matlab代码
隐式mpc+自适应mpc+时变mpc,线性时变模型预测控制附Simulink仿真

本月精选

如何用DamaiHelper实现演唱会门票的智能自动化抢购:完整技术解决方案指南
第4篇:59 倍性能差距的索引瓶颈定位——一次教科书级的全表扫描调优
终极歌词批量下载神器:5分钟解决离线音乐库歌词同步难题

.NET AI智能体集成Python代码执行引擎:安全架构与工程实践

发布时间:2026/8/16 13:24:34
.NET AI智能体集成Python代码执行引擎:安全架构与工程实践 1. 项目概述为AI智能体注入代码执行引擎最近在折腾一个挺有意思的项目核心目标很明确在一个基于 .Net 的 AgentFramework 里让我的 AI 智能体Agent不仅能“思考”和“对话”还能“动手”执行 Python 脚本和运行代码。这听起来像是给一个只会出主意的军师配上了一支能征善战的军队。最终我希望这个能力能无缝对接像 openClaw 这样的技能平台让智能体真正具备调用外部工具、处理复杂任务的能力。为什么非得是 Python在当前的 AI 应用生态里Python 几乎是事实上的标准语言。从数据处理、机器学习模型调用到网络爬虫、自动化办公大量的工具库和技能包都是用 Python 写的。一个不能执行 Python 的 AI 智能体就像被束缚了双手空有大脑而无法直接改造世界。而 .Net 环境特别是结合了 AgentFramework 之后为我们提供了构建稳定、可扩展智能体应用的基础框架。将 Python 的执行能力嵌入到这个框架中相当于打通了“企业级应用开发”与“灵活脚本能力”之间的任督二脉。这个需求并非空穴来风。无论是想做一个能自动分析日志、生成报表的运维助手还是一个能根据自然语言描述自动编写简单数据处理脚本的编程副驾亦或是为 openClaw 技能库增加一个“自定义脚本执行”的技能其底层都需要这个核心能力。我见过很多团队在尝试构建此类功能时要么采用笨重的进程间通信要么就是安全性一塌糊涂。我希望通过这次实践找到一个既安全、高效又易于集成和维护的方案。2. 核心思路与架构设计2.1 能力边界与核心挑战在动手之前我们必须想清楚几个关键问题这决定了整个架构的走向。首先安全是头等大事。让 AI 智能体执行任意代码无异于打开了一个潘多拉魔盒。我们必须假设 AI 生成的或用户提交的代码可能是恶意的。因此代码执行必须在一个严格受限的沙箱环境中进行。这个沙箱需要限制网络访问、文件系统操作尤其是写操作、进程创建以及敏感的系统调用。在 .Net 环境下我们无法像在 Linux 中那样方便地使用seccomp或namespaces这就需要我们寻找或构建合适的隔离方案。其次执行环境的管理。Python 版本众多3.7, 3.8, 3.9...依赖库更是浩如烟海。我们的智能体可能需要为一个任务安装pandas和numpy为另一个任务安装requests和beautifulsoup4。如何管理这些相互可能冲突的依赖是为每个执行任务创建一个全新的、临时的虚拟环境还是维护一个公共的基础环境并动态安装依赖这涉及到执行效率和资源消耗的权衡。第三交互模式的设计。代码执行是“一锤子买卖”还是需要持续交互例如AI 智能体可能先执行一段代码来加载数据再根据结果执行第二段代码进行分析。这就需要保持一个 Python 解释器的会话Session状态。同时如何捕获标准输出、标准错误以及在代码运行超时或出错时如何优雅地终止都是必须考虑的问题。最后与 AgentFramework 的集成。这个能力最终要封装成 AgentFramework 中一个或多个可被智能体调用的“技能”Skill或“工具”Tool。它需要提供清晰的接口接收来自智能体决策引擎的指令包括代码文本和参数并返回结构化的结果成功/失败、输出内容、执行时间等。2.2 技术方案选型与理由基于以上挑战我设计了一套以“进程隔离 动态环境管理 会话保持”为核心的技术方案。1. 执行引擎Python.NET vs. 进程调用最初我考虑过使用 Python.NET 它允许在 .Net 进程中直接嵌入 CPython 解释器内存数据交换效率高。但经过评估我放弃了这条路主要原因有两个一是对 Python 环境的隔离能力几乎为零恶意代码可能直接破坏宿主 .Net 进程二是对第三方 C 扩展库的支持有时会出现兼容性问题管理起来复杂。因此我选择了更传统但更稳健的进程调用方式。通过System.Diagnostics.Process启动独立的python进程来执行代码。这样做的好处显而易见强隔离性每个代码执行任务都在独立的操作系统进程中运行崩溃了也不会影响主智能体服务。安全性可以结合操作系统级别的限制如 Windows 上的 Job Object 限制 CPU/内存Linux 上的ulimit和chroot思路的变体。灵活性可以自由选择 Python 解释器的路径轻松支持多版本。简单可靠技术栈简单调试方便社区案例丰富。2. 环境管理Conda 虚拟环境为了解决依赖问题我选择了Conda作为虚拟环境管理工具而不是 Python 自带的venv。原因在于 Conda 不仅能管理 Python 包还能管理非 Python 的二进制依赖比如某些机器学习库需要的 MKL 数学库这对于 AI 相关的任务尤其重要。我们的架构可以这样设计预置一个基础的“智能体 Python”环境包含pip,setuptools等必要工具。当智能体需要执行代码时系统根据任务 ID 或会话 ID从基础环境克隆出一个临时环境。代码执行前可以通过conda install或pip install在临时环境中安装所需的包。任务执行完毕后销毁这个临时环境释放资源。对于高频但轻量的任务可以考虑环境复用池来提升性能。3. 会话保持基于文件或内存的上下文传递对于需要多步交互的代码执行我们需要保持状态。一个简单有效的方法是使用临时文件作为上下文载体。第一步代码执行后将其产生的变量通过pickle或json序列化写入一个临时文件。第二步代码执行时先加载这个临时文件恢复上下文然后执行新代码。所有临时文件的生命周期与会话绑定会话结束即删除。对于更复杂的场景可以考虑运行一个轻量级的 Python RPC 服务如用flask或fastapi写个简单接口让 .Net 端通过 HTTP 与之交互但这会引入额外的复杂性和网络延迟。4. 安全沙箱基于进程和资源的限制这是方案中最关键的一环。我们通过多层防护来构建沙箱子进程限制在启动python进程时配置ProcessStartInfo重定向输入输出并设置一个严格的超时时间例如 30 秒。资源限额在 Linux 上可以通过prlimit在启动前设置进程的内存、CPU 时间限制。在 Windows 上可以使用 Job Object API通过 P/Invoke 调用来达到类似目的。模块黑名单在执行的 Python 代码外围包裹一个“安全层”脚本。这个脚本会先检查待执行代码是否导入了危险模块如os,subprocess,socket,shutil等或者使用ast模块进行简单的语法树分析禁止某些函数调用。注意这种方法并非绝对安全但能防住大部分无心之失或简单攻击。无网络、只读文件系统理想情况下沙箱环境应该没有网络访问权限并且对文件系统只有特定临时目录的读写权限。这通常需要更底层的容器技术如 Docker支持我们将其作为高级可选方案。注意没有任何一种纯软件沙箱是100%安全的。对于执行完全不可信的用户代码最安全的方式是放在一个独立的、资源受限的容器或虚拟机中。我们当前方案的目标是应对“智能体生成的、目标相对明确的代码”而非公开的、任意的用户代码执行服务。2.3 与 AgentFramework 及 openClaw 的集成构想在 AgentFramework 中我们通常将外部能力封装为ITool接口的实现。我们的 Python 执行引擎就可以是一个PythonScriptTool。这个工具需要定义清晰的输入和输出 Schema。例如输入code(字符串要执行的 Python 代码)、session_id(可选用于会话保持)、timeout_seconds(可选执行超时时间)、requirements(可选执行前需要安装的包列表)。输出一个结构化对象包含success(布尔值)、stdout(标准输出)、stderr(标准错误)、execution_time(执行耗时)、result(如果代码最后是一个表达式可以尝试捕获其值) 等字段。当智能体决定要使用这个工具时框架会调用ExecuteAsync方法传入参数然后由我们的引擎负责安全的执行并返回结果。至于对接openClaw其本质是一个技能Skill市场或调度平台。我们的PythonScriptTool本身就可以被包装成一个 openClaw Skill。这个 Skill 的描述会告诉 openClaw“我能执行 Python 代码”。当其他智能体或工作流通过 openClaw 调度到这个 Skill 时openClaw 就会将任务参数代码等转发给我们部署好的这个服务即集成了 Python 执行引擎的智能体执行完毕后再将结果返回。这样一来任何接入 openClaw 的智能体无需自己内置 Python 引擎就能通过技能调用的方式获得代码执行能力实现了能力的解耦和复用。3. 核心模块实现详解3.1 Python 执行引擎核心类设计下面我将深入核心代码部分。我们首先定义一个PythonExecutionEngine类它是整个功能的心脏。using System; using System.Diagnostics; using System.IO; using System.Threading; using System.Threading.Tasks; using System.Text; using System.Collections.Generic; namespace AgentFramework.PythonTools { public class PythonExecutionResult { public bool Success { get; set; } public string StandardOutput { get; set; } string.Empty; public string StandardError { get; set; } string.Empty; public TimeSpan ExecutionTime { get; set; } public Exception? Exception { get; set; } // 可以扩展用于存储序列化的执行结果 public string? ResultPayload { get; set; } } public interface IPythonExecutionEngine { TaskPythonExecutionResult ExecuteCodeAsync(string code, string? sessionId null, string[]? requirements null, int timeoutSeconds 30, CancellationToken cancellationToken default); Taskbool CleanupSessionAsync(string sessionId); } public class PythonExecutionEngine : IPythonExecutionEngine { private readonly string _pythonExecutablePath; private readonly string _tempDirectoryRoot; private readonly ILoggerPythonExecutionEngine _logger; // 会话上下文存储key为sessionIdvalue为该会话的临时工作目录 private readonly ConcurrentDictionarystring, string _sessionWorkspaces new(); public PythonExecutionEngine(string pythonExecutablePath, string tempDirectoryRoot, ILoggerPythonExecutionEngine logger) { _pythonExecutablePath pythonExecutablePath ?? throw new ArgumentNullException(nameof(pythonExecutablePath)); _tempDirectoryRoot tempDirectoryRoot ?? Path.Combine(Path.GetTempPath(), AgentPython); _logger logger; // 确保根目录存在 Directory.CreateDirectory(_tempDirectoryRoot); } // ... 核心方法在下面实现 } }这个类定义了执行结果容器PythonExecutionResult和引擎接口IPythonExecutionEngine。PythonExecutionEngine构造函数需要知道 Python 解释器的路径和一个用于存放所有临时文件的根目录。3.2 安全执行流程与代码注入ExecuteCodeAsync方法是重中之重它包含了从准备到清理的完整生命周期。public async TaskPythonExecutionResult ExecuteCodeAsync(string code, string? sessionId null, string[]? requirements null, int timeoutSeconds 30, CancellationToken cancellationToken default) { var result new PythonExecutionResult(); var stopwatch Stopwatch.StartNew(); string tempScriptPath string.Empty; string? sessionWorkspacePath null; try { // 1. 获取或创建会话工作空间 sessionWorkspacePath GetOrCreateSessionWorkspace(sessionId); // 2. 处理依赖安装 if (requirements ! null requirements.Length 0) { var installResult await InstallRequirementsAsync(requirements, sessionWorkspacePath, cancellationToken); if (!installResult.Success) { result.Success false; result.StandardError $Failed to install requirements: {installResult.Error}; return result; } } // 3. 生成安全的执行脚本 tempScriptPath Path.Combine(sessionWorkspacePath, $script_{Guid.NewGuid():N}.py); // 关键将用户代码包裹在安全控制和结果捕获的模板中 var wrappedCode GenerateSafeWrappedCode(code); await File.WriteAllTextAsync(tempScriptPath, wrappedCode, Encoding.UTF8, cancellationToken); // 4. 配置并启动进程 var processStartInfo new ProcessStartInfo { FileName _pythonExecutablePath, Arguments $\{tempScriptPath}\, // 注意路径转义 WorkingDirectory sessionWorkspacePath, // 限制工作目录 RedirectStandardOutput true, RedirectStandardError true, RedirectStandardInput true, UseShellExecute false, // 必须为false才能重定向IO和进行资源控制 CreateNoWindow true, // 重要可以在这里设置环境变量例如禁用PYTHONPATH以防止导入非预期模块 EnvironmentVariables { [PYTHONPATH] string.Empty } }; using var process new Process { StartInfo processStartInfo }; var outputBuilder new StringBuilder(); var errorBuilder new StringBuilder(); process.OutputDataReceived (sender, e) { if (e.Data ! null) outputBuilder.AppendLine(e.Data); }; process.ErrorDataReceived (sender, e) { if (e.Data ! null) errorBuilder.AppendLine(e.Data); }; _logger.LogInformation(Starting Python process for session {SessionId}, sessionId); if (!process.Start()) { throw new InvalidOperationException(Failed to start Python process.); } process.BeginOutputReadLine(); process.BeginErrorReadLine(); // 5. 异步等待进程结束支持超时和取消 var processTask Task.Run(() process.WaitForExit(), cancellationToken); var completedTask await Task.WhenAny(processTask, Task.Delay(TimeSpan.FromSeconds(timeoutSeconds), cancellationToken)); if (completedTask ! processTask) { // 超时处理 _logger.LogWarning(Python process timeout after {Timeout}s, killing it., timeoutSeconds); try { process.Kill(entireProcessTree: true); } catch { /* Ignore */ } result.Success false; result.StandardError $Execution timeout ({timeoutSeconds}s). Process was terminated.; } else if (process.ExitCode ! 0) { // 进程异常退出 result.Success false; result.StandardError $Python process exited with code {process.ExitCode}.\n{errorBuilder}; } else { // 执行成功 result.Success true; // 从输出中解析出我们模板捕获的结果如果有 var output outputBuilder.ToString(); // 简单示例假设最后一行是JSON格式的结果 // 实际中需要更鲁棒的协议比如用特殊标记分隔输出和结果 result.StandardOutput output; } result.StandardOutput outputBuilder.ToString().TrimEnd(); result.StandardError errorBuilder.ToString().TrimEnd(); } catch (OperationCanceledException) { _logger.LogInformation(Python execution was cancelled.); result.Success false; result.StandardError Execution was cancelled by user.; result.Exception new TaskCanceledException(); } catch (Exception ex) { _logger.LogError(ex, An error occurred during Python execution.); result.Success false; result.StandardError $Engine error: {ex.Message}; result.Exception ex; } finally { stopwatch.Stop(); result.ExecutionTime stopwatch.Elapsed; // 6. 清理临时脚本文件保留工作空间以供会话复用 try { if (File.Exists(tempScriptPath)) File.Delete(tempScriptPath); } catch (Exception ex) { _logger.LogWarning(ex, Failed to delete temp script: {Path}, tempScriptPath); } } return result; }这段代码逻辑清晰但最关键的一步是第3点的GenerateSafeWrappedCode。我们不是直接执行用户代码而是将其嵌入到一个我们控制的“安全壳”中。这个壳负责限制危险操作、捕获结果并控制输出格式。3.3 安全壳脚本与依赖管理让我们看看GenerateSafeWrappedCode和依赖安装的核心实现。private string GenerateSafeWrappedCode(string userCode) { // 这是一个简化的安全包装模板。 // 实际应用中应该更加强大例如使用 ast 模块进行静态分析。 string safeTemplate import sys import json import builtins # ---- 安全策略模块黑名单 ---- BLOCKED_MODULES {os, subprocess, socket, shutil, sys} # 示例可扩展 for mod in BLOCKED_MODULES: if mod in sys.modules: del sys.modules[mod] # 尝试卸载已导入的虽然很难完全阻止 sys.modules[mod] None # 防止导入 # ---- 重写危险的内置函数 ---- original_import __builtins__.__import__ def safe_import(name, *args, **kwargs): if name in BLOCKED_MODULES or any(name.startswith(f{blocked}.) for blocked in BLOCKED_MODULES): raise ImportError(fImport of module {name!r} is blocked for security reasons.) return original_import(name, *args, **kwargs) __builtins__.__import__ safe_import # ---- 用户代码执行区 ---- __exec_result__ None __exec_stdout__ try: # 重定向 stdout 以捕获打印内容 from io import StringIO old_stdout sys.stdout sys.stdout mystdout StringIO() # 执行用户代码 exec(r{USER_CODE}) # 获取所有局部变量尝试获取最后一个表达式的结果如果存在 local_vars locals() # 这里是一个简单的启发式方法如果用户代码是单个表达式可能会被 eval。 # 更复杂的实现需要解析代码。此处我们简单地将最后一条语句可能产生的值赋给 __exec_result__。 # 实际上更可靠的做法是要求用户代码将结果赋值给一个特定变量如 __result__ ... if __result__ in local_vars: __exec_result__ local_vars[__result__] __exec_stdout__ mystdout.getvalue() sys.stdout old_stdout except Exception as e: # 任何异常都会传播到外层被 .Net 引擎捕获 raise finally: # ---- 结果序列化与输出 ---- # 将结果和输出以特定格式打印供 .Net 端解析 output_data { stdout: __exec_stdout__, result: __exec_result__ } # 使用一个明确的标记来分隔引擎输出和用户输出 print(\n---AGENT_PYTHON_RESULT_START---) print(json.dumps(output_data, defaultstr)) # 使用 defaultstr 处理不可序列化对象 print(---AGENT_PYTHON_RESULT_END---) ; // 将用户代码进行转义防止破坏 Python 字符串字面量 var escapedCode userCode.Replace(\, \\).Replace(, \).Replace(\, \); return safeTemplate.Replace({USER_CODE}, escapedCode); }这个安全壳做了几件事1) 定义模块黑名单并尝试拦截导入2) 重写__import__函数进行运行时检查3) 使用exec执行用户代码4) 重定向stdout以捕获print输出5) 约定用户将最终结果赋值给__result__变量6) 将执行结果和输出以 JSON 格式包裹在特殊标记中打印。重要提示这个安全壳只是一个基础示例远非完美。Python 的动态性使得完全沙箱化极其困难例如用户可以通过().__class__.__bases__[0].__subclasses__()这样的方式绕过限制。对于生产环境强烈建议将不可信代码放在 Docker 容器中运行并配合严格的seccomp和capabilities配置。依赖安装InstallRequirementsAsync的实现则相对直接就是调用pip installprivate async Task(bool Success, string? Error) InstallRequirementsAsync(string[] requirements, string workspacePath, CancellationToken ct) { var tempReqFile Path.Combine(workspacePath, $requirements_{Guid.NewGuid():N}.txt); await File.WriteAllLinesAsync(tempReqFile, requirements, ct); var processStartInfo new ProcessStartInfo { FileName _pythonExecutablePath, Arguments $-m pip install -r \{tempReqFile}\ --quiet, WorkingDirectory workspacePath, RedirectStandardOutput true, RedirectStandardError true, UseShellExecute false, CreateNoWindow true, }; try { using var process Process.Start(processStartInfo); await process.WaitForExitAsync(ct); if (process.ExitCode 0) return (true, null); else return (false, await process.StandardError.ReadToEndAsync(ct)); } finally { File.Delete(tempReqFile); } }4. 集成到 AgentFramework 与 openClaw 技能4.1 封装为 AgentFramework 工具有了执行引擎接下来就是将其包装成 AgentFramework 能识别的工具。这里以微软 Semantic Kernel一个流行的 Agent 框架为例展示如何创建PythonScriptTool。using Microsoft.SemanticKernel; using Microsoft.SemanticKernel.Plugins.Core; namespace AgentFramework.PythonTools.Plugins { public class PythonScriptTool { private readonly IPythonExecutionEngine _engine; public PythonScriptTool(IPythonExecutionEngine engine) { _engine engine; } [KernelFunction, Description(Executes a piece of Python code and returns the output.)] public async Taskstring ExecutePythonAsync( [Description(The Python code to execute.)] string code, [Description(Optional session ID for stateful execution.)] string? sessionId null, [Description(List of pip requirements to install before execution, e.g., [requests, numpy].)] string? requirements null, CancellationToken cancellationToken default) { string[]? reqs null; if (!string.IsNullOrWhiteSpace(requirements)) { reqs requirements.Split(,, StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); } var result await _engine.ExecuteCodeAsync(code, sessionId, reqs, timeoutSeconds: 30, cancellationToken); if (!result.Success) { throw new KernelException($Python execution failed: {result.StandardError}); } // 返回标准输出和结果的组合信息 return $Execution succeeded in {result.ExecutionTime.TotalSeconds:F2}s.\nOutput:\n{result.StandardOutput}; } } }然后在初始化 Kernel 时注册这个插件using Microsoft.SemanticKernel; var kernel Kernel.CreateBuilder() .AddAzureOpenAIChatCompletion(deploymentName, endpoint, apiKey) .Build(); var pythonEngine new PythonExecutionEngine(C:\Python39\python.exe, D:\AgentTemp); var pythonTool new PythonScriptTool(pythonEngine); kernel.ImportPluginFromObject(pythonTool, python); // 现在智能体在规划任务时就可以使用 python.ExecutePythonAsync 这个函数了。4.2 构建 openClaw 技能服务openClaw 技能通常通过 HTTP API 对外提供服务。我们需要创建一个简单的 Web API可以用 ASP.NET Core Minimal API 快速实现来暴露我们的代码执行能力。// Program.cs using AgentFramework.PythonTools; var builder WebApplication.CreateBuilder(args); builder.Services.AddSingletonIPythonExecutionEngine(sp new PythonExecutionEngine( builder.Configuration[PythonPath], builder.Configuration[TempWorkspaceRoot], sp.GetRequiredServiceILoggerPythonExecutionEngine() )); var app builder.Build(); app.MapPost(/execute, async (PythonExecutionRequest request, IPythonExecutionEngine engine) { var result await engine.ExecuteCodeAsync( request.Code, request.SessionId, request.Requirements, request.TimeoutSeconds ?? 30 ); return Results.Json(new PythonExecutionResponse { Success result.Success, Output result.StandardOutput, Error result.StandardError, ExecutionTimeMs result.ExecutionTime.TotalMilliseconds, Result result.ResultPayload }); }); app.Run(); public record PythonExecutionRequest { public required string Code { get; set; } public string? SessionId { get; set; } public string[]? Requirements { get; set; } public int? TimeoutSeconds { get; set; } } public record PythonExecutionResponse { public bool Success { get; set; } public string? Output { get; set; } public string? Error { get; set; } public double ExecutionTimeMs { get; set; } public string? Result { get; set; } }将这个服务部署后你就得到了一个 openClaw 技能端点。在 openClaw 的技能配置中注册这个 HTTP 端点的 URL、描述和输入输出 Schema。当工作流需要执行 Python 代码时openClaw 就会将请求转发到这里。4.3 高级特性会话管理与上下文传递为了实现更复杂的多步交互我们需要增强会话管理。上面的代码已经通过sessionId和_sessionWorkspaces字典为每个会话保留了独立的工作目录。我们可以扩展这个机制在目录中保存一个context.pkl文件来持久化 Python 的全局状态。在安全壳脚本的末尾我们可以添加代码将globals()中所有可序列化的变量用pickle保存起来。在下一次执行时安全壳脚本开头先检查并加载这个上下文文件。这样上一次执行中定义的变量和函数在下一次执行中依然可用。# 在安全壳脚本中的补充代码 import pickle import os CONTEXT_FILE os.path.join(os.path.dirname(__file__), context.pkl) # 执行用户代码前加载上下文 if os.path.exists(CONTEXT_FILE): with open(CONTEXT_FILE, rb) as f: saved_globals pickle.load(f) globals().update(saved_globals) # ... [执行用户代码] ... # 执行用户代码后保存上下文排除安全壳自身的变量 vars_to_save {k: v for k, v in globals().items() if not k.startswith(__) and k not in [BLOCKED_MODULES, safe_import, ...]} with open(CONTEXT_FILE, wb) as f: pickle.dump(vars_to_save, f)在 .Net 引擎的CleanupSessionAsync方法中我们需要删除整个会话工作目录以清理所有临时文件和保存的上下文。5. 部署、调试与性能优化实战5.1 环境部署与配置要点将这套系统投入生产环境有几个关键配置点需要特别注意Python 解释器路径建议使用一个独立的 Conda 环境路径。在 Docker 容器中部署时可以将 Python 环境直接打包进镜像确保一致性。临时目录权限确保_tempDirectoryRoot指定的目录存在且运行服务的账户如www-data,NETWORK SERVICE有完全的读写权限。考虑使用 RAM Disk如/dev/shm来提升 I/O 性能尤其是对于频繁创建小文件的场景。资源限制除了代码中的超时设置务必在操作系统层面进行限制。在 Linux 上可以使用systemd的MemoryMax,CPUQuota来限制整个服务的内存和 CPU。对于每个 Python 子进程可以在启动前通过setrlimit(P/Invoke 调用) 来限制内存和 CPU 时间。日志与监控为PythonExecutionEngine添加详细的日志记录包括每次执行的代码片段可脱敏、会话 ID、执行时间、资源消耗等。这有助于问题排查和审计。5.2 常见问题排查与调试技巧在实际开发和使用中我踩过不少坑这里分享几个典型的排查思路问题一Python 进程启动失败报“系统找不到指定的文件”。排查首先检查_pythonExecutablePath路径是否正确。在 Linux 上可以使用which python3确认。其次检查运行服务的账户是否有该路径的执行权限。在 Docker 中确保 Python 已正确安装且在PATH中。技巧在引擎初始化时可以尝试运行python --version来验证路径有效性。问题二依赖安装超时或失败特别是从官方 PyPI 源下载慢。排查检查网络连通性。查看pip install的错误输出通常是网络超时或包不存在。技巧在构建基础 Docker 镜像时就预装常用包如numpy,pandas,requests。为pip配置国内镜像源如清华源、阿里云源。可以在InstallRequirementsAsync的命令参数中加入-i https://pypi.tuna.tsinghua.edu.cn/simple。问题三用户代码执行成功但 .Net 端捕获不到__result__。排查检查安全壳脚本中结果捕获的逻辑。用户代码是否真的将结果赋值给了__result__安全壳中的print输出是否被正确重定向和解析技巧在开发调试阶段可以暂时注释掉安全壳中重定向stdout的部分并让安全壳打印出locals()的内容看看执行后到底有哪些变量。在 .Net 端将outputBuilder的完整内容记录下来检查---AGENT_PYTHON_RESULT_START---标记是否存在且格式正确。问题四执行包含复杂循环或递归的代码时超时控制似乎不生效。排查Task.Delay产生的超时只会取消等待任务但不会强制终止已经启动的process.WaitForExit()任务。process.Kill()是异步的可能不会立即生效。技巧使用CancellationToken的Register方法注册一个回调在取消时立即调用process.Kill()。考虑使用Process.WaitForExitAsync(CancellationToken).NET 5来更好地整合取消信号。var cts CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); cts.CancelAfter(TimeSpan.FromSeconds(timeoutSeconds)); cts.Token.Register(() { try { if (!process.HasExited) process.Kill(true); } catch { } }); await process.WaitForExitAsync(cts.Token);问题五在并发请求下性能下降明显或临时目录文件冲突。排查检查是否为每个执行请求创建了独立的临时目录或会话目录。Guid.NewGuid()可以保证文件名唯一但目录管理不善会导致清理困难。技巧实现一个简单的“工作目录池”。预创建一批空目录执行时分配一个用完后标记为“待清理”由后台任务异步清理并放回池中避免频繁创建删除目录的开销。同时确保GetOrCreateSessionWorkspace方法是线程安全的。5.3 性能优化与安全加固建议环境预热对于公共基础环境在服务启动时可以预先执行一次import常用库如numpy避免第一次执行时的导入开销。会话复用对于已知的、会频繁进行多步交互的会话不要每次执行后立即清理工作目录。可以设置一个会话过期时间如闲置10分钟后清理。结果缓存如果相同的代码和参数被频繁执行可以考虑在 .Net 端增加一个内存缓存如使用IMemoryCache缓存执行结果。缓存键可以由代码哈希 会话ID 依赖列表构成。安全加固升级使用 Docker 容器隔离这是最推荐的生产方案。每个执行请求在一个新的、资源受限的 Docker 容器中运行。可以使用Docker.DotNet库来动态创建和启动容器。容器镜像基于一个最小化的 Python 环境通过卷挂载将代码和上下文文件传入。代码静态分析在将代码交给 Python 执行前先用Microsoft.CodeAnalysis.Scripting对于 .Net或python的ast模块在另一个预处理步骤中对代码进行简单的语法树分析禁止某些危险的语法节点如Import黑名单模块、Call到eval/exec、访问__开头的特殊属性等。网络访问控制在沙箱内使用防火墙规则或修改/etc/hosts文件来禁止所有网络访问除非任务明确需要这需要更复杂的白名单机制。将 AI 智能体与代码执行能力结合就像是赋予了它“手”和“眼”使其能从纯文本的思考者进化为能直接操作数字世界的行动者。这个过程充满了挑战尤其是在安全与能力之间寻找平衡。我分享的这个基于进程隔离和 Conda 环境的方案是一个在开发效率、运行性能和安全性之间取得不错平衡的起点。对于内部可信场景或原型验证它完全够用而对于面向不可信用户的生产环境务必向容器化隔离方案演进。在实现过程中最深的体会是约定大于配置。与 AI 智能体约定好如何传递代码、如何返回结果、如何管理状态能极大地简化系统复杂度。例如强制要求智能体将需要的结果赋值给__result__变量比我们去猜测代码的最后一行表达式要可靠得多。同时日志一定要打够从 AI 生成的代码到执行环境变量再到完整的输入输出这些日志在排查那些“AI 突然写了段匪夷所思的代码导致挂掉”的问题时是唯一的救命稻草。

关于恒美微站

恒美微站专注于为个体商户、工作室提供极简自助建站服务,让每个人都能轻松拥有专业网站。

快速链接

  • 关于我们
  • 建站服务
  • 主题模板
  • 案例展示
  • 资讯中心

服务项目

  • 可视化建站
  • 拖拽编辑
  • 主题定制
  • SEO 优化
  • 网站托管

联系方式

  • 📍 地址:北京市朝阳区建国路 88 号
  • 📞 电话:400-888-8888
  • ✉️ 邮箱:info@hmyw.cn
  • 🕐 时间:周一至周日 9:00-18:00

© 2024 恒美微站 hmyw.cn 版权所有 | 京 ICP 备 12345678 号